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 source topic key and not the partition id. This will not utilize partitions beyond the number of partitions of source 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 source headers + meta headers about the source of message.

Parameters:



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

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>) —

    source messages to pipe



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

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>) —

    source messages to pipe



90
91
92
93
94
95
96
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 90

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]


62
63
64
65
66
 
# File 'lib/karafka/pro/processing/piping/consumer.rb', line 62

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