@@ -53,6 +53,10 @@ module AMQ end end + def close + destroy + end + def new_session() s = Session.new(@conn.session_new(), self) if block_given? @@ -79,6 +83,10 @@ module AMQ @sess = rwire_session end + def close + destroy + end + def publish(args) args[:body] ||= "" args[:mandatory] ||= false @@ -117,15 +125,25 @@ module AMQ def declare_queue(args) args[:queue] ||= args[:name] || nil - args[:passive] ||= false - args[:durable] ||= false - args[:exclusive] ||= false - args[:auto_delete] ||= false - @sess.declare_queue(args[:queue], - args[:passive], - args[:durable], - args[:exclusive], - args[:auto_delete]) + args[:passive] ||= false + args[:durable] ||= false + args[:exclusive] ||= false + args[:auto_delete] ||= false + @sess.declare_queue(args[:queue], + args[:passive], + args[:durable], + args[:exclusive], + args[:auto_delete]) + + end + + def delete_queue(args) + args[:queue] ||= args[:name] || nil + args[:if_unsed] = true unless args.has_key?(:if_unsed) + args[:if_empty] = true unless args.has_key?(:if_empty) + @sess.delete_queue(args[:queue], + args[:if_unsed], + args[:if_empty]) end @@ -169,9 +187,11 @@ module AMQ end # begin end # while end # loop + end + ensure + if block_given? @sess.basic_cancel(consumer_tag) end - end require 'pp' @@ -179,9 +199,11 @@ module AMQ # (in seconds) for waiting for reply can be specified def request(args) args = args.dup - args[:immediate] = true + args[:immediate] = false args[:mandatory] = true + consumer_tag = nil + # Create a private reply queue queue = declare_and_bind_private_queue() # Send out the request @@ -194,6 +216,8 @@ module AMQ consume(:queue => queue, :exclusive => true) + consumer_tag = @sess.consumer_tag + rc = wait(args[:timeout]) if rc == 0 if basic_returned_count > 0 @@ -208,6 +232,8 @@ module AMQ else raise AMQError.new("Failed. Interrupted while waiting for response.") end + ensure + @sess.basic_cancel(consumer_tag) if consumer_tag end def method_missing(meth, *args, &blk) lib/amq/openamq.rb @@ -491,6 +491,32 @@ static VALUE rwire_amq_client_session_declare_queue(VALUE self, return self; } +static VALUE rwire_amq_client_session_delete_queue(VALUE self, + VALUE queuename, + VALUE if_unused, + VALUE if_empty) +{ + + char *_queuename = NULL; + amq_client_session_t *session = NULL; + + if (queuename != Qnil) + _queuename = StringValuePtr(queuename); + + bool _if_unused = (if_unused != Qfalse); + bool _if_empty = (if_empty != Qfalse); + + Data_Get_Struct(self, amq_client_session_t, session); + amq_client_session_queue_delete( + session, + 0, + _queuename, + _if_unused, + _if_empty); +//TODO check for a more useful value to return + return self; +} + static VALUE rwire_amq_client_session_bind_queue(VALUE self, VALUE queuename, VALUE exchange, @@ -948,6 +974,7 @@ void Init_rwire() RB_DEF_SESS_METHOD(declare_exchange, 6); //RB_DEF_SESS_METHOD(exchange_delete, 0); RB_DEF_SESS_METHOD(declare_queue, 5); + RB_DEF_SESS_METHOD(delete_queue, 3); RB_DEF_SESS_METHOD(bind_queue, 3); //RB_DEF_SESS_METHOD(queue_unbind, 0); //RB_DEF_SESS_METHOD(queue_purge, 0); rwire.c 595f3f6a2e13bc87ddd737e40d8c2cffc3ec48b9 Chris Wong chris@chriswongstudio.com http://github.com/lightwave/ropenamq/commit/3d0d023dcb090b49258310ecf744ed5875192f65 3d0d023dcb090b49258310ecf744ed5875192f65 2009-04-27T20:35:22-07:00 2009-04-27T20:35:22-07:00 Properly release consumer during error condition in consume and request methods. 6751799e70db3ba42a9708e67a4bc19b6a097cf2 Chris Wong chris@chriswongstudio.com