Class: Snerdmq::SnerdQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/snerdmq/queue.rb

Instance Method Summary collapse

Constructor Details

#initialize(binary_path: nil, storage_path: nil) ⇒ SnerdQueue

Returns a new instance of SnerdQueue.



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/snerdmq/queue.rb', line 6

def initialize(binary_path: nil, storage_path: nil)
  @binary_path = binary_path
  @storage_path = storage_path
  
  if @binary_path.nil?
    ext = RbConfig::CONFIG['host_os'].match?(/mswin|msys|mingw|cygwin|bccwin|wince|emc/) ? '.exe' : ''
    # Assume the binary was downloaded into the gem's bin/ directory via snerdmq-install
    @binary_path = File.expand_path("../../bin/snerdmq#{ext}", __dir__)
  end

  unless File.exist?(@binary_path)
    raise "[Snerd] Binary not found at #{@binary_path}. Ensure it is compiled or run 'snerdmq-install'."
  end

  @handlers = {}
  @handlers_mutex = Mutex.new
  
  @stdin_mutex = Mutex.new
  @shutting_down = false
  @io = nil
  @listener_thread = nil
  @pending_enqueues = {}
  @pending_mutex = Mutex.new
  @ws_clients = []
  @ws_mutex = Mutex.new
end

Instance Method Details

#enqueue(task_id:, task_type:, data:, max_retries: 3, retry_after_hours: 0.0, rate_limit_group: nil, max_per_minute: nil, auto_dedupe: false, urgency_score: nil) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/snerdmq/queue.rb', line 68

def enqueue(task_id:, task_type:, data:, max_retries: 3, retry_after_hours: 0.0, rate_limit_group: nil, max_per_minute: nil, auto_dedupe: false, urgency_score: nil)
  raise "[Snerd] Cannot enqueue task: Queue is not running. Call start_listening first." if @io.nil? || @shutting_down
  
  payload = {
    action: "enqueue",
    task_id: task_id,
    task_type: task_type,
    task_data: data.to_json,
    max_retries: max_retries,
    retry_after_hours: retry_after_hours
  }

  payload[:rate_limit_group] = rate_limit_group if rate_limit_group
  payload[:max_per_minute] = max_per_minute if max_per_minute
  payload[:auto_dedupe] = auto_dedupe if auto_dedupe
  payload[:urgency_score] = urgency_score if urgency_score

  cond = ConditionVariable.new
  result = nil
  
  @pending_mutex.synchronize do
    @pending_enqueues[task_id] = { cond: cond, result: nil }
  end

  send_message(payload)

  @pending_mutex.synchronize do
    # Wait up to 5 seconds for Ack
    cond.wait(@pending_mutex, 5.0) if @pending_enqueues[task_id][:result].nil?
    pending = @pending_enqueues.delete(task_id)
    result = pending[:result] if pending
  end

  if result.nil?
    raise "[Snerd] Timeout waiting for daemon Ack on task #{task_id}"
  elsif result.is_a?(StandardError)
    raise result
  end
  
  true
end

#register_handler(task_type, &block) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/snerdmq/queue.rb', line 33

def register_handler(task_type, &block)
  @handlers_mutex.synchronize do
    @handlers[task_type] = block
  end

  if @io && !@shutting_down
    send_message({
      action: "register",
      task_type: task_type
    })
  end
end

#shutdown ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/snerdmq/queue.rb', line 110

def shutdown
  @shutting_down = true
  begin
    Process.kill("TERM", @io.pid) if @io && @io.pid
  rescue Errno::ESRCH, Errno::ECHILD
    # Process already dead
  end
  
  @listener_thread.join(2) if @listener_thread
  @io.close if @io && !@io.closed?
end

#start_listening ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/snerdmq/queue.rb', line 46

def start_listening
  args = []
  args << @storage_path if @storage_path

  # Open a bidirectional pipe to the Rust daemon
  @io = IO.popen([@binary_path] + args, "r+")

  # Re-register all existing handlers
  @handlers_mutex.synchronize do
    @handlers.keys.each do |task_type|
      send_message({
        action: "register",
        task_type: task_type
      })
    end
  end

  @listener_thread = Thread.new do
    listen_to_stdout
  end
end