# HG changeset patch
# User Daniel White <daniel@whitehouse.id.au>
# Date 1363103633 -39600
# Node ID e8d394af7ba341096e6643cd53ea022cbde55fd6
# Parent  34fd24fa456bfec70493b4f760c6a9c0980fc19e
base64 encode correlation_id in rpc requests

This prevents the rpc client from sending an invalid correlation_id
when the counter reaches 128.  The 0.9.1 AMQP spec defines the
property as a "short string" which requires it to be valid UTF-8.

The encoded identifier is also used as the dictionary key for the
continunation to avoid decoding the correlation id when handling the
response.

diff -r 34fd24fa456b -r e8d394af7ba3 src/amqp_rpc_client.erl
--- a/src/amqp_rpc_client.erl	Thu Mar 07 15:04:46 2013 +0000
+++ b/src/amqp_rpc_client.erl	Wed Mar 13 02:53:53 2013 +1100
@@ -93,7 +93,8 @@
                        routing_key = RoutingKey,
                        correlation_id = CorrelationId,
                        continuations = Continuations}) ->
-    Props = #'P_basic'{correlation_id = <<CorrelationId:64>>,
+    EncodedCorrelationId = base64:encode(<<CorrelationId:64>>),
+    Props = #'P_basic'{correlation_id = EncodedCorrelationId,
                        content_type = <<"application/octet-stream">>,
                        reply_to = Q},
     Publish = #'basic.publish'{exchange = X,
@@ -102,7 +103,7 @@
     amqp_channel:call(Channel, Publish, #amqp_msg{props = Props,
                                                   payload = Payload}),
     State#state{correlation_id = CorrelationId + 1,
-                continuations = dict:store(CorrelationId, From, Continuations)}.
+                continuations = dict:store(EncodedCorrelationId, From, Continuations)}.
 
 %%--------------------------------------------------------------------------
 %% gen_server callbacks
@@ -158,7 +159,7 @@
 
 %% @private
 handle_info({#'basic.deliver'{delivery_tag = DeliveryTag},
-             #amqp_msg{props = #'P_basic'{correlation_id = <<Id:64>>},
+             #amqp_msg{props = #'P_basic'{correlation_id = Id},
                        payload = Payload}},
             State = #state{continuations = Conts, channel = Channel}) ->
     From = dict:fetch(Id, Conts),
diff -r 34fd24fa456b -r e8d394af7ba3 test/amqp_client_SUITE.erl
--- a/test/amqp_client_SUITE.erl	Thu Mar 07 15:04:46 2013 +0000
+++ b/test/amqp_client_SUITE.erl	Wed Mar 13 02:53:53 2013 +1100
@@ -62,6 +62,7 @@
 sync_async_method_serialization_test_() -> ?RUN([]).
 teardown_test_()                        -> ?RUN([repeat]).
 rpc_test_()                             -> ?RUN([]).
+rpc_client_test_()                      -> ?RUN([]).
 pub_and_close_test_()                   -> ?RUN([]).
 channel_tune_negotiation_test_()        -> ?RUN([]).
 confirm_test_()                         -> ?RUN([]).
diff -r 34fd24fa456b -r e8d394af7ba3 test/test_util.erl
--- a/test/test_util.erl	Thu Mar 07 15:04:46 2013 +0000
+++ b/test/test_util.erl	Wed Mar 13 02:53:53 2013 +1100
@@ -973,6 +973,65 @@
     wait_for_death(Connection),
     ok.
 
+%% This tests if the RPC continues to generate valid correlation ids
+%% over a series of requests.
+rpc_client_test() ->
+    {ok, Connection} = new_connection(),
+    {ok, Channel} = amqp_connection:open_channel(Connection),
+    Q = uuid(),
+    Latch = 255, % enough requests to tickle bad correlation ids
+    %% Start a server to return correlation ids to the client.
+    Server = spawn_link(fun() ->
+                                rpc_correlation_server(Channel, Q)
+                        end),
+    %% Generate a series of RPC requests on the same client.
+    Client = amqp_rpc_client:start(Connection, Q),
+    Parent = self(),
+    [spawn(fun() ->
+                   Reply = amqp_rpc_client:call(Client, <<>>),
+                   Parent ! {finished, Reply}
+           end) || _ <- lists:seq(1, Latch)],
+    %% Verify that the correlation ids are valid UTF-8 strings.
+    CorrelationIds = latch_loop(Latch),
+    [?assertMatch(<<_/binary>>, DecodedId)
+     || DecodedId <- [unicode:characters_to_binary(Id, utf8)
+                      || Id <- CorrelationIds]],
+    %% Cleanup.
+    Server ! stop,
+    amqp_rpc_client:stop(Client),
+    teardown(Connection, Channel),
+    ok.
+
+%% Consumer of RPC requests that replies with the CorrelationId.
+rpc_correlation_server(Channel, Q) ->
+    amqp_channel:register_return_handler(Channel, self()),
+    amqp_channel:call(Channel, #'queue.declare'{queue = Q}),
+    amqp_channel:call(Channel, #'basic.consume'{queue = Q}),
+    rpc_client_consume_loop(Channel),
+    amqp_channel:unregister_return_handler(Channel).
+
+rpc_client_consume_loop(Channel) ->
+    receive
+        stop ->
+            ok;
+        {#'basic.deliver'{delivery_tag = DeliveryTag},
+         #amqp_msg{props = Props}} ->
+            #'P_basic'{correlation_id = CorrelationId,
+                       reply_to = Q} = Props,
+            Properties = #'P_basic'{correlation_id = CorrelationId},
+            Publish = #'basic.publish'{exchange = <<>>,
+                                       routing_key = Q,
+                                       mandatory = true},
+            amqp_channel:call(Channel, Publish, #amqp_msg{props = Properties,
+                                                          payload = CorrelationId}),
+            amqp_channel:call(Channel, #'basic.ack'{delivery_tag = DeliveryTag}),
+            rpc_client_consume_loop(Channel);
+        _ ->
+            rpc_client_consume_loop(Channel)
+    after 3000 ->
+            exit(no_request_received)
+    end.
+
 %%---------------------------------------------------------------------------
 
 setup_publish(Channel) ->
