Module: WaterDrop::Producer::Buffer
- Included in:
- WaterDrop::Producer
- Defined in:
- lib/waterdrop/producer/buffer.rb
Overview
Component for buffered operations
Instance Method Summary collapse
-
#buffer(message) ⇒ Object
Adds given message into the internal producer buffer without flushing it to Kafka.
-
#buffer_many(messages) ⇒ Object
Adds given messages into the internal producer buffer without flushing them to Kafka.
-
#flush_async ⇒ Array<Rdkafka::Producer::DeliveryHandle>
Flushes the internal buffer to Kafka in an async way.
-
#flush_sync ⇒ Array<Rdkafka::Producer::DeliveryReport>
Flushes the internal buffer to Kafka in a sync way.
Instance Method Details
#buffer(message) ⇒ Object
Adds given message into the internal producer buffer without flushing it to Kafka
12 13 14 15 16 17 18 19 20 21 22 23 24 |
# File 'lib/waterdrop/producer/buffer.rb', line 12 def buffer() ensure_active! = middleware.run() () @monitor.instrument( 'message.buffered', producer_id: id, message: , buffer: @messages ) { @messages << } end |
#buffer_many(messages) ⇒ Object
Adds given messages into the internal producer buffer without flushing them to Kafka
32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 |
# File 'lib/waterdrop/producer/buffer.rb', line 32 def buffer_many() ensure_active! = middleware.run_many() .each { || () } @monitor.instrument( 'messages.buffered', producer_id: id, messages: , buffer: @messages ) do .each { || @messages << } end end |
#flush_async ⇒ Array<Rdkafka::Producer::DeliveryHandle>
Flushes the internal buffer to Kafka in an async way
52 53 54 55 56 57 58 |
# File 'lib/waterdrop/producer/buffer.rb', line 52 def flush_async @monitor.instrument( 'buffer.flushed_async', producer_id: id, messages: @messages ) { flush(false) } end |
#flush_sync ⇒ Array<Rdkafka::Producer::DeliveryReport>
Flushes the internal buffer to Kafka in a sync way
63 64 65 66 67 68 69 |
# File 'lib/waterdrop/producer/buffer.rb', line 63 def flush_sync @monitor.instrument( 'buffer.flushed_sync', producer_id: id, messages: @messages ) { flush(true) } end |