Module: Julewire::Karafka

Extended by:

Core::Integration::Configurable

Defined in:

lib/julewire/karafka.rb,
lib/julewire/karafka/version.rb,
lib/julewire/karafka/installer.rb,
lib/julewire/karafka/fork_hooks.rb,
lib/julewire/karafka/configuration.rb,
lib/julewire/karafka/event_payload.rb,
lib/julewire/karafka/event_severity.rb,
lib/julewire/karafka/payload_reader.rb,
lib/julewire/karafka/message_context.rb,
lib/julewire/karafka/monitor_listener.rb,
lib/julewire/karafka/message_execution.rb,
lib/julewire/karafka/waterdrop_installer.rb,
lib/julewire/karafka/messaging_attributes.rb,
lib/julewire/karafka/monitor_subscription.rb,
lib/julewire/karafka/waterdrop_middleware.rb

Defined Under Namespace

Modules: EventPayload, EventSeverity, ForkHooks, Installer, MessageContext, MessageExecution, MessagingAttributes, PayloadReader, WaterdropInstaller Classes: Configuration, Error, InstallResult, MonitorListener, WaterdropMiddleware

Constant Summary

IntegrationHealth =

Core::Integration::Health.scoped(:karafka)

VERSION =

"1.0.0"

Class Method Summary

• .inject!(message, configuration: config) ⇒ Object
• .install!(app: nil, monitor: nil, producer: nil, consumer: true, configuration: config) ⇒ Object
• .with_message(message, configuration: config) ⇒ Object
• .with_message_execution(message, configuration: config, **execution_options) ⇒ Object

Class Method Details

.inject!(message, configuration: config) ⇒ Object

# File 'lib/julewire/karafka.rb', line 29

def inject!(message, configuration: config)
  WaterdropMiddleware.new(configuration: configuration).call(message)
end

.install!(app: nil, monitor: nil, producer: nil, consumer: true, configuration: config) ⇒ Object

# File 'lib/julewire/karafka.rb', line 18

def install!(app: nil, monitor: nil, producer: nil, consumer: true, configuration: config)
  return false unless configuration.enabled?

  consumer_result = Installer.install!(app: app, monitor: monitor, configuration: configuration) if consumer
  producer_result = WaterdropInstaller.install!(producer, configuration: configuration) if producer

  return InstallResult.new(consumer_result, producer_result) if consumer && producer

  producer ? producer_result : consumer_result
end

.with_message(message, configuration: config) ⇒ Object

# File 'lib/julewire/karafka.rb', line 33

def with_message(message, configuration: config, &)
  MessageContext.call(message, configuration: configuration, &)
end

.with_message_execution(message, configuration: config, **execution_options) ⇒ Object

# File 'lib/julewire/karafka.rb', line 37

def with_message_execution(message, configuration: config, **execution_options, &)
  MessageExecution.call(message, configuration: configuration, **execution_options, &)
end