Class: Rdkafka::Callbacks::DeliveryCallback

Inherits:

Object

• Object
• Rdkafka::Callbacks::DeliveryCallback

Defined in:

lib/rdkafka/callbacks.rb

Class Method Summary

• .call(_client_ptr, message_ptr, opaque_ptr) ⇒ Object

  Handles message delivery callbacks.

Class Method Details

.call(_client_ptr, message_ptr, opaque_ptr) ⇒ Object

Handles message delivery callbacks

Parameters:

• _client_ptr (FFI::Pointer) —

  unused client pointer

• message_ptr (FFI::Pointer) —

  pointer to the delivered message

• opaque_ptr (FFI::Pointer) —

  pointer to the opaque object for callback context

# File 'lib/rdkafka/callbacks.rb', line 246

def self.call(_client_ptr, message_ptr, opaque_ptr)
  message = Rdkafka::Bindings::Message.new(message_ptr)
  delivery_handle_ptr_address = message[:_private].address
  if delivery_handle = Rdkafka::Producer::DeliveryHandle.remove(delivery_handle_ptr_address)
    topic_name = Rdkafka::Bindings.rd_kafka_topic_name(message[:rkt])

    # Update delivery handle
    delivery_handle[:response] = message[:err]
    delivery_handle[:partition] = message[:partition]
    delivery_handle[:offset] = message[:offset]
    # The topic is not stored on the handle here: it is already set as the handle's
    # `topic` Ruby attribute during produce, which spares a native string copy that
    # would otherwise be allocated and retained for every message.

    begin
      # Call delivery callback on opaque
      if opaque = Rdkafka::Config.opaques[opaque_ptr.to_i]
        opaque.call_delivery_callback(
          Rdkafka::Producer::DeliveryReport.new(
            message[:partition],
            message[:offset],
            topic_name,
            message[:err],
            delivery_handle.label
          ),
          delivery_handle
        )
      end
    rescue Exception => err
      # A user delivery callback that raises must not escape this FFI callback: it runs on
      # librdkafka's polling thread (abort_on_exception = true), so an escaping exception
      # would crash the whole process. We log and swallow it, matching the rebalance
      # callback. The `ensure` below still unlocks the handle, so `wait` returns the
      # delivery report instead of blocking until its timeout and raising
      # WaitTimeoutError for a message that was in fact delivered.
      Rdkafka::Config.logger.error("Unhandled exception: #{err.class} - #{err.message}")
    ensure
      delivery_handle.unlock
    end
  end
end