Add test case to verify zmq reconnect behaviour (#84) (#1052)

This PR is a partial duplicate of sonic-net/sonic-dash-ha#84
Since crates/swss-common will be moved in this repo.

why
During investigating issue #75, I created test case to verify zmq behaviour in handling no connection and connection loss. It is found that with the current swss-common zmq implementation with below attributes, we don't need to handle reconnect explicitly.

PUSH/PULL model
ZMQ_IMMEDIATE = 0 (default)
ZMQ_SNDHWM = 10000
what this PR does
add unit test for zmq late connect and reconnect for regression
remove unneeded code for falling back to ProducerStateTable in test environment since zmq client won't fail if zmq server is not connected.
update test_utils/README.md with missing argument to start redis in test environment

Author: qiluo-msft

crates/swss-common/tests/sync.rs

@@ -190,6 +190,47 @@ fn zmq_consumer_producer_state_tables_sync_api_basic_test() -> Result<(), Except
     Ok(())
 }
 
+// Below test covers 2 scenarios:
+// 1. late connect when zmq server is started after client sending messages.
+// 2. reconnect when zmq server is stopped and restarted. messages from client during
+//    the time should be queued by client and resent when server is restarted.
+#[test]
+fn zmq_consumer_producer_state_tables_sync_api_connect_late_reconnect() -> Result<(), Exception> {
+    use SelectResult::*;
+    enum TestPhase {
+        LateConnect,
+        Reconnect,
+    }
+    let (endpoint, _delete) = random_zmq_endpoint();
+
+    let zmqc = ZmqClient::new(&endpoint)?;
+    let redis = Redis::start();
+    let zpst = ZmqProducerStateTable::new(redis.db_connector(), "table_a", zmqc, false)?;
+
+    for _ in [TestPhase::LateConnect, TestPhase::Reconnect] {
+        let kfvs = random_kfvs();
+        for kfv in &kfvs {
+            match kfv.operation {
+                KeyOperation::Set => zpst.set(&kfv.key, kfv.field_values.clone())?,
+                KeyOperation::Del => zpst.del(&kfv.key)?,
+            }
+        }
+
+        let mut zmqs = ZmqServer::new(&endpoint)?;
+        let zcst = ZmqConsumerStateTable::new(redis.db_connector(), "table_a", &mut zmqs, None, None)?;
+        let mut kfvs_seen = Vec::new();
+        while kfvs_seen.len() != kfvs.len() {
+            assert_eq!(zcst.read_data(Duration::from_millis(2000), true)?, Data);
+            kfvs_seen.extend(zcst.pops()?);
+        }
+        assert_eq!(kfvs, kfvs_seen);
+        drop(zcst);
+        drop(zmqs);
+    }
+
+    Ok(())
+}
+
 #[test]
 fn table_sync_api_basic_test() -> Result<(), Exception> {
     let redis = Redis::start();