improve topic payload templating

src/emqtt_bench.erl

@@ -568,7 +568,7 @@ connect(Parent, N, PubSub, Opts) ->
                , {publish_signal_mref, MRef}
                , {pub_start_wait, RandomPubWaitMS}
                | Opts],
-    TopicPayloadRend = render_payload(proplists:get_value(topics_payload, Opts), AllOpts0 ++ MqttOpts),
+    TopicPayloadRend = render_topic_payload(proplists:get_value(topics_payload, Opts), AllOpts0 ++ MqttOpts),
     AllOpts = replace_opts(AllOpts0, [{topics_payload, TopicPayloadRend}]),
     {ok, Client} = emqtt:start_link(MqttOpts1),
     ConnectFun = connect_fun(Opts),
@@ -590,7 +590,7 @@ connect(Parent, N, PubSub, Opts) ->
                          undefined when TopicPayloadRend == undefined ->
                             erlang:send_after(RandomPubWaitMS, self(), publish);
                          undefined ->
-                            maps:foreach(fun(TopicName,  #{name := TopicName, interval_ms := DelayMs}) ->
+                            maps:foreach(fun(TopicName,  #{interval_ms := DelayMs}) ->
                                                erlang:send_after(RandomPubWaitMS + DelayMs, self(), {publish, TopicName})
                                          end, TopicPayloadRend);
                          _ ->
@@ -850,7 +850,8 @@ publish(Client, Opts) ->
     end.
 
 
-publish_topic(Client, Topic, #{ name := Topic
+publish_topic(Client, Topic, #{ name := TopicRendered
+                              , is_retain := IsRetain
                               , qos := QoS
                               , inject_ts := TsUnit
                               , payload := PayloadTemplate
@@ -869,7 +870,7 @@ publish_topic(Client, Topic, #{ name := Topic
       end,
    update_publish_start_at(Topic),
    case emqtt:publish_async(Client, via(LogicStream, #{priority => StreamPriority}),
-                          feed_var(Topic, ClientOpts), #{}, NewPayload, [{qos, QoS}],
+                          TopicRendered, #{}, NewPayload, [{qos, QoS}, {retain, IsRetain}],
                           infinity,
                           {fun(Caller, Res) ->
                                  Caller ! {publish_async_res, Res}
@@ -1473,6 +1474,7 @@ parse_topics_payload(Opts) ->
                             <<"payload_encoding">> := PayloadEncoding,
                             <<"payload">> := Payload} = Spec, Acc) ->
                            RFieldName = maps:get(<<"render">>, Spec, undefined),
+                           IsRetain = maps:get(<<"retain">>, Spec, false),
                            TSUnit = case WithTS of
                                        <<"s">> -> second;
                                        <<"ms">> -> millisecond;
@@ -1483,6 +1485,7 @@ parse_topics_payload(Opts) ->
                                     end,
                            Acc#{TopicName =>
                                    #{ name => TopicName
+                                    , is_retain => IsRetain
                                     , interval_ms => binary_to_integer(IntervalMS)
                                     , inject_ts => TSUnit
                                     , qos => QoS
@@ -1495,15 +1498,18 @@ parse_topics_payload(Opts) ->
                      end, #{} ,TopicSpecs)
    end.
 
--spec render_payload(TopicsPayload :: undefined | map(), Opts :: proplists:proplist()) -> NewTopicPayload :: undefined | map().
-render_payload(undefined, _Opts) ->
+-spec render_topic_payload(TopicsPayload :: undefined | map(), Opts :: proplists:proplist()) -> NewTopicPayload :: undefined | map().
+render_topic_payload(undefined, _Opts) ->
    undefined;
-render_payload(TopicsMap, Opts) when is_map(TopicsMap) ->
-   maps:map(fun(_K, V) -> do_render_payload(V, Opts) end, TopicsMap).
-
-do_render_payload(#{render_field := undefined} = Spec, _Opts) ->
-   Spec;
-do_render_payload(#{payload := Payload, render_field := FieldName} = Spec, Opts) when is_binary(FieldName) ->
+render_topic_payload(TopicsMap, Opts) when is_map(TopicsMap) ->
+   maps:map(fun(_K, V) ->
+                    do_render_topic_payload(V, Opts)
+            end, TopicsMap).
+
+do_render_topic_payload(#{render_field := undefined, name := TopicName} = Spec, Opts) ->
+   Spec#{name := feed_var(TopicName, Opts)};
+do_render_topic_payload(#{name := TopicName, payload := Payload, render_field := FieldName} = Spec, Opts)
+  when is_binary(FieldName) ->
    Template = maps:get(FieldName, Payload, ""),
    SRs = [ {<<"%i">>, integer_to_binary(proplists:get_value(seq, Opts))}
          , {<<"%c">>, proplists:get_value(client_id, Opts)}
@@ -1515,7 +1521,7 @@ do_render_payload(#{payload := Payload, render_field := FieldName} = Spec, Opts)
                            ({Search, Replace}, Acc) ->
                               binary:replace(Acc, Search, Replace, [global])
                         end, Template, SRs),
-   Spec#{payload := Payload#{FieldName := NewVal}}.
+   Spec#{name := feed_var(TopicName, Opts), payload := Payload#{FieldName := NewVal}}.
 
 %% @doc shared_paddings, utilize binary ref for shallow copy
 shared_paddings(Pages) when is_binary(Pages) ->

topic_spec.json

@@ -1,23 +1,24 @@
 {
     "topics" : [
-        {"name": "Topic1",
+        {"name": "Topic1/%i",
+         "retain": true,
          "interval_ms": "100",
          "inject_timestamp" : true,
          "QoS": 1,
          "payload_encoding": "eterm",
          "payload": {"foo1" : "bar1", "timestamp": "0", "Data": "%p10"},
-         "stream" : 1,
+         "stream" : 0,
          "stream_priority": 200,
          "render": "Data",
         },
 
-        {"name": "Topic2",
+        {"name": "Topic2/%i",
          "interval_ms": "5000",
          "inject_timestamp" : "ms",
          "QoS": 1,
          "payload_encoding": "eterm",
          "payload": {"foo2" : "bar2", "timestamp": "0", "VIN" : "VIN", "Data": "%p2500"},
-         "stream" : 2,
+         "stream" : 0,
          "stream_priority": 8,
          "render": "Data",
         }

topic_spec.md

@@ -9,9 +9,15 @@ It defines topics and their associated properties.
 ### topics array:
 
 Contains a list of topic definitions.
-name string:
+
+### name string:
 
 The unique identifier or name of the MQTT topic.
+support templating with %i, %c, %u
+
+### retain boolean:
+
+If true, publishs message with retain flag set.
 
 ### interval_ms string:
 
@@ -47,7 +53,9 @@ Other fields (e.g., foo1, foo2, VIN) might have specific meanings depending on t
 
 ### stream integer:
 
-A logical stream identifier,
+A logical stream identifier.
+
+Must be 0 for None QUIC transport.
 
 ### stream_priority integer: