Module: Karafka::Pro::Processing::Piping::Consumer

Defined in:
lib/karafka/pro/processing/piping/consumer.rb

Overview

Consumer piping functionality

It provides way to pipe data in a consistent way with extra traceability headers similar to those in the enhanced DLQ.

Instance Method Summary collapse

Instance Method Details

#pipe_async(topic:, message:) ⇒ Object

Note:

It will NOT deserialize the payload so it is fast

Note:

We assume that there can be different number of partitions in the target topic, this is why we use key based on the original topic key and not the partition id. This will not utilize partitions beyond the number of partitions of original topic, but will accommodate for topics with less partitions.

Pipes given message to the provided topic with expected details. Useful for pass-through operations where deserialization is not needed. Upon usage it will include all the original headers + meta headers about the source of message.

Parameters:



42
43
44
45
46
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 42

def pipe_async(topic:, message:)
  produce_async(
    build_pipe_message(topic: topic, message: message)
  )
end

#pipe_many_async(topic:, messages:) ⇒ Object

Note:

If transactional producer in use and dispatch is not wrapped with a transaction, it will automatically wrap the dispatch with a transaction

Async multi-message pipe

Parameters:

  • topic (String, Symbol) —

    where we want to send the message

  • messages (Array<Karafka::Messages::Message>) —

    original messages to pipe



66
67
68
69
70
71
72
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 66

def pipe_many_async(topic:, messages:)
  messages = messages.map do |message|
    build_pipe_message(topic: topic, message: message)
  end

  produce_many_async(messages)
end

#pipe_many_sync(topic:, messages:) ⇒ Object

Note:

If transactional producer in use and dispatch is not wrapped with a transaction, it will automatically wrap the dispatch with a transaction

Sync multi-message pipe

Parameters:

  • topic (String, Symbol) —

    where we want to send the message

  • messages (Array<Karafka::Messages::Message>) —

    original messages to pipe



81
82
83
84
85
86
87
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 81

def pipe_many_sync(topic:, messages:)
  messages = messages.map do |message|
    build_pipe_message(topic: topic, message: message)
  end

  produce_many_sync(messages)
end

#pipe_sync(topic:, message:) ⇒ Object

Sync version of pipe for one message

Parameters:

See Also:

  • Karafka::Pro::Processing::Piping::Consumer.[[#pipe_async]


53
54
55
56
57
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 53

def pipe_sync(topic:, message:)
  produce_sync(
    build_pipe_message(topic: topic, message: message)
  )
end