Class: Mammoth::Application

Inherits:
Object
  • Object
show all
Defined in:
lib/mammoth/application.rb

Overview

Top-level Mammoth application runtime.

Application wires Mammoth's delivery-side runtime pieces: configuration, SQLite operational memory, replication consumer, delivery worker, checkpoint store, dead letter store, and webhook sink. Upstream PostgreSQL transport composition stays outside this class so the application runtime consumes an injected CDC work source rather than owning upstream CDC source-adapter lifecycle decisions.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, source: nil, sink: nil, sleeper: Kernel.method(:sleep), lifecycle_hooks: LifecycleHooks.new) ⇒ Application

Returns a new instance of Application.

Parameters:

  • config (Mammoth::Configuration) —

    loaded configuration

  • source (#each, nil) (defaults to: nil) —

    injectable event source for tests and demos

  • sink (#deliver, nil) (defaults to: nil) —

    optional destination sink

  • sleeper (#call) (defaults to: Kernel.method(:sleep)) —

    retry sleep strategy

  • lifecycle_hooks (Mammoth::LifecycleHooks, Hash) (defaults to: LifecycleHooks.new) —

    local lifecycle callbacks



20
21
22
23
24
25
26
27
 
# File 'lib/mammoth/application.rb', line 20

def initialize(config, source: nil, sink: nil, sleeper: Kernel.method(:sleep), lifecycle_hooks: LifecycleHooks.new)
  @config = config
  @lifecycle_hooks = build_lifecycle_hooks(lifecycle_hooks)
  @state_adapter = build_state_adapter
  @checkpoint_store = state_adapter.checkpoint_store
  @consumer = ReplicationConsumer.new(source: source || build_source, delivery_unit: delivery_unit)
  @delivery_worker = sink ? build_delivery_worker(sink: sink, sleeper: sleeper) : build_configured_delivery_worker(sleeper:)
end

Instance Attribute Details

#checkpoint_store ⇒ Object (readonly)

Returns the value of attribute checkpoint_store.



13
14
15
 
# File 'lib/mammoth/application.rb', line 13

def checkpoint_store
  @checkpoint_store
end

#config ⇒ Object (readonly)

Returns the value of attribute config.



13
14
15
 
# File 'lib/mammoth/application.rb', line 13

def config
  @config
end

#consumer ⇒ Object (readonly)

Returns the value of attribute consumer.



13
14
15
 
# File 'lib/mammoth/application.rb', line 13

def consumer
  @consumer
end

#delivery_worker ⇒ Object (readonly)

Returns the value of attribute delivery_worker.



13
14
15
 
# File 'lib/mammoth/application.rb', line 13

def delivery_worker
  @delivery_worker
end

#lifecycle_hooks ⇒ Object (readonly)

Returns the value of attribute lifecycle_hooks.



13
14
15
 
# File 'lib/mammoth/application.rb', line 13

def lifecycle_hooks
  @lifecycle_hooks
end

#state_adapter ⇒ Object (readonly)

Returns the value of attribute state_adapter.



13
14
15
 
# File 'lib/mammoth/application.rb', line 13

def state_adapter
  @state_adapter
end

Instance Method Details

#sqlite_store ⇒ Mammoth::SQLiteStore

Returns underlying SQLite store for compatibility.

Returns:



30
31
32
 
# File 'lib/mammoth/application.rb', line 30

def sqlite_store
  state_adapter.respond_to?(:sqlite_store) ? state_adapter.sqlite_store : nil
end

#start ⇒ Integer

Start the application runtime and deliver consumed CDC work.

Returns:

  • (Integer) —

    number of processed work units



37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/mammoth/application.rb', line 37

def start
  runtime = build_runtime
  processed = nil

  lifecycle_hooks.call(:before_start, application_context(runtime: runtime))
  processed = process_consumer(runtime)
  lifecycle_hooks.call(:after_start, application_context(runtime: runtime, processed: processed))
  processed
ensure
  lifecycle_hooks.call(:before_shutdown, application_context(runtime: runtime, processed: processed))
  runtime.shutdown if runtime.respond_to?(:shutdown)
  lifecycle_hooks.call(:after_shutdown, application_context(runtime: runtime, processed: processed))
end