logstash-core/lib/logstash/pipeline.rb
def worker_loop(batch_size, batch_delay)
    shutdown_requested = false

    @filter_queue_client.set_batch_dimensions(batch_size, batch_delay)

    while true
      signal = @signal_queue.empty? ? NO_SIGNAL : @signal_queue.pop
      shutdown_requested |= signal.shutdown? # latch on shutdown signal

      batch = @filter_queue_client.read_batch # metrics are started in read_batch
      @events_consumed.increment(batch.size)
      filter_batch(batch)
      flush_filters_to_batch(batch, :final => false) if signal.flush?
      output_batch(batch)
      break if @force_shutdown.true? # Do not ack the current batch
      @filter_queue_client.close_batch(batch)

      # keep break at end of loop, after the read_batch operation, some pipeline specs rely on this "final read_batch" before shutdown.
      break if shutdown_requested && !draining_queue?
    end

    # we are shutting down, queue is drained if it was required, now  perform a final flush.
    # for this we need to create a new empty batch to contain the final flushed events
    batch = @filter_queue_client.new_batch
    @filter_queue_client.start_metrics(batch) # explicitly call start_metrics since we dont do a read_batch here
    flush_filters_to_batch(batch, :final => true)
    return if @force_shutdown.true? # Do not ack the current batch
    output_batch(batch)
    @filter_queue_client.close_batch(batch)
  end
# Main body of what a worker thread does
# Repeatedly takes batches off the queue, filters, then outputs them
Main body of what a worker thread does
Repeatedly takes batches off the queue, filters, then outputs them
Main body of what a worker thread does Repeatedly takes batches off the queue filters then outputs them
def worker_loop(batch_size, batch_delay)
    shutdown_requested = false
    @filter_queue_client.set_batch_dimensions(batch_size, batch_delay)
    while true
      signal = @signal_queue.empty? ? NO_SIGNAL : @signal_queue.pop
      shutdown_requested |= signal.shutdown? 
      batch = @filter_queue_client.read_batch 
      @events_consumed.increment(batch.size)
      filter_batch(batch)
      flush_filters_to_batch(batch, :final => false) if signal.flush?
      output_batch(batch)
      break if @force_shutdown.true? 
      @filter_queue_client.close_batch(batch)
      break if shutdown_requested && !draining_queue?
    end
    batch = @filter_queue_client.new_batch
    @filter_queue_client.start_metrics(batch) 
    flush_filters_to_batch(batch, :final => true)
    return if @force_shutdown.true? 
    output_batch(batch)
    @filter_queue_client.close_batch(batch)
  end
def worker_loop ( batch_size , batch_delay ) shutdown_requested = false @filter_queue_client . set_batch_dimensions ( batch_size , batch_delay ) while true signal = @signal_queue . empty? ? NO_SIGNAL : @signal_queue . pop shutdown_requested |= signal . shutdown? batch = @filter_queue_client . read_batch @events_consumed . increment ( batch . size ) filter_batch ( batch ) flush_filters_to_batch ( batch , :final => false ) if signal . flush? output_batch ( batch ) break if @force_shutdown . true? @filter_queue_client . close_batch ( batch ) break if shutdown_requested && ! draining_queue? end batch = @filter_queue_client . new_batch @filter_queue_client . start_metrics ( batch ) flush_filters_to_batch ( batch , :final => true ) return if @force_shutdown . true? output_batch ( batch ) @filter_queue_client . close_batch ( batch ) end
Main body of what a worker thread does
Repeatedly takes batches off the queue, filters, then outputs them
Main body of what a worker thread does Repeatedly takes batches off the queue filters then outputs them
# latch on shutdown signal # metrics are started in read_batch # Do not ack the current batch # keep break at end of loop, after the read_batch operation, some pipeline specs rely on this "final read_batch" before shutdown. # we are shutting down, queue is drained if it was required, now  perform a final flush. # for this we need to create a new empty batch to contain the final flushed events # explicitly call start_metrics since we dont do a read_batch here # Do not ack the current batch
batch_size
batch_delay