Module: Rdkafka::Callbacks

Defined in:
lib/rdkafka/callbacks.rb,
lib/rdkafka/callbacks/base_handler.rb,
lib/rdkafka/callbacks/create_acl_handler.rb,
lib/rdkafka/callbacks/delete_acl_handler.rb,
lib/rdkafka/callbacks/create_topic_handler.rb,
lib/rdkafka/callbacks/delete_topic_handler.rb,
lib/rdkafka/callbacks/describe_acl_handler.rb,
lib/rdkafka/callbacks/list_offsets_handler.rb,
lib/rdkafka/callbacks/delete_groups_handler.rb,
lib/rdkafka/callbacks/describe_configs_handler.rb,
lib/rdkafka/callbacks/create_partitions_handler.rb,
lib/rdkafka/callbacks/incremental_alter_configs_handler.rb

Overview

Callback handlers for librdkafka events

Defined Under Namespace

Classes: BackgroundEventCallback, BaseHandler, CreateAclHandler, CreateAclResult, CreatePartitionsHandler, CreateTopicHandler, DeleteAclHandler, DeleteAclResult, DeleteGroupsHandler, DeleteTopicHandler, DeliveryCallback, DescribeAclHandler, DescribeAclResult, DescribeConfigsHandler, DescribeConfigsResult, GroupResult, IncrementalAlterConfigsHandler, IncrementalAlterConfigsResult, ListOffsetsHandler, ListOffsetsResult, TopicResult

Constant Summary collapse

@@mutex = 
Mutex.new
@@current_pid = 
nil

Class Method Summary collapse

Class Method Details

.ensure_ffi_running ⇒ Object

Defines or recreates after fork callbacks that require FFI thread so the callback thread is always correctly initialized

See Also:



299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
 
# File 'lib/rdkafka/callbacks.rb', line 299

def ensure_ffi_running
  @@mutex.synchronize do
    return if @@current_pid == ::Process.pid

    if const_defined?(:BackgroundEventCallbackFunction, false)
      send(:remove_const, :BackgroundEventCallbackFunction)
      send(:remove_const, :DeliveryCallbackFunction)
    end

    # FFI Function used for Create Topic and Delete Topic callbacks
    background_event_callback_function = FFI::Function.new(
      :void, [:pointer, :pointer, :pointer]
    ) do |client_ptr, event_ptr, opaque_ptr|
      BackgroundEventCallback.call(client_ptr, event_ptr, opaque_ptr)
    end

    # FFI Function used for Message Delivery callbacks
    delivery_callback_function = FFI::Function.new(
      :void, [:pointer, :pointer, :pointer]
    ) do |client_ptr, message_ptr, opaque_ptr|
      DeliveryCallback.call(client_ptr, message_ptr, opaque_ptr)
    end

    const_set(:BackgroundEventCallbackFunction, background_event_callback_function)
    const_set(:DeliveryCallbackFunction, delivery_callback_function)

    @@current_pid = ::Process.pid
  end
end