Fixing a bug in cps_recv, track the last viewed time.

Cps_recv had not been properly tracking the last viewed cycle, so every time it looked at the shared memory it saw what it thought was a new message and sent (and resent) the data that was present in the buffer was re-broadcasted (in this case written into the mbuf in same spot multiple times).

Added a test case to the message queue that was used in helping to narrow down issues.

src/pub_sub_stream/sub_recv_bufferred.cc

@@ -983,6 +983,7 @@ main( int argc, char** argv )
             usleep( 2000 );
             latest_in_buffer = circular_buffer.latest( );
         } while ( latest_in_buffer == last_received );
+        last_received = latest_in_buffer;
 
         usleep( delay_cycles );
 

src/pub_sub_stream/tests/test_message_queue.cc

@@ -119,6 +119,23 @@ TEST_CASE( "You can optionally time out a pop operation on a message_queue " )
     REQUIRE( val.operator bool( ) == false );
 }
 
+TEST_CASE( "An empty message queue blocks until data is present, if timeouts are not requested")
+{
+    Message_queue<int, 5> test_queue;
+
+    std::thread t([&test_queue]() {
+        std::this_thread::sleep_for(std::chrono::milliseconds(300));
+        test_queue.emplace(42);
+    });
+    auto start = std::chrono::steady_clock::now();
+    int val = test_queue.pop();
+    auto end = std::chrono::steady_clock::now();
+    REQUIRE(val == 42);
+    auto duration = end-start;
+    REQUIRE(duration > std::chrono::milliseconds(200));
+    t.join();
+}
+
 TEST_CASE(
     "You can push as much into a message queue as long as you take out things" )
 {