Class: Kafka::AsyncProducer::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/async_producer.rb

Instance Method Summary collapse

Constructor Details

#initialize(queue:, producer:, delivery_threshold:) ⇒ Worker

Returns a new instance of Worker.



137
138
139
140
141
 
# File 'lib/kafka/async_producer.rb', line 137

def initialize(queue:, producer:, delivery_threshold:)
  @queue = queue
  @producer = producer
  @delivery_threshold = delivery_threshold
end

Instance Method Details

#run ⇒ Object 



143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/kafka/async_producer.rb', line 143

def run
  loop do
    operation, payload = @queue.pop

    case operation
    when :produce
      @producer.produce(*payload)
      deliver_messages if threshold_reached?
    when :deliver_messages
      deliver_messages
    when :shutdown
      # Deliver any pending messages first.
      deliver_messages

      # Stop the run loop.
      break
    else
      raise "Unknown operation #{operation.inspect}"
    end
  end
ensure
  @producer.shutdown
end