Module: WaterDrop::Producer::Sync
- Included in:
- WaterDrop::Producer
- Defined in:
- lib/waterdrop/producer/sync.rb
Overview
Component for synchronous producer operations
Instance Method Summary collapse
-
#produce_many_sync(messages) ⇒ Array<Rdkafka::Producer::DeliveryReport>
Produces many messages to Kafka and waits for them to be delivered.
-
#produce_sync(message) ⇒ Rdkafka::Producer::DeliveryReport
Produces a message to Kafka and waits for it to be delivered.
Instance Method Details
#produce_many_sync(messages) ⇒ Array<Rdkafka::Producer::DeliveryReport>
Produces many messages to Kafka and waits for them to be delivered
59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 |
# File 'lib/waterdrop/producer/sync.rb', line 59 def produce_many_sync() = middleware.run_many() .each { || () } dispatched = [] @monitor.instrument('messages.produced_sync', producer_id: id, messages: ) do with_transaction_if_transactional do .each do || dispatched << produce() end end dispatched.map! do |handler| wait(handler) end dispatched end rescue *SUPPORTED_FLOW_ERRORS => e re_raised = Errors::ProduceManyError.new(dispatched, e.inspect) @monitor.instrument( 'error.occurred', producer_id: id, messages: , dispatched: dispatched, error: re_raised, type: 'messages.produce_many_sync' ) raise re_raised end |
#produce_sync(message) ⇒ Rdkafka::Producer::DeliveryReport
Produces a message to Kafka and waits for it to be delivered
18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
# File 'lib/waterdrop/producer/sync.rb', line 18 def produce_sync() = middleware.run() () @monitor.instrument( 'message.produced_sync', producer_id: id, message: ) do wait(produce()) end rescue *SUPPORTED_FLOW_ERRORS => e # We use this syntax here because we want to preserve the original `#cause` when we # instrument the error and there is no way to manually assign `#cause` value begin raise Errors::ProduceError, e.inspect rescue Errors::ProduceError => ex @monitor.instrument( 'error.occurred', producer_id: id, message: , error: ex, type: 'message.produce_sync' ) raise ex end end |