Class: RosettaQueue::Gateway::StompAdapter
- Inherits:
-
BaseAdapter
- Object
- BaseAdapter
- RosettaQueue::Gateway::StompAdapter
- Defined in:
- lib/rosetta_queue/adapters/stomp.rb
Instance Method Summary collapse
- #ack(msg) ⇒ Object
- #disconnect(message_handler) ⇒ Object
-
#initialize(user, password, host, port) ⇒ StompAdapter
constructor
A new instance of StompAdapter.
- #receive(options) ⇒ Object
- #receive_once(destination, opts) ⇒ Object
- #receive_with(message_handler) ⇒ Object
- #send_message(destination, message, options) ⇒ Object
- #subscribe(destination, options) ⇒ Object
- #unsubscribe(destination) ⇒ Object
Constructor Details
#initialize(user, password, host, port) ⇒ StompAdapter
Returns a new instance of StompAdapter.
12 13 14 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 12 def initialize(user, password, host, port) @conn = Stomp::Connection.open(user, password, host, port, true) end |
Instance Method Details
#ack(msg) ⇒ Object
8 9 10 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 8 def ack(msg) @conn.ack(msg.headers["message-id"]) end |
#disconnect(message_handler) ⇒ Object
16 17 18 19 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 16 def disconnect() unsubscribe(destination_for()) @conn.disconnect end |
#receive(options) ⇒ Object
21 22 23 24 25 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 21 def receive() msg = @conn.receive ack(msg) unless [:ack].nil? msg end |
#receive_once(destination, opts) ⇒ Object
27 28 29 30 31 32 33 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 27 def receive_once(destination, opts) subscribe(destination, opts) msg = receive(opts).body unsubscribe(destination) RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") filter_receiving(msg) end |
#receive_with(message_handler) ⇒ Object
35 36 37 38 39 40 41 42 43 44 45 46 47 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 35 def receive_with() = () destination = destination_for() @conn.subscribe(destination, ) running do msg = receive().body Thread.current[:processing] = true RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") .(filter_receiving(msg)) Thread.current[:processing] = false end end |
#send_message(destination, message, options) ⇒ Object
49 50 51 52 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 49 def (destination, , ) RosettaQueue.logger.info("Publishing to #{destination} :: #{message}") @conn.send(destination, , ) end |
#subscribe(destination, options) ⇒ Object
54 55 56 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 54 def subscribe(destination, ) @conn.subscribe(destination, ) end |
#unsubscribe(destination) ⇒ Object
58 59 60 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 58 def unsubscribe(destination) @conn.unsubscribe(destination) end |