Use translation for retained messages.

Author: rsandbach

src/rabbit_mqtt_processor.erl

@@ -334,7 +334,7 @@ maybe_send_retained_message(RPid, #mqtt_topic{name = S, qos = SubscribeQos}, Msg
                     retain = Msg#mqtt_msg.retain
                  }, variable = #mqtt_frame_publish{
                     message_id = Id,
-                    topic_name = S
+                    topic_name = rabbit_mqtt_util:amqp2mqtt(S)
                  },
                  payload = Msg#mqtt_msg.payload}, PState),
                  case Qos of

src/rabbit_mqtt_retainer.erl

@@ -42,16 +42,16 @@ start_link(RetainStoreMod, VHost) ->
     gen_server2:start_link(?MODULE, [RetainStoreMod, VHost], []).
 
 retain(Pid, Topic, Msg = #mqtt_msg{retain = true}) ->
-    gen_server2:cast(Pid, {retain, Topic, Msg});
+    gen_server2:cast(Pid, {retain, rabbit_mqtt_util:amqp2mqtt(Topic), Msg});
 
 retain(_Pid, _Topic, Msg = #mqtt_msg{retain = false}) ->
     throw({error, {retain_is_false, Msg}}).
 
 fetch(Pid, Topic) ->
-    gen_server2:call(Pid, {fetch, Topic}, ?TIMEOUT).
+    gen_server2:call(Pid, {fetch, rabbit_mqtt_util:amqp2mqtt(Topic)}, ?TIMEOUT).
 
 clear(Pid, Topic) ->
-    gen_server2:cast(Pid, {clear, Topic}).
+    gen_server2:cast(Pid, {clear, rabbit_mqtt_util:amqp2mqtt(Topic)}).
 
 %%----------------------------------------------------------------------------