Class: TimeBucketStream::Writer

Inherits:
Object
  • Object
show all
Defined in:
lib/time_bucket_stream/writer.rb

Constant Summary collapse

BUCKET_FORMAT = 
"%Y%m%d%H%M"
SYNC_MODES = 
%i[none flush fsync].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path:, sync: :flush, clock: Time, codec: Codecs::Json.new) ⇒ Writer

Returns a new instance of Writer.



14
15
16
17
18
19
20
21
22
23
 
# File 'lib/time_bucket_stream/writer.rb', line 14

def initialize(path:, sync: :flush, clock: Time, codec: Codecs::Json.new)
  @paths = Paths.new(path: path)
  @sync = sync.respond_to?(:to_sym) ? sync.to_sym : sync
  @clock = clock
  @codec = Codecs.validate!(codec)
  @mutex = Mutex.new

  validate_sync!
  ensure_directories
end

Instance Attribute Details

#codec ⇒ Object (readonly)

Returns the value of attribute codec.



12
13
14
 
# File 'lib/time_bucket_stream/writer.rb', line 12

def codec
  @codec
end

#paths ⇒ Object (readonly)

Returns the value of attribute paths.



12
13
14
 
# File 'lib/time_bucket_stream/writer.rb', line 12

def paths
  @paths
end

#sync ⇒ Object (readonly)

Returns the value of attribute sync.



12
13
14
 
# File 'lib/time_bucket_stream/writer.rb', line 12

def sync
  @sync
end

Instance Method Details

#append(payload) ⇒ Object 



25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/time_bucket_stream/writer.rb', line 25

def append(payload)
  @mutex.synchronize do
    prepare_writer
    @sequence += 1

    entry = {
      "id" => @sequence,
      "payload" => payload
    }

    write_line(encode_entry(entry))
    entry.fetch("id")
  end
end

#close ⇒ Object 



40
41
42
43
44
 
# File 'lib/time_bucket_stream/writer.rb', line 40

def close
  @mutex.synchronize do
    close_writer
  end
end

#close_stale ⇒ Object 



46
47
48
49
50
51
52
 
# File 'lib/time_bucket_stream/writer.rb', line 46

def close_stale
  bucket = current_bucket

  @mutex.synchronize do
    close_writer if @writer_bucket && @writer_bucket < bucket
  end
end