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.



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
32
33
 
# File 'lib/snerdmq/queue.rb', line 7

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 = {}
  @max_retry_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, execute_at: nil, cron: nil, webhook_url: nil, max_execution_seconds: nil) ⇒ Object 



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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/snerdmq/queue.rb', line 76

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, execute_at: nil, cron: nil, webhook_url: nil, max_execution_seconds: 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
  
  if execute_at
    payload[:execute_at] = execute_at.respond_to?(:iso8601) ? execute_at.iso8601 : execute_at.to_s
  end
  payload[:cron] = cron if cron
  payload[:webhook_url] = webhook_url if webhook_url
  payload[:max_execution_seconds] = max_execution_seconds if max_execution_seconds

  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 



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

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

#register_max_retry_handler(task_type, &block) ⇒ Object 



48
49
50
51
52
 
# File 'lib/snerdmq/queue.rb', line 48

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

#shutdown ⇒ Object 



125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/snerdmq/queue.rb', line 125

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 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/snerdmq/queue.rb', line 54

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