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



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

def initialize(config, source: nil, sink: nil, sleeper: Kernel.method(:sleep))
  @config = config
  @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

#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:



28
29
30
 
# File 'lib/mammoth/application.rb', line 28

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



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/mammoth/application.rb', line 35

def start
  runtime = build_runtime
  processed = 0
  batch = [nil].compact

  consumer.start do |work|
    if runtime_batching?(runtime)
      batch << work
      next unless batch.size >= runtime_batch_size

      processed += process_batch(runtime, batch)
      batch = []
    else
      process_work(runtime, work)
      processed += 1
    end
  end

  processed += process_batch(runtime, batch) if runtime_batching?(runtime) && batch.any?
  processed
ensure
  runtime.shutdown if runtime.respond_to?(:shutdown)
end