Class: RabbitMQClient::Queue
- Inherits:
-
Object
- Object
- RabbitMQClient::Queue
- Defined in:
- lib/rabbitmq_client.rb
Instance Method Summary collapse
- #bind(exchange, routing_key = '') ⇒ Object
- #delete ⇒ Object
-
#initialize(name, channel, durable = false) ⇒ Queue
constructor
A new instance of Queue.
- #loop_subscribe(&block) ⇒ Object
- #persistent_publish(message_body, props = MessageProperties::PERSISTENT_TEXT_PLAIN) ⇒ Object
-
#publish(message_body, props = nil) ⇒ Object
Set props for different type of message.
- #purge ⇒ Object
- #retrieve ⇒ Object
- #subscribe(&block) ⇒ Object
- #unbind ⇒ Object
Constructor Details
#initialize(name, channel, durable = false) ⇒ Queue
Returns a new instance of Queue.
39 40 41 42 43 44 45 |
# File 'lib/rabbitmq_client.rb', line 39 def initialize(name, channel, durable=false) @name = name @durable = durable @channel = channel @channel.queue_declare(name, durable) self end |
Instance Method Details
#bind(exchange, routing_key = '') ⇒ Object
47 48 49 50 51 52 53 |
# File 'lib/rabbitmq_client.rb', line 47 def bind(exchange, routing_key='') raise RabbitMQClientError, "queue and exchange has different durable property" unless @durable == exchange.durable @routing_key = routing_key @exchange = exchange @channel.queue_bind(@name, @exchange.name, @routing_key) self end |
#delete ⇒ Object
117 118 119 |
# File 'lib/rabbitmq_client.rb', line 117 def delete @channel.queue_delete(@name) end |
#loop_subscribe(&block) ⇒ Object
97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
# File 'lib/rabbitmq_client.rb', line 97 def loop_subscribe(&block) no_ack = false consumer = QueueingConsumer.new(@channel) @channel.basic_consume(@name, no_ack, consumer) loop do begin delivery = consumer.next_delivery = Marshal.load(String.from_java_bytes(delivery.get_body)) block.call @channel.basic_ack(delivery.get_envelope.get_delivery_tag, false) rescue InterruptedException => ie next end end end |
#persistent_publish(message_body, props = MessageProperties::PERSISTENT_TEXT_PLAIN) ⇒ Object
73 74 75 76 |
# File 'lib/rabbitmq_client.rb', line 73 def persistent_publish(, props=MessageProperties::PERSISTENT_TEXT_PLAIN) raise RabbitMQClientError, "can only publish persistent message to durable queue" unless @durable publish(, props) end |
#publish(message_body, props = nil) ⇒ Object
Set props for different type of message. Currently they are: RabbitMQClient::MessageProperties::MINIMAL_BASIC RabbitMQClient::MessageProperties::MINIMAL_PERSISTENT_BASIC RabbitMQClient::MessageProperties::BASIC RabbitMQClient::MessageProperties::PERSISTENT_BASIC RabbitMQClient::MessageProperties::TEXT_PLAIN RabbitMQClient::MessageProperties::PERSISTENT_TEXT_PLAIN
66 67 68 69 70 71 |
# File 'lib/rabbitmq_client.rb', line 66 def publish(, props=nil) auto_bind = Marshal.dump().to_java_bytes @channel.basic_publish(@exchange.name, @routing_key, props, ) end |
#purge ⇒ Object
113 114 115 |
# File 'lib/rabbitmq_client.rb', line 113 def purge @channel.queue_purge(@name) end |
#retrieve ⇒ Object
78 79 80 81 82 83 84 85 86 87 88 89 90 |
# File 'lib/rabbitmq_client.rb', line 78 def retrieve auto_bind = nil no_ack = false response = @channel.basic_get(@name, no_ack) if response props = response.get_props = Marshal.load(String.from_java_bytes(response.get_body)) delivery_tag = response.get_envelope.get_delivery_tag @channel.basic_ack(delivery_tag, false) end end |
#subscribe(&block) ⇒ Object
92 93 94 95 |
# File 'lib/rabbitmq_client.rb', line 92 def subscribe(&block) no_ack = false @channel.basic_consume(@name, no_ack, QueueConsumer.new(@channel, block)) end |
#unbind ⇒ Object
55 56 57 |
# File 'lib/rabbitmq_client.rb', line 55 def unbind @channel.queue_unbind(@name, @exchange.name, @routing_key) if @exchange end |