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

Instance Method Details

#enqueue(task_id:, task_type:, data:, max_retries: 3, retry_after_hours: 0.0) ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/snerdmq/queue.rb', line 64

def enqueue(task_id:, task_type:, data:, max_retries: 3, retry_after_hours: 0.0)
  raise "[Snerd] Cannot enqueue task: Queue is not running. Call start_listening first." if @io.nil? || @shutting_down
  
  send_message({
    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
  })
end

#register_handler(task_type, &block) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/snerdmq/queue.rb', line 29

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 



77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/snerdmq/queue.rb', line 77

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 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/snerdmq/queue.rb', line 42

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