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
permalink #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
permalink #ack(msg) ⇒ Object
[View source]
8 9 10 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 8 def ack(msg) @conn.ack(msg.headers["message-id"]) end |
permalink #disconnect(message_handler) ⇒ Object
[View source]
16 17 18 19 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 16 def disconnect() unsubscribe(destination_for()) @conn.disconnect end |
permalink #receive(options) ⇒ Object
[View source]
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 |
permalink #receive_once(destination, opts) ⇒ Object
[View source]
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 |
permalink #receive_with(message_handler) ⇒ Object
[View source]
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 |
permalink #send_message(destination, message, options) ⇒ Object
[View source]
49 50 51 52 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 49 def (destination, , ) RosettaQueue.logger.info("Publishing to #{destination} :: #{}") @conn.send(destination, , ) end |
permalink #subscribe(destination, options) ⇒ Object
[View source]
54 55 56 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 54 def subscribe(destination, ) @conn.subscribe(destination, ) end |
permalink #unsubscribe(destination) ⇒ Object
[View source]
58 59 60 |
# File 'lib/rosetta_queue/adapters/stomp.rb', line 58 def unsubscribe(destination) @conn.unsubscribe(destination) end |