Merged stable into default

deps/rabbitmq_stomp/src/rabbit_stomp_frame.erl

@@ -38,6 +38,7 @@ initial_state() -> none.
 %%  o   Escape codes for header names and values include \r for CR
 %%      and CR is not allowed.
 %%  o   Header names and values are not limited to UTF-8 strings.
+%%  o   Header values may contain unescaped colons
 %%
 %%  frame_seq   ::= *(noise frame)
 %%  noise       ::= *(NUL | eol)
@@ -113,8 +114,6 @@ parser(              Rest,          headers ,  State) -> goto(headers,  hdrname,
 parser(<<?COLON,     Rest/binary>>, hdrname ,  State) -> goto(hdrname,  hdrvalue, Rest, State);
 parser(<<?LF,        Rest/binary>>, hdrname ,  State) -> goto(hdrname,  headers,  Rest, State);
 parser(<<?LF,        Rest/binary>>, hdrvalue,  State) -> goto(hdrvalue, headers,  Rest, State);
-%% trap invalid colons
-parser(<<?COLON,    _Rest/binary>>, hdrvalue, _State) -> {error, {unexpected_char_in_header_value, [?COLON]}};
 %% accumulate
 parser(<<Ch:8,       Rest/binary>>, Term    ,  State) -> parser(Rest, Term, accum(Ch, State)).
 

deps/rabbitmq_stomp/src/rabbit_stomp_processor.erl

@@ -691,6 +691,8 @@ ensure_reply_to(Frame = #stomp_frame{headers = Headers}, State) ->
         {ok, ReplyTo} ->
             {ok, Destination} = rabbit_routing_util:parse_endpoint(ReplyTo),
             case rabbit_routing_util:dest_temp_queue(Destination) of
+                none ->
+                    {Frame, State};
                 TempQueueId ->
                     {ReplyQueue, State1} =
                         ensure_reply_queue(TempQueueId, State),
@@ -698,9 +700,7 @@ ensure_reply_to(Frame = #stomp_frame{headers = Headers}, State) ->
                        headers = lists:keyreplace(
                                    ?HEADER_REPLY_TO, 1, Headers,
                                    {?HEADER_REPLY_TO, ReplyQueue})},
-                     State1};
-                none ->
-                    {Frame, State}
+                     State1}
             end
     end.
 

deps/rabbitmq_stomp/test/src/destinations.py

@@ -318,6 +318,34 @@ class TestReplyQueue(base.BaseTest):
             conn2.stop()
             conn3.stop()
 
+    def test_perm_reply_queue(self):
+        '''As test_reply_queue, but with a non-temp reply queue'''
+
+        known = '/queue/known'
+        reply = '/queue/reply'
+
+        ## Client 1 uses pre-supplied connection and listener
+        ## Set up client 2
+        conn1, listener1 = self.create_subscriber_connection(reply)
+        conn2, listener2 = self.create_subscriber_connection(known)
+
+        try:
+            conn1.send("test", destination=known,
+                       headers = {"reply-to": reply})
+
+            self.assertTrue(listener2.await(5))
+            self.assertEquals(1, len(listener2.messages))
+
+            reply_to = listener2.messages[0]['headers']['reply-to']
+            self.assertTrue(reply_to == reply)
+
+            conn2.send("reply", destination=reply_to)
+            self.assertTrue(listener1.await(5))
+            self.assertEquals("reply", listener1.messages[0]['message'])
+        finally:
+            conn1.stop()
+            conn2.stop()
+
 class TestDurableSubscription(base.BaseTest):
 
     ID = 'test.subscription'

deps/rabbitmq_stomp/test/src/rabbit_stomp_test_frame.erl

@@ -160,7 +160,11 @@ header_value_with_cr_test() ->
 
 header_value_with_colon_test() ->
     Content = "COMMAND\nheader:val:ue\n\n\0",
-    {error, {unexpected_char_in_header_value, ":"}} = parse(Content).
+    {ok, Frame, _} = parse(Content),
+    ?assertEqual(Frame,
+                 #stomp_frame{ command     = "COMMAND",
+                               headers     = [{"header", "val:ue"}],
+                               body_iolist = []}).
 
 headers_escaping_roundtrip_test() ->
     Content = "COMMAND\nhead\\r\\c\\ner:\\c\\n\\r\\\\\n\n\0",