Class: Textus::Surface::Watcher

Inherits:
Object
  • Object
show all
Defined in:
lib/textus/surface/watcher.rb

Instance Method Summary collapse

Constructor Details

#initialize(container:) ⇒ Watcher

Returns a new instance of Watcher.



6
7
8
9
 
# File 'lib/textus/surface/watcher.rb', line 6

def initialize(container:)
  @container = container
  @registry = Textus::Workflow::Registry.new(container.workflows)
end

Instance Method Details

#run(poll: nil) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/textus/surface/watcher.rb', line 29

def run(poll: nil)
  interval = poll || @container.manifest.data.worker_config[:poll] || 2
  lock = Infra::Locks::WatcherLock.new(@container.root)
  lock.acquire
  begin
    loop do
      tick
      sleep(interval)
    end
  ensure
    lock.release
  end
end

#tick ⇒ Object 



11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/textus/surface/watcher.rb', line 11

def tick
  call = Textus::Value::Call.build(role: Textus::Value::Role::AUTOMATION, correlation_id: SecureRandom.uuid)

  Textus::Workflow::Scheduler.new(
    registry: @registry,
    workflow_queue: @container.store.workflow_queue,
    reader: ->(key:) { @container.store_engine.read(key:) },
  ).seed_expired

  consumer = Textus::Workflow::Consumer.new(registry: @registry, container: @container, call:)
  loop do
    row = @container.store.workflow_queue.pop
    break unless row

    consumer.consume(row, @container.store.workflow_queue)
  end
end