ws redis command improvements + test script

ws/test_ws_redis.sh

@@ -0,0 +1,25 @@
+#!/bin/sh
+
+# Handle Ctrl-C by killing all sub-processing AND exiting
+trap cleanup INT
+
+function cleanup {
+    kill `cat /tmp/pidfile.subscribe`
+    exit 1
+}
+
+REDIS_HOST=${REDIS_HOST:=localhost}
+
+ws redis_subscribe --pidfile /tmp/pidfile.subscribe --host $REDIS_HOST foo &
+
+# Wait for the subscriber to be ready
+sleep 0.5
+
+# Now publish messages
+ws redis_publish -c 100000 --host ${REDIS_HOST} foo bar
+
+# Wait a little for all messages to be received
+sleep 0.5
+
+# Cleanup
+cleanup

ws/ws.cpp

@@ -49,6 +49,7 @@ int main(int argc, char** argv)
     int transferTimeout = 1800;
     int maxRedirects = 5;
     int delayMs = -1;
+    int count = 1;
 
     CLI::App* sendApp = app.add_subcommand("send", "Send a file");
     sendApp->add_option("url", url, "Connection url")->required();
@@ -106,6 +107,7 @@ int main(int argc, char** argv)
     redisPublishApp->add_option("--password", password, "Password");
     redisPublishApp->add_option("channel", channel, "Channel")->required();
     redisPublishApp->add_option("message", message, "Message")->required();
+    redisPublishApp->add_option("-c", count, "Count");
 
     CLI::App* redisSubscribeApp = app.add_subcommand("redis_subscribe", "Redis subscriber");
     redisSubscribeApp->add_option("--port", redisPort, "Port");
@@ -113,6 +115,7 @@ int main(int argc, char** argv)
     redisSubscribeApp->add_option("--password", password, "Password");
     redisSubscribeApp->add_option("channel", channel, "Channel")->required();
     redisSubscribeApp->add_flag("-v", verbose, "Verbose");
+    redisSubscribeApp->add_option("--pidfile", pidfile, "Pid file");
 
     CLI11_PARSE(app, argc, argv);
 
@@ -169,7 +172,8 @@ int main(int argc, char** argv)
     }
     else if (app.got_subcommand("redis_publish"))
     {
-        return ix::ws_redis_publish_main(hostname, redisPort, password, channel, message);
+        return ix::ws_redis_publish_main(hostname, redisPort, password,
+                                         channel, message, count);
     }
     else if (app.got_subcommand("redis_subscribe"))
     {

ws/ws.h

@@ -44,7 +44,8 @@ namespace ix
                               int port,
                               const std::string& password,
                               const std::string& channel,
-                              const std::string& message);
+                              const std::string& message,
+                              int count);
 
     int ws_redis_subscribe_main(const std::string& hostname,
                                 int port,

ws/ws_redis_publish.cpp

@@ -14,7 +14,8 @@ namespace ix
                               int port,
                               const std::string& password,
                               const std::string& channel,
-                              const std::string& message)
+                              const std::string& message,
+                              int count)
     {
         RedisClient redisClient;
         if (!redisClient.connect(hostname, port))
@@ -35,12 +36,15 @@ namespace ix
             std::cout << "Auth response: " << authResponse << ":" << port << std::endl;
         }
 
-        std::cerr << "Publishing message " << message
-                  << " to " << channel << "..." << std::endl;
-        if (!redisClient.publish(channel, message))
+        for (int i = 0; i < count; i++)
         {
-            std::cerr << "Error publishing to channel " << channel << std::endl;
-            return 1;
+            //std::cerr << "Publishing message " << message
+            //          << " to " << channel << "..." << std::endl;
+            if (!redisClient.publish(channel, message))
+            {
+                std::cerr << "Error publishing to channel " << channel << std::endl;
+                return 1;
+            }
         }
 
         return 0;

ws/ws_redis_subscribe.cpp

@@ -7,6 +7,8 @@
 #include <iostream>
 #include <sstream>
 #include <chrono>
+#include <thread>
+#include <atomic>
 #include "IXRedisClient.h"
 
 namespace ix
@@ -36,11 +38,10 @@ namespace ix
             std::cout << "Auth response: " << authResponse << ":" << port << std::endl;
         }
 
-        std::chrono::time_point<std::chrono::steady_clock> lastTimePoint;
-        int msgPerSeconds = 0;
-        int msgCount = 0;
+        std::atomic<int> msgPerSeconds(0);
+        std::atomic<int> msgCount(0);
 
-        auto callback = [&lastTimePoint, &msgPerSeconds, &msgCount, verbose]
+        auto callback = [&msgPerSeconds, &msgCount, verbose]
                          (const std::string& message)
         {
             if (verbose)
@@ -49,21 +50,7 @@ namespace ix
             }
 
             msgPerSeconds++;
-
-            auto now = std::chrono::steady_clock::now();
-            if (now - lastTimePoint > std::chrono::seconds(1))
-            {
-                lastTimePoint = std::chrono::steady_clock::now();
-
-                msgCount += msgPerSeconds;
-
-                // #messages 901 msg/s 150
-                std::cout << "#messages " << msgCount << " "
-                          << "msg/s " << msgPerSeconds
-                          << std::endl;
-
-                msgPerSeconds = 0;
-            }
+            msgCount++;
         };
 
         auto responseCallback = [](const std::string& redisResponse)
@@ -71,6 +58,22 @@ namespace ix
             std::cout << "Redis subscribe response: " << redisResponse << std::endl;
         };
 
+        auto timer = [&msgPerSeconds, &msgCount]
+        {
+            while (true)
+            {
+                std::cout << "#messages " << msgCount << " "
+                          << "msg/s " << msgPerSeconds
+                          << std::endl;
+
+                msgPerSeconds = 0;
+                auto duration = std::chrono::seconds(1);
+                std::this_thread::sleep_for(duration);
+            }
+        };
+
+        std::thread t(timer);
+
         std::cerr << "Subscribing to " << channel << "..." << std::endl;
         if (!redisClient.subscribe(channel, responseCallback, callback))
         {