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:



34
35
36
37
38
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 34

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



58
59
60
61
62
63
64
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 58

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



73
74
75
76
77
78
79
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 73

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]


45
46
47
48
49
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 45

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