dev: zmq hwm rebind (#1480)

* move hwm into the zmq constructor and reconstruct the socket instead of rebind. seems to work better than rebind (most connections cant bind so fast?)

* fix after merge

* added tests to reconnect zmq sockets when setting rx zmqport and rx zmqhwm, changed the tests scripts a bit to make the receiver starting tcp port configurable

* tests: slsreceiver also starting up  with 2000 as default tcp port, using latest cli args for receiver and multi receiver

* releasr notes

---------

Co-authored-by: AliceMazzoleni99 <alice.mazzoleni@psi.ch>

slsDetectorSoftware/src/Detector.cpp

@@ -1096,28 +1096,35 @@ void Detector::setNumberofUDPInterfaces_(int n) {
     if (!size()) {
         throw RuntimeError("No modules added.");
     }
-    bool previouslyClientStreaming = pimpl->getDataStreamingToClient();
-    uint16_t clientStartingPort = getClientZmqPort({0}).squash(0);
+
+    // get starting ports and disable zmq streaming
+    // rx
     bool useReceiver = getUseReceiverFlag().squash(false);
     bool previouslyReceiverStreaming = false;
     uint16_t rxStartingPort = 0;
     if (useReceiver) {
         previouslyReceiverStreaming = getRxZmqDataStream().squash(true);
+        setRxZmqDataStream(false);
         rxStartingPort = getRxZmqPort({0}).squash(0);
     }
+    // client
+    bool previouslyClientStreaming = pimpl->getDataStreamingToClient();
+    uint16_t clientStartingPort = getClientZmqPort({0}).squash(0);
+    pimpl->setDataStreamingToClient(false);
+
     pimpl->Parallel(&Module::setNumberofUDPInterfaces, {}, n);
+
     // ensure receiver zmq socket ports are multiplied by 2 (2 interfaces)
     setClientZmqPort(clientStartingPort, -1);
-    if (getUseReceiverFlag().squash(false)) {
+    if (useReceiver) {
         setRxZmqPort(rxStartingPort, -1);
     }
+
     // redo the zmq sockets if enabled
     if (previouslyClientStreaming) {
-        pimpl->setDataStreamingToClient(false);
         pimpl->setDataStreamingToClient(true);
     }
-    if (previouslyReceiverStreaming) {
-        setRxZmqDataStream(false);
+    if (useReceiver && previouslyReceiverStreaming) {
         setRxZmqDataStream(true);
     }
 }
@@ -1638,7 +1645,12 @@ void Detector::setClientZmqIp(const IpAddr ip, Positions pos) {
 int Detector::getClientZmqHwm() const { return pimpl->getClientStreamingHwm(); }
 
 void Detector::setClientZmqHwm(const int limit) {
+    bool previouslyClientStreaming = pimpl->getDataStreamingToClient();
     pimpl->setClientStreamingHwm(limit);
+    if (previouslyClientStreaming) {
+        pimpl->setDataStreamingToClient(false);
+        pimpl->setDataStreamingToClient(true);
+    }
 }
 
 Result<int> Detector::getRxZmqHwm(Positions pos) const {

slsDetectorSoftware/src/DetectorImpl.cpp

@@ -470,22 +470,13 @@ void DetectorImpl::createReceivingDataSockets() {
     size_t numSockets = modules.size() * numUDPInterfaces;
 
     for (size_t iSocket = 0; iSocket < numSockets; ++iSocket) {
-        uint32_t portnum =
-            (modules[iSocket / numUDPInterfaces]->getClientStreamingPort());
+        auto imod = iSocket / numUDPInterfaces;
+        uint32_t portnum = modules[imod]->getClientStreamingPort();
         portnum += (iSocket % numUDPInterfaces);
         try {
+            auto ip = modules[imod]->getClientStreamingIP().str();
             zmqSocket.push_back(
-                make_unique<ZmqSocket>(modules[iSocket / numUDPInterfaces]
-                                           ->getClientStreamingIP()
-                                           .str()
-                                           .c_str(),
-                                       portnum));
-            // set high water mark
-            int hwm = shm()->zmqHwm;
-            if (hwm >= 0) {
-                zmqSocket[iSocket]->SetReceiveHighWaterMark(hwm);
-                // need not reconnect. cannot be connected (detector idle)
-            }
+                make_unique<ZmqSocket>(ip.c_str(), portnum, shm()->zmqHwm));
             LOG(logINFO) << "Zmq Client[" << iSocket << "] at "
                          << zmqSocket.back()->GetZmqServerAddress() << "[hwm: "
                          << zmqSocket.back()->GetReceiveHighWaterMark() << "]";
@@ -1064,23 +1055,7 @@ void DetectorImpl::setClientStreamingHwm(const int limit) {
     }
     // update shm
     shm()->zmqHwm = limit;
-
-    // streaming enabled
-    if (client_downstream) {
-        // custom limit, set it directly
-        if (limit >= 0) {
-            for (auto &it : zmqSocket) {
-                it->SetReceiveHighWaterMark(limit);
-                // need not reconnect. cannot be connected (detector idle)
-            }
-            LOG(logINFO) << "Setting Client Zmq socket rcv hwm to " << limit;
-        }
-        // default, disable and enable to get default
-        else {
-            setDataStreamingToClient(false);
-            setDataStreamingToClient(true);
-        }
-    }
+    LOG(logINFO) << "Setting Client Zmq socket rcv hwm to " << limit;
 }
 
 void DetectorImpl::registerAcquisitionFinishedCallback(void (*func)(double, int,