(cobra client) pass the message position to the subscription data callback

docs/CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [8.2.3] - 2020-03-13
+
+(cobra client) pass the message position to the subscription data callback
+
 ## [8.2.2] - 2020-03-12
 
 (openssl tls backend) Fix a hand in OpenSSL when using TLS v1.3 ... by disabling TLS v1.3

ixbots/ixbots/IXCobraToSentryBot.cpp

@@ -201,10 +201,10 @@ namespace ix
                 conn.subscribe(channel,
                                filter,
                                [&jsonWriter, verbose, &throttled, &receivedCount, &queueManager](
-                                   const Json::Value& msg) {
+                                   const Json::Value& msg, const std::string& position) {
                                    if (verbose)
                                    {
-                                       spdlog::info("Subscriber received message -> {}", jsonWriter.write(msg));
+                                       spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
                                    }
 
                                    // If we cannot send to sentry fast enough, drop the message

ixbots/ixbots/IXCobraToStatsdBot.cpp

@@ -178,10 +178,10 @@ namespace ix
                     conn.subscribe(channel,
                                    filter,
                                    [&jsonWriter, &queueManager, verbose, &receivedCount](
-                                       const Json::Value& msg) {
+                                       const Json::Value& msg, const std::string& position) {
                                        if (verbose)
                                        {
-                                           spdlog::info(jsonWriter.write(msg));
+                                           spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
                                        }
 
                                        receivedCount++;

ixcobra/ixcobra/IXCobraConnection.cpp

@@ -441,9 +441,12 @@ namespace ix
         if (!body.isMember("messages")) return false;
         Json::Value messages = body["messages"];
 
+        if (!body.isMember("position")) return false;
+        std::string position = body["position"].asString();
+
         for (auto&& msg : messages)
         {
-            cb->second(msg);
+            cb->second(msg, position);
         }
 
         return true;

ixcobra/ixcobra/IXCobraConnection.h

@@ -42,7 +42,7 @@ namespace ix
         CobraConnection_PublishMode_Batch = 1
     };
 
-    using SubscriptionCallback = std::function<void(const Json::Value&)>;
+    using SubscriptionCallback = std::function<void(const Json::Value&, const std::string&)>;
     using EventCallback = std::function<void(CobraConnectionEventType,
                                              const std::string&,
                                              const WebSocketHttpHeaders&,

ixsnake/ixsnake/IXSnakeProtocol.cpp

@@ -189,7 +189,7 @@ namespace snake
             nlohmann::json response = {
                 {"action", "rtm/subscription/data"},
                 {"id", id++},
-                {"body", {{"subscription_id", subscriptionId}, {"messages", {msg}}}}};
+                {"body", {{"subscription_id", subscriptionId}, {"position", "0-0"}, {"messages", {msg}}}}};
 
             ws->sendText(response.dump());
         };

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "8.2.2"
+#define IX_WEBSOCKET_VERSION "8.2.3"

test/IXCobraChatTest.cpp

@@ -122,7 +122,8 @@ namespace
     void CobraChat::subscribe(const std::string& channel)
     {
         std::string filter;
-        _conn.subscribe(channel, filter, [this](const Json::Value& msg) {
+        _conn.subscribe(
+            channel, filter, [this](const Json::Value& msg, const std::string& /*position*/) {
                 spdlog::info("receive {}", msg.toStyledString());
 
                 if (!msg.isObject()) return;

test/IXCobraMetricsPublisherTest.cpp

@@ -93,7 +93,8 @@ namespace
             {
                 log("Subscriber authenticated");
                 std::string filter;
-                conn.subscribe(CHANNEL, filter, [](const Json::Value& msg) {
+                conn.subscribe(
+                    CHANNEL, filter, [](const Json::Value& msg, const std::string& /*position*/) {
                         log(msg.toStyledString());
 
                         std::string id = msg["id"].asString();

ws/ws_cobra_metrics_to_redis.cpp

@@ -126,7 +126,7 @@ namespace ix
                     channel,
                     filter,
                     [&msgPerSeconds, &msgCount, &conditionVariableMutex, &condition, &queue](
-                        const Json::Value& msg) {
+                        const Json::Value& msg, const std::string& /*position*/) {
                         {
                             std::unique_lock<std::mutex> lock(conditionVariableMutex);
                             queue.push(msg);

ws/ws_cobra_subscribe.cpp

@@ -14,21 +14,28 @@
 
 namespace ix
 {
-    void writeToStdout(bool fluentd, Json::FastWriter& jsonWriter, const Json::Value& msg)
+    void writeToStdout(bool fluentd,
+                       Json::FastWriter& jsonWriter,
+                       const Json::Value& msg,
+                       const std::string& position)
     {
         Json::Value enveloppe;
         if (fluentd)
         {
             enveloppe["producer"] = "cobra";
             enveloppe["consumer"] = "fluentd";
-            enveloppe["message"] = msg;
+
+            Json::Value msgWithPosition(msg);
+            msgWithPosition["position"] = position;
+            enveloppe["message"] = msgWithPosition;
+
+            std::cout << jsonWriter.write(enveloppe);
         }
         else
         {
             enveloppe = msg;
+            std::cout << position << " " << jsonWriter.write(enveloppe);
         }
-
-        std::cout << jsonWriter.write(enveloppe);
     }
 
     int ws_cobra_subscribe_main(const ix::CobraConfig& config,
@@ -82,10 +89,10 @@ namespace ix
                     conn.subscribe(channel,
                                    filter,
                                    [&jsonWriter, &quiet, &msgPerSeconds, &msgCount, &fluentd](
-                                       const Json::Value& msg) {
+                                       const Json::Value& msg, const std::string& position) {
                                        if (!quiet)
                                        {
-                                           writeToStdout(fluentd, jsonWriter, msg);
+                                           writeToStdout(fluentd, jsonWriter, msg, position);
                                        }
 
                                        msgPerSeconds++;