Class: AgUi::RunStore::InMemory

Inherits:
Object
  • Object
show all
Defined in:
lib/ag_ui/run_store.rb

Overview

Single-process store. Fiber/thread-safe: state mutations are locked, and live fanout uses Async::Queue so connect streams receive events as they are recorded.

Constant Summary collapse

FINISHED = 
:finished

Instance Method Summary collapse

Constructor Details

#initialize ⇒ InMemory

Returns a new instance of InMemory.



24
25
26
27
28
29
 
# File 'lib/ag_ui/run_store.rb', line 24

def initialize
  @threads = Hash.new do |hash, key|
    hash[key] = { historic: [], live: nil, task: nil, subscribers: [] }
  end
  @mutex = Mutex.new
end

Instance Method Details

#attach_task(thread_id, task) ⇒ Object

Only meaningful while the run is live — a late attach (the builder calls on_task after Async returns, which for an already-completed run is after finish_run) must not resurrect a stopped handle.



68
69
70
71
72
73
74
 
# File 'lib/ag_ui/run_store.rb', line 68

def attach_task(thread_id, task)
  @mutex.synchronize do
    if @threads[thread_id][:live]
      @threads[thread_id][:task] = task
    end
  end
end

#begin_run(thread_id, run_id) ⇒ Object 



31
32
33
34
35
 
# File 'lib/ag_ui/run_store.rb', line 31

def begin_run(thread_id, run_id)
  @mutex.synchronize do
    @threads[thread_id][:live] = { run_id: run_id, events: [] }
  end
end

#finish_run(thread_id) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/ag_ui/run_store.rb', line 50

def finish_run(thread_id)
  subscribers = @mutex.synchronize do
    thread = @threads[thread_id]
    if thread[:live]
      thread[:historic] << thread[:live]
      thread[:live] = nil
    end
    thread[:task] = nil
    drained = thread[:subscribers]
    thread[:subscribers] = []
    drained
  end
  subscribers.each { |queue| queue.enqueue(FINISHED) }
end

#open_subscription(thread_id) ⇒ Object

Atomic snapshot + live subscription: the returned events are everything recorded so far (historic runs + the live run's events), and when a run is in flight, queue receives each subsequent event and finally FINISHED. Atomicity means no gap and no duplicates between snapshot and subscription.



94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/ag_ui/run_store.rb', line 94

def open_subscription(thread_id)
  @mutex.synchronize do
    thread = @threads[thread_id]
    events = thread[:historic].flat_map { |run| run[:events] }
    queue = nil
    if thread[:live]
      events += thread[:live][:events]
      queue = Async::Queue.new
      thread[:subscribers] << queue
    end
    { events: events, queue: queue }
  end
end

#record(thread_id, payload) ⇒ Object 



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

def record(thread_id, payload)
  subscribers = @mutex.synchronize do
    thread = @threads[thread_id]
    if thread[:live]
      thread[:live][:events] << payload
      thread[:subscribers].dup
    else
      []
    end
  end
  subscribers.each { |queue| queue.enqueue(payload) }
end

#stop(thread_id) ⇒ Object

Cancel the thread's in-flight run. The stream still closes cleanly (StreamBuilder's ensure) — matching the Node runtime, which ends an aborted run with a plain RUN_FINISHED-then-close.



79
80
81
82
83
84
85
86
87
 
# File 'lib/ag_ui/run_store.rb', line 79

def stop(thread_id)
  task = @mutex.synchronize { @threads[thread_id][:task] }
  if task
    task.stop
    true
  else
    false
  end
end