Class: Racecar::AsyncPartitionProcessor

Inherits:
Object
  • Object
show all
Defined in:
lib/racecar/async_partition_processor.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic:, partition:, logger:, config:, consumer:, consumer_class:, instrumenter:, rdkafka_consumer:) ⇒ AsyncPartitionProcessor

Returns a new instance of AsyncPartitionProcessor.



14
15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/racecar/async_partition_processor.rb', line 14

def initialize(topic:, partition:, logger:, config:, consumer:, consumer_class:, instrumenter:, rdkafka_consumer:)
  @topic      = topic
  @partition  = partition
  @logger     = logger
  @config     = config
  @consumer   = consumer
  @consumer_class = consumer_class
  @instrumenter = instrumenter
  @rdkafka_consumer = rdkafka_consumer
  @backpressure_paused = Concurrent::AtomicBoolean.new
  @tpl = build_tpl(topic, partition)
  setup_async_processing
end

Instance Attribute Details

#thread ⇒ Object (readonly)

Returns the value of attribute thread.



8
9
10
 
# File 'lib/racecar/async_partition_processor.rb', line 8

def thread
  @thread
end

Class Method Details

.thread_key(topic, partition) ⇒ Object 



10
11
12
 
# File 'lib/racecar/async_partition_processor.rb', line 10

def self.thread_key(topic, partition)
  "#{topic}/#{partition}"
end

Instance Method Details

#process(message) ⇒ Object 



28
29
30
 
# File 'lib/racecar/async_partition_processor.rb', line 28

def process(message)
  push(message)
end

#process_batch(messages) ⇒ Object 



32
33
34
 
# File 'lib/racecar/async_partition_processor.rb', line 32

def process_batch(messages)
  push(messages)
end

#rebalance! ⇒ Object 



36
37
38
39
 
# File 'lib/racecar/async_partition_processor.rb', line 36

def rebalance!
  processor.rebalance!
  @queue << nil
end

#rebalancing_or_shutting_down? ⇒ Boolean

Returns:

  • (Boolean) 


46
47
48
 
# File 'lib/racecar/async_partition_processor.rb', line 46

def rebalancing_or_shutting_down?
  processor.rebalancing_or_shutting_down?
end

#resume_paused_partition ⇒ Object 



50
51
52
 
# File 'lib/racecar/async_partition_processor.rb', line 50

def resume_paused_partition
  processor.resume_paused_partition
end

#shut_down! ⇒ Object 



41
42
43
44
 
# File 'lib/racecar/async_partition_processor.rb', line 41

def shut_down!
  processor.shut_down!
  @queue << nil
end