Class: Batchwatch::Spool

Inherits:
Object
  • Object
show all
Defined in:
lib/batchwatch/spool.rb

Overview

Append-only JSONL file of completed measurements waiting to be sent.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path, max_bytes: MAX_BYTES, logger: nil) ⇒ Spool

Returns a new instance of Spool.



39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/batchwatch/spool.rb', line 39

def initialize(path, max_bytes: MAX_BYTES, logger: nil)
  @path = path
  @pending = "#{path}.pending"
  @max_bytes = max_bytes
  @logger = logger
  # The measurements come from background threads, one per completed call.
  # Without the lock, two concurrent completions lose one of themselves:
  # "find the end, write" is not atomic. Seen in a test with three at once:
  # two lines made it.
  @lock = Mutex.new
end

Instance Attribute Details

#max_bytes ⇒ Object (readonly)

Returns the value of attribute max_bytes.



37
38
39
 
# File 'lib/batchwatch/spool.rb', line 37

def max_bytes
  @max_bytes
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



37
38
39
 
# File 'lib/batchwatch/spool.rb', line 37

def path
  @path
end

#pending ⇒ Object (readonly)

Returns the value of attribute pending.



37
38
39
 
# File 'lib/batchwatch/spool.rb', line 37

def pending
  @pending
end

Instance Method Details

#append(record) ⇒ Object

Store one completed measurement. Returns true if it was stored.

Never raises: failing a spool must not be worse than the network error that triggered the spool.



57
58
59
60
61
62
 
# File 'lib/batchwatch/spool.rb', line 57

def append(record)
  @lock.synchronize { write_one(record) }
rescue StandardError => e
  debug("batchwatch: could not spool: #{e}")
  false
end

#keep(remaining) ⇒ Object

Put records back after a partial or failed drain.



78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/batchwatch/spool.rb', line 78

def keep(remaining)
  @lock.synchronize do
    if remaining && !remaining.empty?
      write_all(@pending, remaining)
    elsif File.exist?(@pending)
      File.delete(@pending)
    end
  end
rescue StandardError => e
  debug("batchwatch: could not write spool back: #{e}")
end

#size ⇒ Object

Number of records waiting on disk. Best effort, never raises.



91
92
93
94
95
96
 
# File 'lib/batchwatch/spool.rb', line 91

def size
  @lock.synchronize { read_all(@pending).length + read_all(@path).length }
rescue StandardError => e
  debug("batchwatch: could not count spool: #{e}")
  0
end

#take ⇒ Object

Move everything spooled into the pending file and return it.

Returns a list of records. An empty list means there is nothing to send - also when the spool simply could not be read.



70
71
72
73
74
75
 
# File 'lib/batchwatch/spool.rb', line 70

def take
  @lock.synchronize { take_all }
rescue StandardError => e
  debug("batchwatch: could not read spool: #{e}")
  []
end