Merge with default

deps/rabbitmq_stomp/Makefile

@@ -7,8 +7,6 @@ TEST_APPS=rabbit_stomp
 TEST_SCRIPTS=./test/test.py
 UNIT_TEST_COMMANDS=eunit:test([rabbit_stomp_test_util,rabbit_stomp_test_frame],[verbose])
 
-TEST_ARGS=-rabbit_stomp listeners "[{\"0.0.0.0\",61613}]"
-
 include ../include.mk
 
 testdeps:

deps/rabbitmq_stomp/README.md

@@ -32,21 +32,26 @@ You need to install the rabbit\_stomp.ez and amqp\_client.ez packages.
 
 ## Running the STOMP adapter
 
-### Configuring the server to start the plugin automatically
+When no configuration is specified the STOMP Adapter will listen on
+all interfaces on port 61613.
 
-Most RabbitMQ server packages are set up to cause the server to pick
+To change this, edit your [Configuration file](http://www.rabbitmq.com/install.html#configfile),
-up configuration from `/etc/rabbitmq/rabbitmq.conf`. To tell the
+to contain a tcp_listeners variable for the rabbit_stomp application.
-server to start your plugin, first make sure it is compiled, and then
-add the following text to `/etc/rabbitmq/rabbitmq.conf`:
 
-    SERVER_START_ARGS='-rabbit_stomp listeners [{"0.0.0.0",61613}]'
+For example, a complete configuration file which changes the listener
+port to 12345 would look like:
 
-Then restart the server with
+[
+  {rabbit_stomp, [{tcp_listeners, [12345]} ]}
+].
 
-    sudo /etc/init.d/rabbitmq-server restart
+while one which changes the listener to listen only on localhost (for
+both IPv4 and IPv6) would look like:
 
-When no configuration is specified the STOMP Adapter will listen on 
+[
-localhost by default.
+  {rabbit_stomp, [{tcp_listeners, [{"127.0.0.1", 61613},
+                                   {"::1",       61613} ]} ]}
+].
 
 ### Checking that the adapter is running
 

deps/rabbitmq_stomp/ebin/rabbit_stomp.app.in

@@ -12,5 +12,5 @@
   ]},
   {registered, []},
   {mod, {rabbit_stomp, []}},
-  {env, [{listeners, [{"127.0.0.1", 61613}]}]},
+  {env, [{tcp_listeners, [61613]}]},
   {applications, [kernel, stdlib, rabbit]}]}.

deps/rabbitmq_stomp/src/rabbit_stomp.erl

@@ -45,7 +45,7 @@ stop(_State) ->
     ok.
 
 parse_listener_configuration() ->
-    case application:get_env(listeners) of
+    case application:get_env(tcp_listeners) of
         undefined -> throw({error, {stomp_configuration_not_found}});
         {ok, Listeners} -> Listeners
     end.

deps/rabbitmq_stomp/src/rabbit_stomp_processor.erl

@@ -42,7 +42,7 @@
                 connection, subscriptions, version,
                 start_heartbeat_fun}).
 
--record(subscription, {dest_hdr, channel, multi_ack}).
+-record(subscription, {dest_hdr, channel, multi_ack, description}).
 
 -define(SUPPORTED_VERSIONS, ["1.0", "1.1"]).
 -define(DEFAULT_QUEUE_PREFETCH, 1).
@@ -124,6 +124,8 @@ handle_cast(client_timeout, State) ->
 
 handle_info(#'basic.consume_ok'{}, State) ->
     {noreply, State};
+handle_info(#'basic.cancel_ok'{}, State) ->
+    {noreply, State};
 handle_info({Delivery = #'basic.deliver'{},
              #amqp_msg{props = Props, payload = Payload}}, State) ->
     {noreply, send_delivery(Delivery, Props, Payload, State)}.
@@ -234,20 +236,42 @@ cancel_subscription({error, _}, State) ->
           "UNSUBSCRIBE must include a 'destination' or 'id' header\n",
           State);
 
-cancel_subscription({ok, ConsumerTag}, State = #state{subscriptions = Subs}) ->
+cancel_subscription({ok, ConsumerTag, Description},
+                    State = #state{channel       = MainChannel,
+                                   subscriptions = Subs}) ->
     case dict:find(ConsumerTag, Subs) of
         error ->
             error("No subscription found",
-                  "UNSUBSCRIBE must refer to an existing subscription\n",
+                  "UNSUBSCRIBE must refer to an existing subscription.\n"
+                  "Subscription to ~p not found.\n",
+                  [Description],
                   State);
-        {ok, #subscription{channel = Channel}} ->
+        {ok, #subscription{channel = SubChannel}} ->
-            ok(send_method(#'basic.cancel'{consumer_tag = ConsumerTag,
+            case amqp_channel:call(SubChannel,
-                                           nowait       = true},
+                                   #'basic.cancel'{
-                           Channel,
+                                     consumer_tag = ConsumerTag}) of
-                           State#state{subscriptions =
+                #'basic.cancel_ok'{consumer_tag = ConsumerTag} ->
-                                           dict:erase(ConsumerTag, Subs)}))
+                    NewSubs = dict:erase(ConsumerTag, Subs),
+                    ensure_subchannel_closed(SubChannel,
+                                             MainChannel,
+                                             State#state{
+                                               subscriptions = NewSubs});
+                _ ->
+                    error("Failed to cancel subscription",
+                          "UNSUBSCRIBE to ~p failed.\n",
+                          [Description],
+                          State)
+            end
     end.
 
+ensure_subchannel_closed(SubChannel, MainChannel, State)
+  when SubChannel == MainChannel ->
+    ok(State);
+
+ensure_subchannel_closed(SubChannel, _MainChannel, State) ->
+    amqp_channel:close(SubChannel),
+    ok(State).
+
 with_destination(Command, Frame, State, Fun) ->
     case rabbit_stomp_frame:header(Frame, "destination") of
         {ok, DestHdr} ->
@@ -320,7 +344,7 @@ do_subscribe(Destination, DestHdr, Frame,
 
     {ok, Queue} = ensure_queue(subscribe, Destination, Channel),
 
-    {ok, ConsumerTag} = rabbit_stomp_util:consumer_tag(Frame),
+    {ok, ConsumerTag, Description} = rabbit_stomp_util:consumer_tag(Frame),
 
     amqp_channel:subscribe(Channel,
                            #'basic.consume'{
@@ -336,9 +360,10 @@ do_subscribe(Destination, DestHdr, Frame,
 
     ok(State#state{subscriptions =
                        dict:store(ConsumerTag,
-                                  #subscription{dest_hdr  = DestHdr,
+                                  #subscription{dest_hdr    = DestHdr,
-                                                channel   = Channel,
+                                                channel     = Channel,
-                                                multi_ack = IsMulti},
+                                                multi_ack   = IsMulti,
+                                                description = Description},
                                   Subs)}).
 
 do_send(Destination, _DestHdr,
@@ -400,7 +425,7 @@ send_delivery(Delivery = #'basic.deliver'{consumer_tag = ConsumerTag},
               State);
         error ->
             send_error("Subscription not found",
-                       "There is no current subscription '~s'.",
+                       "There is no current subscription with tag '~s'.",
                        [ConsumerTag],
                        State)
     end.
@@ -414,6 +439,9 @@ send_method(Method, State = #state{channel = Channel}) ->
 
 send_method(Method, Properties, BodyFragments,
             State = #state{channel = Channel}) ->
+    send_method(Method, Channel, Properties, BodyFragments, State).
+
+send_method(Method, Channel, Properties, BodyFragments, State) ->
     amqp_channel:call(Channel, Method, #amqp_msg{
                                 props = Properties,
                                 payload = lists:reverse(BodyFragments)}),
@@ -508,8 +536,7 @@ abort_transaction(Transaction, State0) ->
 perform_transaction_action({Method}, State) ->
     send_method(Method, State);
 perform_transaction_action({Channel, Method}, State) ->
-    amqp_channel:call(Channel, Method),
+    send_method(Method, Channel, State);
-    State;
 perform_transaction_action({Method, Props, BodyFragments}, State) ->
     send_method(Method, Props, BodyFragments, State).
 

deps/rabbitmq_stomp/src/rabbit_stomp_sup.erl

@@ -52,25 +52,24 @@ init([Listeners]) ->
     {ok, {{one_for_all, 10, 10}, ChildSpecs}}.
 
 make_listener_specs(Listeners) ->
-    lists:foldl(
+    [make_listener_spec(Spec)
-      fun({Host, Port}, Acc) ->
+     || Spec <- lists:append([rabbit_networking:check_tcp_listener_address(
-              {IPAddress, Name} = rabbit_networking:check_tcp_listener_address(
+                                rabbit_stomp_listener_sup, Listener)
-                                    rabbit_stomp_listener_sup,
+                              || Listener <- Listeners])].
-                                    Host,
-                                    Port),
-              [{Name,
-                {tcp_listener_sup, start_link,
-                 [IPAddress, Port,
-                  [binary,
-                   {packet, raw},
-                   {reuseaddr, true},
-                   {backlog, 128}],
-                  {?MODULE, listener_started, []},
-                  {?MODULE, listener_stopped, []},
-                  {?MODULE, start_client, []}, "STOMP Listener"]},
-                transient, infinity, supervisor, [tcp_listener_sup]} | Acc]
 
-      end, [], Listeners).
+make_listener_spec({IPAddress, Port, Family, Name}) ->
+    {Name,
+     {tcp_listener_sup, start_link,
+      [IPAddress, Port,
+       [Family,
+        binary,
+        {packet, raw},
+        {reuseaddr, true},
+        {backlog, 128}],
+       {?MODULE, listener_started, []},
+       {?MODULE, listener_stopped, []},
+       {?MODULE, start_client, []}, "STOMP Listener"]},
+     transient, infinity, supervisor, [tcp_listener_sup]}.
 
 listener_started(IPAddress, Port) ->
     rabbit_networking:tcp_listener_started(stomp, IPAddress, Port).

deps/rabbitmq_stomp/src/rabbit_stomp_util.erl

@@ -60,11 +60,11 @@
 consumer_tag(Frame) ->
     case rabbit_stomp_frame:header(Frame, "id") of
         {ok, Str} ->
-            {ok, list_to_binary("T_" ++ Str)};
+            {ok, list_to_binary("T_" ++ Str), "id='" ++ Str ++ "'"};
         not_found ->
             case rabbit_stomp_frame:header(Frame, "destination") of
                 {ok, DestHdr} ->
-                    {ok, list_to_binary("Q_" ++ DestHdr)};
+                    {ok, list_to_binary("Q_" ++ DestHdr), "destination='" ++ DestHdr ++ "'"};
                 not_found ->
                     {error, missing_destination_header}
             end

deps/rabbitmq_stomp/test/base.py

@@ -49,7 +49,7 @@ class BaseTest(unittest.TestCase):
         self.assertEquals("foo", msg['message'])
         self.assertEquals(dest, msg['headers']['destination'])
 
-   def assertListener(self, errMsg, numMsgs=0, numErrs=0, numRcts=0, timeout=3):
+   def assertListener(self, errMsg, numMsgs=0, numErrs=0, numRcts=0, timeout=1):
         if numMsgs + numErrs + numRcts > 0:
             self.assertTrue(self.listener.await(timeout), errMsg + " (#awaiting)")
         else:
@@ -58,7 +58,7 @@ class BaseTest(unittest.TestCase):
         self.assertEquals(numErrs, len(self.listener.errors), errMsg + " (#errors)")
         self.assertEquals(numRcts, len(self.listener.receipts), errMsg + " (#receipts)")
 
-   def assertListenerAfter(self, verb, errMsg="", numMsgs=0, numErrs=0, numRcts=0, timeout=3):
+   def assertListenerAfter(self, verb, errMsg="", numMsgs=0, numErrs=0, numRcts=0, timeout=1):
         num = numMsgs + numErrs + numRcts
         self.listener.reset(num if num>0 else 1)
         verb()
@@ -77,13 +77,13 @@ class WaitableListener(object):
 
     def on_receipt(self, headers, message):
         if self.debug:
-            print '(on_message) message:', message, 'headers:', headers
+            print '(on_receipt) message:', message, 'headers:', headers
         self.receipts.append({'message' : message, 'headers' : headers})
         self.latch.countdown()
 
     def on_error(self, headers, message):
         if self.debug:
-            print '(on_message) message:', message, 'headers:', headers
+            print '(on_error) message:', message, 'headers:', headers
         self.errors.append({'message' : message, 'headers' : headers})
         self.latch.countdown()
 
@@ -93,11 +93,12 @@ class WaitableListener(object):
         self.messages.append({'message' : message, 'headers' : headers})
         self.latch.countdown()
 
-    def reset(self,count=1):
+    def reset(self, count=1):
         if self.debug:
-            print '(reset listener) #messages:', len(self.messages),
+            print '(reset listener)',
-            print '#errors', len(self.errors),
+            print '#messages:', len(self.messages),
-            print '#receipts', len(self.receipts), 'Now expecting:', count
+            print '#errors:', len(self.errors),
+            print '#receipts:', len(self.receipts), 'Now expecting:', count
         self.messages = []
         self.errors = []
         self.receipts = []
@@ -116,7 +117,8 @@ class Latch(object):
 
    def countdown(self):
       self.cond.acquire()
-      self.count -= 1
+      if self.count > 0:
+         self.count -= 1
       if self.count == 0:
          self.cond.notify_all()
       self.cond.release()

deps/rabbitmq_stomp/test/destinations.py

@@ -1,6 +1,7 @@
 import unittest
 import stomp
 import base
+import time
 
 class TestExchange(base.BaseTest):
 
@@ -32,7 +33,7 @@ class TestExchange(base.BaseTest):
         self.assertEquals("not_found", err['headers']['message'])
         self.assertEquals("no exchange 'does.not.exist' in vhost '/'\n",
                           err['message'])
-
+        time.sleep(1)
         self.assertFalse(self.conn.is_connected())
 
     def __test_exchange_send_rec(self, exchange, route = None):

deps/rabbitmq_stomp/test/errors.py

@@ -65,4 +65,3 @@ class TestErrors(base.BaseTest):
                               dtype + " destination\n",
                           err['message'])
 
-

deps/rabbitmq_stomp/test/lifecycle.py

@@ -7,21 +7,46 @@ class TestLifecycle(base.BaseTest):
 
     def test_unsubscribe_exchange_destination(self):
         ''' Test UNSUBSCRIBE command with exchange'''
-        self.unsub_test(self.sub_and_send("/exchange/amq.fanout"))
+        d = "/exchange/amq.fanout"
+        self.unsub_test(d, self.sub_and_send(d))
+
+    def test_unsubscribe_exchange_destination_with_receipt(self):
+        ''' Test receipted UNSUBSCRIBE command with exchange'''
+        d = "/exchange/amq.fanout"
+        self.unsub_test(d, self.sub_and_send(d, receipt="unsub.rct"), numRcts=1)
 
     def test_unsubscribe_queue_destination(self):
         ''' Test UNSUBSCRIBE command with queue'''
-        self.unsub_test(self.sub_and_send("/queue/unsub01"))
+        d = "/queue/unsub01"
+        self.unsub_test(d, self.sub_and_send(d))
+
+    def test_unsubscribe_queue_destination_with_receipt(self):
+        ''' Test receipted UNSUBSCRIBE command with queue'''
+        d = "/queue/unsub02"
+        self.unsub_test(d, self.sub_and_send(d, receipt="unsub.rct"), numRcts=1)
 
     def test_unsubscribe_exchange_id(self):
         ''' Test UNSUBSCRIBE command with exchange by id'''
-        self.unsub_test(self.subid_and_send("/exchange/amq.fanout", "exchid"))
+        d = "/exchange/amq.fanout"
+        self.unsub_test(d, self.sub_and_send(d, subid="exchid"))
+
+    def test_unsubscribe_exchange_id_with_receipt(self):
+        ''' Test receipted UNSUBSCRIBE command with exchange by id'''
+        d = "/exchange/amq.fanout"
+        self.unsub_test(d, self.sub_and_send(d, subid="exchid", receipt="unsub.rct"), numRcts=1)
 
     def test_unsubscribe_queue_id(self):
         ''' Test UNSUBSCRIBE command with queue by id'''
-        self.unsub_test(self.subid_and_send("/queue/unsub02", "queid"))
+        d = "/queue/unsub03"
+        self.unsub_test(d, self.sub_and_send(d, subid="queid"))
+
+    def test_unsubscribe_queue_id_with_receipt(self):
+        ''' Test receipted UNSUBSCRIBE command with queue by id'''
+        d = "/queue/unsub04"
+        self.unsub_test(d, self.sub_and_send(d, subid="queid", receipt="unsub.rct"), numRcts=1)
 
     def test_connect_version_1_1(self):
+        ''' Test CONNECT with version 1.1'''
         self.conn.disconnect()
         new_conn = self.create_connection(version="1.1,1.0")
         try:
@@ -30,6 +55,7 @@ class TestLifecycle(base.BaseTest):
             new_conn.disconnect()
 
     def test_heartbeat_disconnects_client(self):
+        ''' Test heartbeat disconnection'''
         self.conn.disconnect()
         new_conn = self.create_connection(heartbeat="1500,0")
         try:
@@ -42,9 +68,8 @@ class TestLifecycle(base.BaseTest):
             if new_conn.is_connected():
                 new_conn.disconnect()
 
-
-
     def test_unsupported_version(self):
+        ''' Test unsupported version on CONNECT command'''
         self.conn.disconnect()
         new_conn = stomp.Connection(user="guest",
                                     passcode="guest",
@@ -62,29 +87,35 @@ class TestLifecycle(base.BaseTest):
                 new_conn.disconnect()
 
     def test_disconnect(self):
-        ''' Run DISCONNECT command '''
+        ''' Test DISCONNECT command'''
         self.conn.disconnect()
         self.assertFalse(self.conn.is_connected())
 
-    def unsub_test(self, verbs):
+    def unsub_test(self, dest, verbs, numRcts=0):
+        def afterfun():
+            self.conn.send("after-test", destination=dest)
         subverb, unsubverb = verbs
-        self.assertListenerAfter(subverb,
+        self.assertListenerAfter(subverb, numMsgs=1,
-                           numMsgs=1, errMsg="FAILED to subscribe and send")
+                           errMsg="FAILED to subscribe and send")
-        self.assertListenerAfter(unsubverb,
+        self.assertListenerAfter(unsubverb, numRcts=numRcts, 
+                           errMsg="Incorrect responses from UNSUBSCRIBE")
+        self.assertListenerAfter(afterfun,
                            errMsg="Still receiving messages")
 
-    def subid_and_send(self, dest, subid):
+    def sub_and_send(self, dest, subid="", receipt=""):
         def subfun():
-            self.conn.subscribe(destination=dest, id=subid)
+            if subid=="":
+                self.conn.subscribe(destination=dest)
+            else:
+                self.conn.subscribe(destination=dest, id=subid)
             self.conn.send("test", destination=dest)
         def unsubfun():
-            self.conn.unsubscribe(id=subid)
+            if subid=="" and receipt=="":
-        return subfun, unsubfun
+                self.conn.unsubscribe(destination=dest)
-
+            elif receipt=="":
-    def sub_and_send(self, dest):
+                self.conn.unsubscribe(id=subid)
-        def subfun():
+            elif subid=="":
-            self.conn.subscribe(destination=dest)
+                self.conn.unsubscribe(destination=dest, receipt=receipt)
-            self.conn.send("test", destination=dest)
+            else:
-        def unsubfun():
+                self.conn.unsubscribe(id=subid, receipt=receipt)
-            self.conn.unsubscribe(destination=dest)
         return subfun, unsubfun

deps/rabbitmq_stomp/test/rabbit_stomp_test_util.erl

@@ -150,11 +150,11 @@ ack_mode_client_individual_test() ->
 
 consumer_tag_id_test() ->
     Frame = #stomp_frame{headers = [{"id", "foo"}]},
-    {ok, <<"T_foo">>} = rabbit_stomp_util:consumer_tag(Frame).
+    {ok, <<"T_foo">>, _} = rabbit_stomp_util:consumer_tag(Frame).
 
 consumer_tag_destination_test() ->
     Frame = #stomp_frame{headers = [{"destination", "foo"}]},
-    {ok, <<"Q_foo">>} = rabbit_stomp_util:consumer_tag(Frame).
+    {ok, <<"Q_foo">>, _} = rabbit_stomp_util:consumer_tag(Frame).
 
 consumer_tag_invalid_test() ->
     Frame = #stomp_frame{headers = []},