Merge branch 'developer' into dev/fix-pattern-generator

.github/workflows/cmake.yaml

@@ -1,6 +1,6 @@
 name: Native CMake Build
 
-on: [push, pull_request] 
+on: [push, pull_request, workflow_dispatch] 
 
 env:
   # Customize the CMake build type here (Release, Debug, RelWithDebInfo, etc.)

RELEASE.md

@@ -71,6 +71,8 @@ allow disabling one UDP interface in the receiver.
 
 setting number of UDP interfaces can only be set at detector level and not at module level. (individual modules)
 
+zmqsocket - allowing hwm to be set already at the constructor level. As before, can still be set after construction with socket reconnection.
+
 2  On-board Detector Server Compatibility
 ==========================================
 

python/tests/test_free.py

@@ -21,7 +21,7 @@ from slsdet import Detector, Ctb, freeSharedMemory
 from utils_for_test import (
     Log,
     LogLevel,
-    SERVER_START_PORTNO
+    DET_START_TCP_PORTNO
 )
 
 from conftest import session_simulator
@@ -33,7 +33,7 @@ def test_exptime_after_free_should_raise(session_simulator):
 
 
     d = Ctb() # creates multi shm (assuming no shm exists)
-    d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it
+    d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it
 
     d.free() # frees the shm, d should not map to it anymore
 
@@ -47,7 +47,7 @@ def test_exptime_after_free_should_raise(session_simulator):
 
 def free_and_create_shm():
     k = Ctb() # opens existing shm if it exists
-    k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct
+    k.hostname = f"localhost:{DET_START_TCP_PORTNO}" # free and recreate shm, maps to local shm struct
 
 @pytest.mark.detectorintegration
 @pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
@@ -56,7 +56,7 @@ def test_exptime_after_not_passing_var_should_raise(session_simulator):
 
 
     d = Ctb() # creates multi shm (assuming no shm exists)
-    d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it
+    d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it
 
     free_and_create_shm() # ctb() opens multi shm, hostname command frees and recreates mod shm but shm struct is local. d still maps to old shm struct
 
@@ -72,7 +72,7 @@ def test_exptime_after_not_passing_var_should_raise(session_simulator):
 
 def free_and_create_shm_passing_ctb_var(k):
     k = Ctb() # opens existing shm if it exists (disregards k as its new Ctb only local to this function)
-    k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct
+    k.hostname = f"localhost:{DET_START_TCP_PORTNO}" # free and recreate shm, maps to local shm struct
 
 @pytest.mark.detectorintegration
 @pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
@@ -80,7 +80,7 @@ def test_exptime_after_passing_ctb_var_should_raise(session_simulator):
     Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_passing_ctb_var_should_raise')
 
     d = Ctb() # creates multi shm (assuming no shm exists)
-    d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it
+    d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it
 
     free_and_create_shm_passing_ctb_var(d) # ctb() opens multi shm, hostname command frees and recreates mod shm but shm struct is local. d still maps to old shm struct
 
@@ -95,7 +95,7 @@ def test_exptime_after_passing_ctb_var_should_raise(session_simulator):
 
 def free_and_create_shm_returning_ctb():
     k = Ctb() # opens existing shm if it exists (disregards k as its new Ctb only local to this function)
-    k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct
+    k.hostname = f"localhost:{DET_START_TCP_PORTNO}" # free and recreate shm, maps to local shm struct
     return k
 
 @pytest.mark.detectorintegration
@@ -128,8 +128,8 @@ def test_hostname_twice_acess_old_should_raise(session_simulator):
     Log(LogLevel.INFOBLUE, f'\nRunning test_hostname_twice_acess_old_should_raise')
 
     d = Ctb() # creates multi shm (assuming no shm exists)
-    d.hostname = f"localhost:{SERVER_START_PORTNO}" # hostname command creates mod shm, d maps to it
-    d.hostname = f"localhost:{SERVER_START_PORTNO}"  # Freeing and recreating shm while mapping d to it (old shm is out of scope)
+    d.hostname = f"localhost:{DET_START_TCP_PORTNO}" # hostname command creates mod shm, d maps to it
+    d.hostname = f"localhost:{DET_START_TCP_PORTNO}"  # Freeing and recreating shm while mapping d to it (old shm is out of scope)
     
     # this should not throw 
     exptime_val = d.exptime

python/tests/test_zmq.py

@@ -0,0 +1,51 @@
+import pytest 
+import sys 
+import time
+
+from conftest import session_simulator
+
+from slsdet import Detector
+from slsdet.utils import element_if_equal
+
+from slsdet._slsdet import slsDetectorDefs
+from utils_for_test import (
+    Log,
+    LogLevel,
+)
+
+detectorType = slsDetectorDefs.detectorType
+
+@pytest.mark.detectorintegration
+@pytest.mark.parametrize(
+    "session_simulator",
+    [
+        ("eiger", 1, 20)
+    ],
+    indirect=True,
+)
+def test_zmq_reconnect(session_simulator, request):
+    """ Test changing zmq ports and zmq hwm with zmq sockets connected (reconnect). """
+    det_type, num_interfaces, num_mods, d = session_simulator
+    assert d is not None
+    assert d.type == detectorType.EIGER
+
+
+    d.rx_zmqstream = True
+    assert d.rx_zmqstream == True
+
+    port = 14000
+    hwm = 2
+
+    for _ in range(10):
+        d.rx_zmqport = port
+        d.rx_zmqhwm = hwm
+        Log(LogLevel.INFOGREEN, f"Set zmqport={port}, zmqhwm={hwm}")
+
+        port += 1000
+        hwm += 5
+        time.sleep(1)
+
+
+    Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
+
+

slsDetectorServers/ctbDetectorServer/Makefile

@@ -6,6 +6,7 @@ main_src = ../slsDetectorServer/src/
 support_lib = ../../slsSupportLib/include/
 det_lib = ../../slsDetectorSoftware/include/sls/
 md5_dir = ../../slsSupportLib/src/
+script_dir = ../../etc/
 
 CROSS  		= bfin-uclinux-
 CC    		= $(CROSS)gcc
@@ -28,7 +29,7 @@ boot: $(OBJS)
 version_name=APICTB
 version_path=slsDetectorServers/ctbDetectorServer
 versioning: 
-	cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
+	echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
 
 
 $(PROGS): $(OBJS)

slsDetectorServers/eigerDetectorServer/Makefile

@@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/
 main_src = ../slsDetectorServer/src/
 support_lib = ../../slsSupportLib/include/
 md5_dir = ../../slsSupportLib/src/
+script_dir = ../../etc/
 
 BLACKFIN_CC = bfin-uclinux-gcc
 CROSS		= powerpc-4xx-softfloat-
@@ -28,7 +29,7 @@ boot: $(OBJS)
 version_name=APIEIGER
 version_path=slsDetectorServers/eigerDetectorServer
 versioning: 
-	cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
+	echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
 
 
 $(PROGS):  $(OBJS)

slsDetectorServers/gotthard2DetectorServer/Makefile

@@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/
 main_src = ../slsDetectorServer/src/
 support_lib = ../../slsSupportLib/include/
 md5_dir = ../../slsSupportLib/src/
+script_dir = ../../etc/
 
 CROSS  		= nios2-buildroot-linux-gnu-
 CC    		= $(CROSS)gcc
@@ -27,7 +28,7 @@ boot: $(OBJS)
 version_name=APIGOTTHARD2
 version_path=slsDetectorServers/gotthard2DetectorServer
 versioning: 
-	cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
+	echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
 
 
 $(PROGS): $(OBJS)

slsDetectorServers/jungfrauDetectorServer/Makefile

@@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/
 main_src = ../slsDetectorServer/src/
 support_lib = ../../slsSupportLib/include/
 md5_dir = ../../slsSupportLib/src/
+script_dir = ../../etc/
 
 CROSS  		= bfin-uclinux-
 CC    		= $(CROSS)gcc
@@ -27,7 +28,7 @@ boot: $(OBJS)
 version_name=APIJUNGFRAU
 version_path=slsDetectorServers/jungfrauDetectorServer
 versioning: 
-	cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
+	echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
 
 
 $(PROGS): $(OBJS)

slsDetectorServers/moenchDetectorServer/Makefile

@@ -5,6 +5,7 @@ main_inc = ../slsDetectorServer/include/
 main_src = ../slsDetectorServer/src/
 support_lib = ../../slsSupportLib/include/
 md5_dir = ../../slsSupportLib/src/
+script_dir = ../../etc/
 
 CROSS  		= bfin-uclinux-
 CC    		= $(CROSS)gcc
@@ -27,7 +28,7 @@ boot: $(OBJS)
 version_name=APIMOENCH
 version_path=slsDetectorServers/moenchDetectorServer
 versioning: 
-	cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
+	echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
 
 
 $(PROGS): $(OBJS)

slsDetectorServers/mythen3DetectorServer/Makefile

@@ -6,6 +6,7 @@ main_src = ../slsDetectorServer/src/
 support_lib = ../../slsSupportLib/include/
 det_lib = ../../slsDetectorSoftware/include/sls/
 md5_dir = ../../slsSupportLib/src/
+script_dir = ../../etc/
 
 CROSS  		= nios2-buildroot-linux-gnu-
 CC    		= $(CROSS)gcc
@@ -28,7 +29,7 @@ boot: $(OBJS)
 version_name=APIMYTHEN3
 version_path=slsDetectorServers/mythen3DetectorServer
 versioning: 
-	cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
+	echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
 
 
 $(PROGS): $(OBJS)

slsDetectorServers/slsDetectorServer/src/slsDetectorServer_funcs.c

@@ -1179,12 +1179,20 @@ int processDACEnums(enum dacIndex ind, int val, bool mV) {
         if (val != GET_FLAG) {
             ret = setDAC(serverDacIndex, val, mV, mess);
             // handle if set by user individually
-            if (serverDacIndex == E_VCMP_LL || serverDacIndex == E_VCMP_LR ||
-                serverDacIndex == E_VCMP_RL || serverDacIndex == E_VCMP_RR ||
-                serverDacIndex == E_VRPREAMP || serverDacIndex == E_VCP) {
+            switch (serverDacIndex) {
+            case E_VCMP_LL:
+            case E_VCMP_LR:
+            case E_VCMP_RL:
+            case E_VCMP_RR:
+            case E_VTHRESHOLD:
+            case E_VRPREAMP:
+            case E_VCP:
                 setSettings(UNDEFINED, mess);
                 LOG(logERROR, ("Settings has been changed "
                                "to undefined (changed specific dacs)\n"));
+                break;
+            default:
+                break;
             }
         }
         // get

slsDetectorServers/xilinx_ctbDetectorServer/Makefile

@@ -6,6 +6,7 @@ main_src = ../slsDetectorServer/src/
 support_lib = ../../slsSupportLib/include/
 det_lib = ../../slsDetectorSoftware/include/sls/
 md5_dir = ../../slsSupportLib/src/
+script_dir = ../../etc/
 
 ifeq ($(shell uname -m),aarch64)
     # no cross compilation needed when on aarch64
@@ -39,7 +40,7 @@ boot: $(OBJS)
 version_name=APIXILINXCTB
 version_path=slsDetectorServers/xilinx_ctbDetectorServer
 versioning: 
-	cd ../../ && echo $(PWD) && echo `tput setaf 6; python etc/updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
+	echo `tput setaf 6; python $(script_dir)updateAPIVersion.py $(version_name) $(version_path);  tput sgr0;`
 
 
 $(PROGS): $(OBJS)

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,
@@ -1695,7 +1670,9 @@ void DetectorImpl::updateRxUDPDatastreamMetadata() {
         }
     }
 
-    modules[0]->updateRxUDPPortDisableMetadata(disable);
+    if (modules[0]->getUseReceiverFlag()) {
+        modules[0]->updateRxUDPPortDisableMetadata(disable);
+    }
 }
 
 std::vector<int> DetectorImpl::getRxDisabledUDPPortIndices() const {

slsDetectorSoftware/src/Module.cpp

@@ -196,6 +196,7 @@ void Module::setSettings(detectorSettings isettings) {
             "Cannot set settings for Eiger. Use threshold energy.");
     }
     sendToDetector<int>(F_SET_SETTINGS, isettings);
+    updateRxThresholdEnergyMetadata();
 }
 
 int Module::getThresholdEnergy() const {
@@ -414,9 +415,14 @@ void Module::setAllThresholdEnergy(std::array<int, 3> e_eV,
         throw RuntimeError("setThresholdEnergyAndSettings: Could not set "
                            "settings in Module");
     }
+}
 
+void Module::updateRxThresholdEnergyMetadata() {
     if (shm()->useReceiverFlag) {
-        sendToReceiver(F_RECEIVER_SET_ALL_THRESHOLD, e_eV, nullptr);
+        if (shm()->detType == MYTHEN3) {
+            auto e_eV = getAllThresholdEnergy();
+            sendToReceiver(F_RECEIVER_SET_ALL_THRESHOLD, e_eV, nullptr);
+        }
     }
 }
 
@@ -477,6 +483,7 @@ int Module::getAllTrimbits() const {
 
 void Module::setAllTrimbits(int val) {
     sendToDetector<int>(F_SET_ALL_TRIMBITS, val);
+    updateRxThresholdEnergyMetadata();
 }
 
 std::vector<int> Module::getTrimEn() const {
@@ -541,39 +548,15 @@ void Module::setSynchronization(const bool value) {
 }
 
 std::vector<int> Module::getBadChannels() const {
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_GET_BAD_CHANNELS);
-    client.setFnum(F_GET_BAD_CHANNELS);
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Detector " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
+    auto retval = sendToDetectorVarVector<int>(F_GET_BAD_CHANNELS);
+    for (size_t i = 0; i < retval.size(); ++i) {
+        LOG(logDEBUG1) << i << ":" << retval[i];
     }
-    // receive badchannels
-    auto nch = client.Receive<int>();
-    std::vector<int> badchannels(nch);
-    if (nch > 0) {
-        client.Receive(badchannels);
-        for (size_t i = 0; i < badchannels.size(); ++i) {
-            LOG(logDEBUG1) << i << ":" << badchannels[i];
-        }
-    }
-    return badchannels;
+    return retval;
 }
 
 void Module::setBadChannels(std::vector<int> list) {
-    auto nch = static_cast<int>(list.size());
-    LOG(logDEBUG1) << "Sending bad channels to detector, nch:" << nch;
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_SET_BAD_CHANNELS);
-    client.setFnum(F_SET_BAD_CHANNELS);
-    client.Send(nch);
-    if (nch > 0) {
-        client.Send(list);
-    }
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Detector " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    sendToDetectorVarVector(F_SET_BAD_CHANNELS, list);
 }
 
 int Module::getRow() const { return sendToDetector<int>(F_GET_ROW); }
@@ -808,6 +791,7 @@ void Module::setDAC(int val, dacIndex index, bool mV) {
     }
     int args[]{static_cast<int>(index), static_cast<int>(mV), val};
     sendToDetector<int>(F_SET_DAC, args);
+    updateRxThresholdEnergyMetadata();
 }
 
 bool Module::getPowerChip() const {
@@ -833,21 +817,17 @@ bool Module::isPowerEnabled(defs::powerIndex index) const {
 
 void Module::setPowerEnabled(const std::vector<defs::powerIndex> &indices,
                              bool enable) {
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_SET_POWER);
-    client.setFnum(F_SET_POWER);
-    int count = indices.size();
-    client.Send(count);
+    int count = static_cast<int>(indices.size());
     std::vector<int> indices_int(count);
-    for (size_t i = 0; i < indices.size(); ++i) {
+    for (int i = 0; i < count; ++i) {
         indices_int[i] = static_cast<int>(indices[i]);
     }
+    // sends a variable vector and an enable
+    auto client = createDetectorSocket();
+    client.sendCommand(F_SET_POWER, &count, sizeof(count));
     client.Send(indices_int);
     client.Send(static_cast<int>(enable));
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Detector " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    client.readReply(nullptr, 0);
 }
 
 int Module::getPowerADC(defs::powerIndex index) const {
@@ -1022,72 +1002,15 @@ double Module::getReceiverProgress() const {
 }
 
 std::vector<int64_t> Module::getFramesCaughtByReceiver() const {
-    // TODO!(Erik) Refactor
-    LOG(logDEBUG1) << "Getting frames caught";
-    if (shm()->useReceiverFlag) {
-        auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-        client.Send(F_GET_RECEIVER_FRAMES_CAUGHT);
-        client.setFnum(F_GET_RECEIVER_FRAMES_CAUGHT);
-        if (client.Receive<int>() == FAIL) {
-            throw ReceiverError(
-                "Receiver " + std::to_string(moduleIndex) +
-                " returned error: " + client.readErrorMessage());
-        } else {
-            auto nports = client.Receive<int>();
-            std::vector<int64_t> retval(nports);
-            client.Receive(retval);
-            LOG(logDEBUG1) << "Frames caught of Receiver" << moduleIndex << ": "
-                           << ToString(retval);
-            return retval;
-        }
-    }
-    throw RuntimeError("No receiver to get frames caught.");
+    return sendToReceiverVarVector<int64_t>(F_GET_RECEIVER_FRAMES_CAUGHT);
 }
 
 std::vector<int64_t> Module::getNumMissingPackets() const {
-    // TODO!(Erik) Refactor
-    LOG(logDEBUG1) << "Getting num missing packets";
-    if (shm()->useReceiverFlag) {
-        auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-        client.Send(F_GET_NUM_MISSING_PACKETS);
-        client.setFnum(F_GET_NUM_MISSING_PACKETS);
-        if (client.Receive<int>() == FAIL) {
-            throw ReceiverError(
-                "Receiver " + std::to_string(moduleIndex) +
-                " returned error: " + client.readErrorMessage());
-        } else {
-            auto nports = client.Receive<int>();
-            std::vector<int64_t> retval(nports);
-            client.Receive(retval);
-            LOG(logDEBUG1) << "Missing packets of Receiver" << moduleIndex
-                           << ": " << ToString(retval);
-            return retval;
-        }
-    }
-    throw RuntimeError("No receiver to get missing packets.");
+    return sendToReceiverVarVector<int64_t>(F_GET_NUM_MISSING_PACKETS);
 }
 
 std::vector<int64_t> Module::getReceiverCurrentFrameIndex() const {
-    // TODO!(Erik) Refactor
-    LOG(logDEBUG1) << "Getting frame index";
-    if (shm()->useReceiverFlag) {
-        auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-        client.Send(F_GET_RECEIVER_FRAME_INDEX);
-        client.setFnum(F_GET_RECEIVER_FRAME_INDEX);
-        if (client.Receive<int>() == FAIL) {
-            throw ReceiverError(
-                "Receiver " + std::to_string(moduleIndex) +
-                " returned error: " + client.readErrorMessage());
-        } else {
-            auto nports = client.Receive<int>();
-            std::vector<int64_t> retval(nports);
-            client.Receive(retval);
-            LOG(logDEBUG1) << "Frame index of Receiver" << moduleIndex << ": "
-                           << ToString(retval);
-            return retval;
-        }
-    }
-    throw RuntimeError("No receiver to get frame index.");
+    return sendToReceiverVarVector<int64_t>(F_GET_RECEIVER_FRAME_INDEX);
 }
 
 uint64_t Module::getNextFrameNumber() const {
@@ -1435,46 +1358,11 @@ void Module::setUDPDataStream(const portPosition port, const bool enable) {
 }
 
 void Module::updateRxUDPPortDisableMetadata(const std::vector<int> &disable) {
-    if (!shm()->useReceiverFlag) {
-        return;
-    }
-    LOG(logDEBUG) << "Updating UDP port disable metadata in Receiver 0";
-    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-
-    client.Send(F_RECEIVER_SET_UDP_PORT_DISABLE_META);
-    client.setFnum(F_RECEIVER_SET_UDP_PORT_DISABLE_META);
-
-    auto nports = static_cast<int>(disable.size());
-    client.Send(nports);
-    if (nports > 0) {
-        client.Send(disable);
-    }
-    if (client.Receive<int>() == FAIL) {
-        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    sendToReceiverVarVector(F_RECEIVER_SET_UDP_PORT_DISABLE_META, disable);
 }
 
 std::vector<int> Module::getRxUDPPortDisableMetadata() const {
-    if (!shm()->useReceiverFlag) {
-        throw RuntimeError("No receiver to get disabled udp port indices.");
-    }
-
-    LOG(logDEBUG) << "Getting UDP port disable metadata in Receiver 0";
-    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-
-    client.Send(F_RECEIVER_GET_UDP_PORT_DISABLE_META);
-    client.setFnum(F_RECEIVER_GET_UDP_PORT_DISABLE_META);
-    if (client.Receive<int>() == FAIL) {
-        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
-    auto nports = client.Receive<int>();
-    std::vector<int> retval(nports);
-    if (nports > 0) {
-        client.Receive(retval);
-    }
-    return retval;
+    return sendToReceiverVarVector<int>(F_RECEIVER_GET_UDP_PORT_DISABLE_META);
 }
 
 // Receiver Config
@@ -1638,93 +1526,39 @@ void Module::setRxArping(bool enable) {
 }
 
 std::vector<defs::ROI> Module::getRxROI() const {
-    LOG(logDEBUG1) << "Getting receiver ROI for Module " << moduleIndex;
-    // check number of ports
-    if (!shm()->useReceiverFlag) {
-        throw RuntimeError("No receiver to get ROI.");
-    }
-    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-    client.Send(F_RECEIVER_GET_RECEIVER_ROI);
-    client.setFnum(F_RECEIVER_GET_RECEIVER_ROI);
-    auto nPorts = client.Receive<int>();
-    std::vector<ROI> retval(nPorts);
-    if (nPorts > 0)
-        client.Receive(retval);
-    if (nPorts != shm()->numUDPInterfaces) {
+    auto retval = sendToReceiverVarVector<ROI>(F_RECEIVER_GET_RECEIVER_ROI);
+    auto nPorts = retval.size();
+    if (static_cast<int>(nPorts) != shm()->numUDPInterfaces) {
         throw RuntimeError(
             "Invalid number of rois: " + std::to_string(nPorts) +
             ". Expected: " + std::to_string(shm()->numUDPInterfaces));
     }
-    LOG(logDEBUG1) << "ROI of Receiver" << moduleIndex << ": "
-                   << ToString(retval);
     return retval;
 }
 
 void Module::setRxROI(const std::vector<defs::ROI> &portRois) {
-    LOG(logDEBUG) << "Sending to receiver " << moduleIndex
-                  << " [roi: " << ToString(portRois) << ']';
-    if (!shm()->useReceiverFlag) {
-        throw RuntimeError("No receiver to set ROI.");
-    }
     if ((int)portRois.size() != shm()->numUDPInterfaces) {
         throw RuntimeError(
             "Invalid number of ROIs: " + std::to_string(portRois.size()) +
             ". Expected: " + std::to_string(shm()->numUDPInterfaces));
     }
-    // check number of ports
-    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-    client.Send(F_RECEIVER_SET_RECEIVER_ROI);
-    client.setFnum(F_RECEIVER_SET_RECEIVER_ROI);
-    int size = static_cast<int>(portRois.size());
-    client.Send(size);
-    if (size > 0)
-        client.Send(portRois);
-    if (client.Receive<int>() == FAIL) {
-        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    sendToReceiverVarVector(F_RECEIVER_SET_RECEIVER_ROI, portRois);
 }
 
 std::vector<slsDetectorDefs::ROI> Module::getRxROIMetadata() const {
-    LOG(logDEBUG1) << "Getting receiver ROI metadata for Module "
-                   << moduleIndex;
-    // check number of ports
-    if (!shm()->useReceiverFlag) {
-        throw RuntimeError("No receiver to get ROI metadata.");
-    }
-    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-    client.Send(F_RECEIVER_GET_ROI_METADATA);
-    client.setFnum(F_RECEIVER_GET_ROI_METADATA);
-    auto size = client.Receive<int>();
-    std::vector<slsDetectorDefs::ROI> retval(size);
-    if (size > 0)
-        client.Receive(retval);
-    if (size == 0) {
-        throw RuntimeError("Invalid number of ROI metadata: " +
-                           std::to_string(size) + ". Min: 1.");
-    }
+    auto retval = sendToReceiverVarVector<ROI>(F_RECEIVER_GET_ROI_METADATA);
+    if (retval.size() == 0)
+        throw RuntimeError("Invalid number of ROI metadata: 0. Min: 1.");
     LOG(logDEBUG1) << "ROI metadata of Receiver: " << ToString(retval);
     return retval;
 }
 
 void Module::setRxROIMetadata(const std::vector<slsDetectorDefs::ROI> &args) {
-    LOG(logDEBUG) << "Sending to receiver " << moduleIndex
-                  << " [roi metadata: " << ToString(args) << ']';
-    auto receiver = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-    receiver.Send(F_RECEIVER_SET_RECEIVER_ROI_METADATA);
-    receiver.setFnum(F_RECEIVER_SET_RECEIVER_ROI_METADATA);
-    int size = static_cast<int>(args.size());
-    receiver.Send(size);
-    if (size > 0)
-        receiver.Send(args);
-    if (size < 1) {
+    if (args.size() < 1) {
         throw RuntimeError("Invalid number of ROI metadata: " +
-                           std::to_string(size) + ". Min: 1.");
-    }
-    if (receiver.Receive<int>() == FAIL) {
-        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
-                            " returned error: " + receiver.readErrorMessage());
+                           std::to_string(args.size()) + ". Min: 1.");
     }
+    sendToReceiverVarVector(F_RECEIVER_SET_RECEIVER_ROI_METADATA, args);
 }
 
 // File
@@ -1935,17 +1769,7 @@ void Module::setRateCorrection(int64_t t) {
 }
 
 void Module::sendReceiverRateCorrections(const std::vector<int64_t> &t) {
-    LOG(logDEBUG) << "Sending to receiver 0 [rate corrections: " << ToString(t)
-                  << ']';
-    auto receiver = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-    receiver.Send(F_SET_RECEIVER_RATE_CORRECT);
-    receiver.setFnum(F_SET_RECEIVER_RATE_CORRECT);
-    receiver.Send(static_cast<int>(t.size()));
-    receiver.Send(t);
-    if (receiver.Receive<int>() == FAIL) {
-        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
-                            " returned error: " + receiver.readErrorMessage());
-    }
+    sendToReceiverVarVector(F_SET_RECEIVER_RATE_CORRECT, t);
 }
 
 bool Module::getInterruptSubframe() const {
@@ -2189,51 +2013,42 @@ void Module::sendVetoPhoton(const int chipIndex,
                             const std::vector<int> &gainIndices,
                             const std::vector<int> &values) {
     const int nch = gainIndices.size();
-    if (gainIndices.size() != values.size()) {
-        throw RuntimeError("Number of Gain Indices and values do not match! "
-                           "Gain Indices size: " +
-                           std::to_string(gainIndices.size()) +
-                           ", values size: " + std::to_string(values.size()));
+    if (nch != static_cast<int>(values.size())) {
+        std::ostringstream oss;
+        oss << "Number of Gain Indices and values do not match! Gain Indices "
+               "size: "
+            << nch << ", values size: " << values.size();
+        throw RuntimeError(oss.str());
     }
     LOG(logDEBUG1) << "Sending veto photon/file to detector [chip:" << chipIndex
                    << ", nch:" << nch << "]";
-
     const int args[]{chipIndex, nch};
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_SET_VETO_PHOTON);
-    client.setFnum(F_SET_VETO_PHOTON);
-    client.Send(args);
+    auto client = createDetectorSocket();
+    client.sendCommand(F_SET_VETO_PHOTON, args, sizeof(args));
     client.Send(gainIndices);
     client.Send(values);
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Detector " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    client.readReply(nullptr, 0);
 }
 
 void Module::getVetoPhoton(const int chipIndex,
                            const std::string &fname) const {
     LOG(logDEBUG1) << "Getting veto photon [" << chipIndex << "]\n";
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_GET_VETO_PHOTON);
-    client.setFnum(F_GET_VETO_PHOTON);
-    client.Send(chipIndex);
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Detector " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
-
-    auto nch = client.Receive<int>();
-    if (nch != shm()->nChan.x) {
-        throw DetectorError("Could not get veto photon. Expected " +
-                            std::to_string(shm()->nChan.x) + " channels, got " +
-                            std::to_string(nch));
-    }
+    auto client = createDetectorSocket();
+    client.sendCommand(F_GET_VETO_PHOTON, &chipIndex, sizeof(chipIndex));
+    int nch = 0;
+    client.readReply(&nch, sizeof(nch));
     std::vector<int> gainIndices(nch);
     std::vector<int> values(nch);
     client.Receive(gainIndices);
     client.Receive(values);
 
+    if (nch != shm()->nChan.x) {
+        std::ostringstream oss;
+        oss << "Could not get veto photon. Expected " << shm()->nChan.x
+            << " channels, got " << nch;
+        throw DetectorError(oss.str());
+    }
+
     // save to file
     std::ofstream outfile(fname);
     if (!outfile) {
@@ -2313,10 +2128,10 @@ void Module::setVetoPhoton(const int chipIndex, const int numPhotons,
     }
     // check size
     if ((int)gainIndices.size() != shm()->nChan.x) {
-        throw RuntimeError("Could not set veto photon. Invalid number of "
-                           "entries in file. Expected " +
-                           std::to_string(shm()->nChan.x) + ", read " +
-                           std::to_string(gainIndices.size()));
+        std::ostringstream oss;
+        oss << "Could not set veto photon. Invalid number of entries in file. "
+            << "Expected " << shm()->nChan.x << ", read " << gainIndices.size();
+        throw RuntimeError(oss.str());
     }
 
     sendVetoPhoton(chipIndex, gainIndices, values);
@@ -2733,17 +2548,13 @@ std::string Module::getPatternFileName() const {
 }
 
 void Module::setPattern(const Pattern &pat, const std::string &fname) {
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_SET_PATTERN);
-    client.setFnum(F_SET_PATTERN);
-    client.Send(pat.data(), pat.size());
     char args[MAX_STR_LENGTH]{};
     strcpy_safe(args, fname.c_str());
+
+    auto client = createDetectorSocket();
+    client.sendCommand(F_SET_PATTERN, pat.data(), pat.size());
     client.Send(args);
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Detector " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    client.readReply(nullptr, 0);
 }
 
 Pattern Module::getPattern() {
@@ -2846,69 +2657,44 @@ void Module::startPattern() { sendToDetector(F_START_PATTERN); }
 // Json Header specific
 
 std::map<std::string, std::string> Module::getAdditionalJsonHeader() const {
-    // TODO, refactor this function with a more robust sending.
-    // Now assuming whitespace separated key value
-    if (!shm()->useReceiverFlag) {
-        throw RuntimeError("Set rx_hostname first to use receiver parameters "
-                           "(zmq json header)");
-    }
-    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-    client.Send(F_GET_ADDITIONAL_JSON_HEADER);
-    client.setFnum(F_GET_ADDITIONAL_JSON_HEADER);
-    if (client.Receive<int>() == FAIL) {
-        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    } else {
-        auto size = client.Receive<int>();
-        std::string buff(size, '\0');
-        std::map<std::string, std::string> retval;
-        if (size > 0) {
-            client.Receive(&buff[0], buff.size());
-            std::istringstream iss(buff);
-            std::string key, value;
-            while (iss >> key) {
-                iss >> value;
-                retval[key] = value;
-            }
-        }
-        LOG(logDEBUG) << "Getting additional json header " << ToString(retval);
-        return retval;
+    auto vec = sendToReceiverVarVector<char>(F_GET_ADDITIONAL_JSON_HEADER);
+
+    // convert vector to string
+    std::string longString(vec.begin(), vec.end());
+    // convert string (space separated) to map of key-value pairs
+    std::map<std::string, std::string> retval;
+    std::istringstream iss(longString);
+    std::string key, value;
+    while (iss >> key) {
+        iss >> value;
+        retval[key] = value;
     }
+    LOG(logDEBUG) << "Getting additional json header " << ToString(retval);
+    return retval;
 }
 
 void Module::setAdditionalJsonHeader(
     const std::map<std::string, std::string> &jsonHeader) {
-    if (!shm()->useReceiverFlag) {
-        throw RuntimeError("Set rx_hostname first to use receiver parameters "
-                           "(zmq json header)");
-    }
+    // validate
     for (auto &it : jsonHeader) {
-        if (it.first.empty() || it.first.length() > SHORT_STR_LENGTH ||
-            it.second.length() > SHORT_STR_LENGTH) {
-            throw RuntimeError(
-                it.first + " or " + it.second +
-                " pair has invalid size. "
-                "Key cannot be empty. Both can have max 20 characters");
+        auto key = it.first;
+        auto value = it.second;
+        if (key.empty() || key.length() > SHORT_STR_LENGTH ||
+            value.length() > SHORT_STR_LENGTH) {
+            throw RuntimeError(key + " or " + value +
+                               " pair has invalid size. Key cannot be empty. "
+                               "Both can have max 20 characters");
         }
     }
+    // convert map to string (space separated)
     std::ostringstream oss;
     for (auto &it : jsonHeader)
         oss << it.first << ' ' << it.second << ' ';
-    auto buff = oss.str();
-    const auto size = static_cast<int>(buff.size());
-    LOG(logDEBUG) << "Sending to receiver additional json header "
-                  << ToString(jsonHeader);
-    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
-    client.Send(F_SET_ADDITIONAL_JSON_HEADER);
-    client.setFnum(F_SET_ADDITIONAL_JSON_HEADER);
-    client.Send(size);
-    if (size > 0)
-        client.Send(&buff[0], buff.size());
+    auto longString = oss.str();
+    // convert string to vector<char> to send over socket (variable vector impl)
+    std::vector<char> args(longString.begin(), longString.end());
 
-    if (client.Receive<int>() == FAIL) {
-        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    sendToReceiverVarVector(F_SET_ADDITIONAL_JSON_HEADER, args);
 }
 
 std::string Module::getAdditionalJsonParameter(const std::string &key) const {
@@ -3090,23 +2876,12 @@ IpAddr Module::getLastClientIP() const {
 }
 
 std::string Module::executeCommand(const std::string &cmd) {
-    char arg[MAX_STR_LENGTH]{};
-    char retval[MAX_STR_LENGTH]{};
-    strcpy_safe(arg, cmd.c_str());
     LOG(logINFO) << "Module " << moduleIndex << " (" << shm()->hostname
                  << "): Sending command " << cmd;
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_EXEC_COMMAND);
-    client.setFnum(F_EXEC_COMMAND);
-    client.Send(arg);
-    if (client.Receive<int>() == FAIL) {
-        std::cout << '\n';
-        std::ostringstream os;
-        os << "Module " << moduleIndex << " (" << shm()->hostname << ")"
-           << " returned error: " << client.readErrorMessage();
-        throw DetectorError(os.str());
-    }
-    client.Receive(retval);
+    char args[MAX_STR_LENGTH]{};
+    char retval[MAX_STR_LENGTH]{};
+    strcpy_safe(args, cmd.c_str());
+    sendToDetector(F_EXEC_COMMAND, args, retval);
     LOG(logINFO) << "Module " << moduleIndex << " (" << shm()->hostname
                  << "): command executed";
     return retval;
@@ -3126,14 +2901,18 @@ int64_t Module::getMeasurementTime() const {
 
 // private
 
-void Module::checkArgs(const void *args, size_t args_size, void *retval,
-                       size_t retval_size) const {
+void Module::checkArgs(const void *args, size_t args_size) const {
     if (args == nullptr && args_size != 0)
         throw RuntimeError(
             "Passed nullptr as args to Send function but size is not 0");
     if (args != nullptr && args_size == 0)
         throw RuntimeError(
             "Passed size 0 to Send function but args is not nullptr");
+}
+
+void Module::checkArgs(const void *args, size_t args_size, void *retval,
+                       size_t retval_size) const {
+    checkArgs(args, args_size);
     if (retval == nullptr && retval_size != 0)
         throw RuntimeError(
             "Passed nullptr as retval to Send function but size is not 0");
@@ -3150,13 +2929,42 @@ void Module::checkArgs(const void *args, size_t args_size, void *retval,
     static_assert(!std::is_same<ARG, std::nullptr_t>::value,                   \
                   "nullptr_t type is incompatible with templated " DST);
 
+DetectorSocket Module::createDetectorSocket() const {
+    return DetectorSocket(shm()->hostname, shm()->controlPort);
+}
+
+template <typename Arg>
+void Module::sendToDetectorVarVector(int fnum,
+                                     const std::vector<Arg> &args) const {
+    LOG(logDEBUG1) << "Sending to Detector: ["
+                   << getFunctionNameFromEnum(static_cast<detFuncs>(fnum))
+                   << ", std::vector<" << typeid(Arg).name() << ">, nullptr, 0,"
+                   << "]";
+    auto client = createDetectorSocket();
+    client.sendCommandVariableSize(fnum, args, nullptr, 0);
+    client.close();
+}
+
+template <typename Ret>
+std::vector<Ret> Module::sendToDetectorVarVector(int fnum) const {
+    LOG(logDEBUG1) << "Sending to Detector: ["
+                   << getFunctionNameFromEnum(static_cast<detFuncs>(fnum))
+                   << ", nullptr, 0, std::vector<" << typeid(Ret).name()
+                   << ">]";
+    std::vector<Ret> retval;
+    auto client = createDetectorSocket();
+    client.sendCommandVariableSize(fnum, nullptr, 0, retval);
+    client.close();
+    return retval;
+}
+
 void Module::sendToDetector(int fnum, const void *args, size_t args_size,
                             void *retval, size_t retval_size) const {
     // This is the only function that actually sends data to the detector
     // the other versions use templates to deduce sizes and create
     // the return type
     checkArgs(args, args_size, retval, retval_size);
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
+    auto client = createDetectorSocket();
     client.sendCommandThenRead(fnum, args, args_size, retval, retval_size);
     client.close();
 }
@@ -3379,6 +3187,47 @@ Ret Module::sendToDetectorStop(int fnum, const Arg &args) {
 
 //-------------------------------------------------------------- sendToReceiver
 
+ReceiverSocket Module::createReceiverSocket() const {
+    return ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
+}
+
+template <typename Arg>
+void Module::sendToReceiverVarVector(int fnum,
+                                     const std::vector<Arg> &args) const {
+    LOG(logDEBUG1) << "Sending to Receiver: ["
+                   << getFunctionNameFromEnum(static_cast<detFuncs>(fnum))
+                   << ", std::vector<" << typeid(Arg).name() << ">, nullptr, 0,"
+                   << "]";
+    if (!shm()->useReceiverFlag) {
+        std::ostringstream oss;
+        oss << "Set rx_hostname first to use receiver parameters, ";
+        oss << getFunctionNameFromEnum(static_cast<detFuncs>(fnum));
+        throw RuntimeError(oss.str());
+    }
+    auto receiver = createReceiverSocket();
+    receiver.sendCommandVariableSize(fnum, args, nullptr, 0);
+    receiver.close();
+}
+
+template <typename Ret>
+std::vector<Ret> Module::sendToReceiverVarVector(int fnum) const {
+    LOG(logDEBUG1) << "Sending to Receiver: ["
+                   << getFunctionNameFromEnum(static_cast<detFuncs>(fnum))
+                   << ", nullptr, 0, std::vector<" << typeid(Ret).name()
+                   << ">]";
+    if (!shm()->useReceiverFlag) {
+        std::ostringstream oss;
+        oss << "Set rx_hostname first to use receiver parameters, ";
+        oss << getFunctionNameFromEnum(static_cast<detFuncs>(fnum));
+        throw RuntimeError(oss.str());
+    }
+    std::vector<Ret> retval;
+    auto receiver = createReceiverSocket();
+    receiver.sendCommandVariableSize(fnum, nullptr, 0, retval);
+    receiver.close();
+    return retval;
+}
+
 void Module::sendToReceiver(int fnum, const void *args, size_t args_size,
                             void *retval, size_t retval_size) const {
     // This is the only function that actually sends data to the receiver
@@ -3391,7 +3240,7 @@ void Module::sendToReceiver(int fnum, const void *args, size_t args_size,
         throw RuntimeError(oss.str());
     }
     checkArgs(args, args_size, retval, retval_size);
-    auto receiver = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
+    auto receiver = createReceiverSocket();
     receiver.sendCommandThenRead(fnum, args, args_size, retval, retval_size);
     receiver.close();
 }
@@ -3707,26 +3556,19 @@ void Module::setModule(sls_detector_module &module, bool trimbits) {
                    "these have been replaced with 0/200 or 2800/2400.";
         }
     }
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_SET_MODULE);
-    client.setFnum(F_SET_MODULE);
+    auto client = createDetectorSocket();
+    client.sendCommand(F_SET_MODULE, nullptr, 0);
     sendModule(&module, client);
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Module " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    client.readReply(nullptr, 0);
+    updateRxThresholdEnergyMetadata();
 }
 
 sls_detector_module Module::getModule() {
     LOG(logDEBUG1) << "Getting module";
     sls_detector_module module(shm()->detType);
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_GET_MODULE);
-    client.setFnum(F_GET_MODULE);
-    if (client.Receive<int>() == FAIL) {
-        throw DetectorError("Module " + std::to_string(moduleIndex) +
-                            " returned error: " + client.readErrorMessage());
-    }
+    auto client = createDetectorSocket();
+    client.sendCommand(F_GET_MODULE, nullptr, 0);
+    client.readReply(nullptr, 0);
     receiveModule(&module, client);
     return module;
 }
@@ -4063,7 +3905,7 @@ void Module::sendProgram(bool blackfin, std::vector<char> buffer,
                  << "): Sending " << functionType;
 
     // send fnum and filesize
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
+    auto client = createDetectorSocket();
     client.Send(functionEnum);
     uint64_t filesize = buffer.size();
     client.Send(filesize);
@@ -4187,47 +4029,27 @@ void Module::simulatingActivityinDetector(const std::string &functionType,
 
 std::vector<uint8_t> Module::readSpi(int chip_id, int register_id,
                                      int n_bytes) const {
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_SPI_READ);
-    client.setFnum(F_SPI_READ);
-    client.Send(chip_id);
-    client.Send(register_id);
-    client.Send(n_bytes);
+    auto client = createDetectorSocket();
+    int args[] = {chip_id, register_id, n_bytes};
+    client.sendCommand(F_SPI_READ, &args, sizeof(args));
 
-    if (client.Receive<int>() == FAIL) {
-        std::ostringstream os;
-        os << "Module " << moduleIndex << " (" << shm()->hostname << ")"
-           << " returned error: " << client.readErrorMessage();
-        throw DetectorError(os.str());
-    }
-
-    std::vector<uint8_t> data(n_bytes);
-    client.Receive(data);
-    return data;
+    std::vector<uint8_t> retval(n_bytes);
+    client.readReply(retval.data(), retval.size());
+    return retval;
 }
 
 std::vector<uint8_t> Module::writeSpi(int chip_id, int register_id,
                                       const std::vector<uint8_t> &data) {
-    auto client = DetectorSocket(shm()->hostname, shm()->controlPort);
-    client.Send(F_SPI_WRITE);
-    client.setFnum(F_SPI_WRITE);
-    client.Send(chip_id);
-    client.Send(register_id);
-    client.Send(static_cast<int>(data.size()));
+    int count = static_cast<int>(data.size());
+    auto client = createDetectorSocket();
+    int args[] = {chip_id, register_id, count};
+    client.sendCommand(F_SPI_WRITE, &args, sizeof(args));
     client.Send(data);
-
-    if (client.Receive<int>() == FAIL) {
-        std::ostringstream os;
-        os << "Module " << moduleIndex << " (" << shm()->hostname << ")"
-           << " returned error: " << client.readErrorMessage();
-        throw DetectorError(os.str());
-    }
-
     // Read the output from the SPI write. This contains the data before the
-    // write.
-    std::vector<uint8_t> ret(data.size());
-    client.Receive(ret);
-    return ret;
+    // write
+    std::vector<uint8_t> retval(data.size());
+    client.readReply(retval.data(), retval.size());
+    return retval;
 }
 
 } // namespace sls

slsDetectorSoftware/src/Module.h

@@ -111,6 +111,7 @@ class Module : public virtual slsDetectorDefs {
     void setAllThresholdEnergy(std::array<int, 3> e_eV,
                                detectorSettings isettings, bool trimbits);
     std::string getSettingsDir() const;
+    void updateRxThresholdEnergyMetadata();
     std::string setSettingsDir(const std::string &dir);
     void loadTrimbits(const std::string &fname);
     void saveTrimbits(const std::string &fname);
@@ -626,9 +627,18 @@ class Module : public virtual slsDetectorDefs {
   private:
     std::string getReceiverLongVersion() const;
 
+    void checkArgs(const void *args, size_t args_size) const;
     void checkArgs(const void *args, size_t args_size, void *retval,
                    size_t retval_size) const;
 
+    DetectorSocket createDetectorSocket() const;
+
+    template <typename Arg>
+    void sendToDetectorVarVector(int fnum, const std::vector<Arg> &args) const;
+
+    template <typename Ret>
+    std::vector<Ret> sendToDetectorVarVector(int fnum) const;
+
     /**
      * Send function parameters to detector (control server)
      * @param fnum function enum
@@ -674,6 +684,9 @@ class Module : public virtual slsDetectorDefs {
     Ret sendToDetector(int fnum, const Arg &args) const;
 
     /** Send function parameters to detector (stop server) */
+
+    DetectorSocket createDetectorStopSocket() const;
+
     void sendToDetectorStop(int fnum, const void *args, size_t args_size,
                             void *retval, size_t retval_size);
 
@@ -713,6 +726,15 @@ class Module : public virtual slsDetectorDefs {
     Ret sendToDetectorStop(int fnum, const Arg &args) const;
 
     /** Send function parameters to receiver */
+
+    ReceiverSocket createReceiverSocket() const;
+
+    template <typename Arg>
+    void sendToReceiverVarVector(int fnum, const std::vector<Arg> &args) const;
+
+    template <typename Ret>
+    std::vector<Ret> sendToReceiverVarVector(int fnum) const;
+
     void sendToReceiver(int fnum, const void *args, size_t args_size,
                         void *retval, size_t retval_size);
 

slsDetectorSoftware/tests/Caller/test-Caller-rx-running.cpp

@@ -236,7 +236,6 @@ TEST_CASE("general - cant put if receiver is not idle",
             det.setAcquisitionIndex(prev_findex[i], {i});
             det.setFileWrite(prev_fwrite[i], {i});
             det.setRxFifoDepth(prev_fifodepth[i], {i});
-            det.setRxHostname(rx_hostname[i], {i});
         }
     }
 }

slsReceiverSoftware/src/ClientInterface.cpp

@@ -831,32 +831,17 @@ int ClientInterface::get_file_index(Interface &socket) {
 
 int ClientInterface::get_frame_index(Interface &socket) {
     auto retval = impl()->getCurrentFrameIndex();
-    LOG(logDEBUG1) << "frames index:" << ToString(retval);
-    auto size = static_cast<int>(retval.size());
-    socket.Send(OK);
-    socket.Send(size);
-    socket.Send(retval);
-    return OK;
+    return socket.sendVariableResult(retval);
 }
 
 int ClientInterface::get_missing_packets(Interface &socket) {
     auto missing_packets = impl()->getNumMissingPackets();
-    LOG(logDEBUG1) << "missing packets:" << ToString(missing_packets);
-    auto size = static_cast<int>(missing_packets.size());
-    socket.Send(OK);
-    socket.Send(size);
-    socket.Send(missing_packets);
-    return OK;
+    return socket.sendVariableResult(missing_packets);
 }
 
 int ClientInterface::get_frames_caught(Interface &socket) {
     auto retval = impl()->getFramesCaught();
-    LOG(logDEBUG1) << "frames caught:" << ToString(retval);
-    auto size = static_cast<int>(retval.size());
-    socket.Send(OK);
-    socket.Send(size);
-    socket.Send(retval);
-    return OK;
+    return socket.sendVariableResult(retval);
 }
 
 int ClientInterface::set_file_write(Interface &socket) {
@@ -1113,37 +1098,39 @@ int ClientInterface::stream_rx_dummy_header(Interface &socket) {
 }
 
 int ClientInterface::set_additional_json_header(Interface &socket) {
-    std::map<std::string, std::string> json;
-    auto size = socket.Receive<int>();
-    if (size > 0) {
-        std::string buff(size, '\0');
-        socket.Receive(&buff[0], buff.size());
-        std::istringstream iss(buff);
-        std::string key, value;
-        while (iss >> key) {
-            iss >> value;
-            json[key] = value;
-        }
-    }
     // verifyIdle(socket); allowing it to be set on the fly
-    LOG(logDEBUG1) << "Setting additional json header: " << ToString(json);
-    impl()->setAdditionalJsonHeader(json);
+
+    auto vec = socket.receiveVariableArgs<std::vector<char>>();
+    // convert vector<char> to string (space separated)
+    std::string longString(vec.begin(), vec.end());
+    // convert string to map of key-value pairs
+    std::map<std::string, std::string> args;
+    std::istringstream iss(longString);
+    std::string key, value;
+    while (iss >> key) {
+        iss >> value;
+        args[key] = value;
+    }
+    LOG(logDEBUG1) << "Setting additional json header: " << ToString(args);
+
+    impl()->setAdditionalJsonHeader(args);
     return socket.Send(OK);
 }
 
 int ClientInterface::get_additional_json_header(Interface &socket) {
     std::map<std::string, std::string> json = impl()->getAdditionalJsonHeader();
     LOG(logDEBUG1) << "additional json header:" << ToString(json);
+
+    // convert map to string (space separated) to send over socket
     std::ostringstream oss;
     for (auto &it : json) {
         oss << it.first << ' ' << it.second << ' ';
     }
-    auto buff = oss.str();
-    auto size = static_cast<int>(buff.size());
-    socket.sendResult(size);
-    if (size > 0)
-        socket.Send(buff);
-    return OK;
+    auto longString = oss.str();
+
+    // conver to vector<char> to send over socket (variable vector length impl)
+    std::vector<char> retval(longString.begin(), longString.end());
+    return socket.sendVariableResult(retval);
 }
 
 int ClientInterface::set_udp_socket_buffer_size(Interface &socket) {
@@ -1612,16 +1599,13 @@ int ClientInterface::set_streaming_start_fnum(Interface &socket) {
 }
 
 int ClientInterface::set_rate_correct(Interface &socket) {
-    auto index = socket.Receive<int>();
-    if (index <= 0) {
-        throw RuntimeError("Invalid number of rate correction values: " +
-                           std::to_string(index));
-    }
-    LOG(logDEBUG) << "Number of detectors for rate correction: " << index;
-    std::vector<int64_t> t(index);
-    socket.Receive(t);
     verifyIdle(socket);
-    LOG(logINFO) << "Setting rate corrections[" << index << ']';
+
+    auto t = socket.receiveVariableArgs<std::vector<int64_t>>();
+    if (t.size() <= 0)
+        throw RuntimeError("Invalid number of rate correction values: " +
+                           std::to_string(t.size()));
+    LOG(logINFO) << "Setting rate corrections[" << t.size() << ']';
     impl()->setRateCorrections(t);
     return socket.Send(OK);
 }
@@ -1748,58 +1732,41 @@ int ClientInterface::set_arping(Interface &socket) {
 int ClientInterface::get_receiver_roi(Interface &socket) {
     auto retvals = impl()->getPortROIs();
     LOG(logDEBUG1) << "Receiver roi retval:" << ToString(retvals);
-    auto size = static_cast<int>(retvals.size());
-    if (size != impl()->getNumberofUDPInterfaces()) {
-        throw RuntimeError("Invalid number of ROIs received: " +
-                           std::to_string(size) + ". Expected: " +
-                           std::to_string(impl()->getNumberofUDPInterfaces()));
-    }
-    socket.Send(size);
-    if (size > 0)
-        socket.Send(retvals);
-    return OK;
+    return socket.sendVariableResult(retvals);
 }
 
 int ClientInterface::set_receiver_roi(Interface &socket) {
-    auto roiSize = socket.Receive<int>();
-    std::vector<ROI> args(roiSize);
-    if (roiSize > 0) {
-        socket.Receive(args);
-    }
-    if (roiSize != impl()->getNumberofUDPInterfaces()) {
-        throw RuntimeError("Invalid number of ROIs received: " +
-                           std::to_string(roiSize) + ". Expected: " +
-                           std::to_string(impl()->getNumberofUDPInterfaces()));
-    }
     if (detType == CHIPTESTBOARD || detType == XILINX_CHIPTESTBOARD)
         functionNotImplemented();
-    LOG(logDEBUG1) << "Set Receiver ROI: " << ToString(args);
     verifyIdle(socket);
+
+    auto args = socket.receiveVariableArgs<std::vector<ROI>>();
+    auto numInterfaces = impl()->getNumberofUDPInterfaces();
+    if (static_cast<int>(args.size()) != numInterfaces) {
+        std::ostringstream oss;
+        oss << "Invalid number of ROIs received: " << args.size()
+            << ". Expected: " << numInterfaces;
+        throw RuntimeError(oss.str());
+    }
+    LOG(logDEBUG1) << "Set Receiver ROI: " << ToString(args);
     try {
         impl()->setPortROIs(args);
     } catch (const std::exception &e) {
         throw RuntimeError("Could not set Receiver ROI [" +
                            std::string(e.what()) + ']');
     }
-
     return socket.Send(OK);
 }
 
 int ClientInterface::set_receiver_roi_metadata(Interface &socket) {
-    auto roiSize = socket.Receive<int>();
-    LOG(logDEBUG1) << "Number of ReceiverROI metadata: " << roiSize;
-    if (roiSize < 1) {
-        throw RuntimeError("Invalid number of ROIs received: " +
-                           std::to_string(roiSize) + ". Min: 1.");
-    }
-    std::vector<ROI> rois(roiSize);
-    if (roiSize > 0) {
-        socket.Receive(rois);
-    }
     if (detType == CHIPTESTBOARD || detType == XILINX_CHIPTESTBOARD)
         functionNotImplemented();
     verifyIdle(socket);
-    LOG(logINFO) << "Setting ReceiverROI metadata[" << roiSize << ']';
+
+    auto rois = socket.receiveVariableArgs<std::vector<ROI>>();
+    if (rois.size() < 1)
+        throw RuntimeError("Invalid number of ROI metadata: " +
+                           std::to_string(rois.size()) + ". Min: 1.");
     try {
         impl()->setMultiROIMetadata(rois);
     } catch (const std::exception &e) {
@@ -1896,12 +1863,7 @@ int ClientInterface::get_roi_metadata(Interface &socket) {
     if (detType == CHIPTESTBOARD || detType == XILINX_CHIPTESTBOARD)
         functionNotImplemented();
     auto retvals = impl()->getMultiROIMetadata();
-    LOG(logDEBUG1) << "Receiver ROI metadata retval:" << ToString(retvals);
-    auto size = static_cast<int>(retvals.size());
-    socket.Send(size);
-    if (size > 0)
-        socket.Send(retvals);
-    return OK;
+    return socket.sendVariableResult(retvals);
 }
 
 int ClientInterface::set_readout_speed(Interface &socket) {
@@ -1934,16 +1896,9 @@ int ClientInterface::set_readout_speed(Interface &socket) {
 }
 
 int ClientInterface::set_udp_port_disable_meta(Interface &socket) {
-    auto nports = socket.Receive<int>();
-    std::vector<int> portsDisabled;
-    if (nports > 0) {
-        portsDisabled.resize(nports);
-        socket.Receive(portsDisabled);
-        LOG(logDEBUG1) << "Disabled Ports Metadata:" << ToString(portsDisabled);
-    }
     verifyIdle(socket);
+    auto portsDisabled = socket.receiveVariableArgs<std::vector<int>>();
     try {
-
         impl()->setUDPPortsDisabledMetadata(portsDisabled);
     } catch (const std::exception &e) {
         throw RuntimeError("Could not update UDP ports disabled metadata [" +
@@ -1954,14 +1909,7 @@ int ClientInterface::set_udp_port_disable_meta(Interface &socket) {
 
 int ClientInterface::get_udp_port_disable_meta(Interface &socket) {
     auto retvals = impl()->getUDPPortsDisabledMetadata();
-    LOG(logDEBUG1) << "Receiver disabled udp ports retval:"
-                   << ToString(retvals);
-    socket.Send(OK);
-    auto size = static_cast<int>(retvals.size());
-    socket.Send(size);
-    if (size > 0)
-        socket.Send(retvals);
-    return OK;
+    return socket.sendVariableResult(retvals);
 }
 
 } // namespace sls

slsReceiverSoftware/src/DataStreamer.cpp

@@ -78,16 +78,20 @@ void DataStreamer::RecordFirstIndex(uint64_t fnum, size_t firstImageIndex) {
                    << ", First Streamer Index:" << fnum;
 }
 
+int DataStreamer::GetZmqHwm() const {
+    if (zmqSocket) {
+        return zmqSocket->GetSendHighWaterMark();
+    }
+    return -1;
+}
+
 void DataStreamer::CreateZmqSockets(uint16_t port, int hwm) {
     uint16_t portnum = port + index;
     try {
-        zmqSocket = new ZmqSocket(portnum);
-
-        // set if custom
         if (hwm >= 0) {
-            zmqSocket->SetSendHighWaterMark(hwm);
-            // needed, or HWL is not taken
-            zmqSocket->Rebind();
+            zmqSocket = new ZmqSocket(portnum, hwm);
+        } else {
+            zmqSocket = new ZmqSocket(portnum);
         }
     } catch (std::exception &e) {
         std::ostringstream oss;

slsReceiverSoftware/src/DataStreamer.h

@@ -47,11 +47,13 @@ class DataStreamer : private virtual slsDetectorDefs, public ThreadObject {
      * Creates Zmq Sockets
      * (throws an exception if it couldnt create zmq sockets)
      * @param port streaming port start index
-     * @param hwm streaming high water mark
+     * @param hwm high water mark for zmq socket
      */
     void CreateZmqSockets(uint16_t port, int hwm);
     void CloseZmqSocket();
     void StreamRxDummyHeader();
+    void RestreamStop();
+    int GetZmqHwm() const;
 
   private:
     /**

slsReceiverSoftware/src/Implementation.cpp

@@ -1307,7 +1307,26 @@ void Implementation::setStreamingPort(const uint16_t i) {
     LOG(logINFO) << "Streaming Port: " << streamingPort;
 }
 
-int Implementation::getStreamingHwm() const { return streamingHwm; }
+int Implementation::getStreamingHwm() const {
+    switch (dataStreamer.size()) {
+    case 0:
+        return streamingHwm;
+    case 2:
+        if (dataStreamer[0]->GetZmqHwm() != dataStreamer[1]->GetZmqHwm()) {
+            throw RuntimeError(
+                "Streaming Hwm is not same for all data streamers: " +
+                std::to_string(dataStreamer[0]->GetZmqHwm()) + ", " +
+                std::to_string(dataStreamer[1]->GetZmqHwm()));
+        }
+        [[fallthrough]];
+    case 1:
+        return dataStreamer[0]->GetZmqHwm();
+        break;
+    default:
+        throw RuntimeError("Invalid number of data streamers: " +
+                           std::to_string(dataStreamer.size()));
+    }
+}
 
 void Implementation::setStreamingHwm(const int i) {
     streamingHwm = i;

slsSupportLib/include/sls/ClientSocket.h

@@ -14,13 +14,36 @@ class ClientSocket : public DataSocket {
     ClientSocket(std::string stype, const std::string &hostname,
                  uint16_t port_number);
     ClientSocket(std::string stype, struct sockaddr_in addr);
-    int sendCommandThenRead(int fnum, const void *args, size_t args_size,
-                            void *retval, size_t retval_size);
-
+    void sendCommandThenRead(int fnum, const void *args, size_t args_size,
+                             void *retval, size_t retval_size);
     std::string readErrorMessage();
 
+    template <typename Arg>
+    void sendCommandVariableSize(int fnum, const std::vector<Arg> &args,
+                                 void *retval, size_t retval_size) {
+        int count = args.size();
+        sendCommand(fnum, &count, sizeof(count));
+        if (count > 0) {
+            Send(args);
+        }
+        readReply(retval, retval_size);
+    }
+
+    template <typename Ret>
+    void sendCommandVariableSize(int fnum, const void *args, size_t args_size,
+                                 std::vector<Ret> &retval) {
+        sendCommand(fnum, args, args_size);
+        int count = 0;
+        readReply(&count, sizeof(count));
+        retval.resize(count);
+        if (count > 0) {
+            Receive(retval);
+        }
+    }
+    void sendCommand(int fnum, const void *args, size_t args_size);
+    void readReply(void *retval, size_t retval_size);
+
   private:
-    void readReply(int &ret, void *retval, size_t retval_size);
     [[noreturn]] void throwError(const std::string &msg) const;
     struct sockaddr_in serverAddr {};
     std::string socketType;

slsSupportLib/include/sls/ServerInterface.h

@@ -30,6 +30,25 @@ class ServerInterface : public DataSocket {
         Send(retval);
         return defs::OK;
     }
+
+    template <typename T> int sendVariableResult(T &&retval) {
+        int count = static_cast<int>(retval.size());
+        Send(defs::OK);
+        Send(count);
+        if (count > 0)
+            Send(retval);
+        return defs::OK;
+    }
+
+    template <typename T> T receiveVariableArgs() {
+        int count = 0;
+        Receive(count);
+        T retval(static_cast<typename T::size_type>(count));
+        if (count > 0) {
+            Receive(retval);
+        }
+        return retval;
+    }
 };
 
 } // namespace sls

slsSupportLib/include/sls/ZmqSocket.h

@@ -95,22 +95,23 @@ class ZmqSocket {
 
   public:
     // Socket Options for optimization
-    // ZMQ_LINGER default is already -1 means no messages discarded. use this
-    // options if optimizing required ZMQ_SNDHWM default is 0 means no limit.
-    // use this to optimize if optimizing required eg. int value = -1; if
-    // (zmq_setsockopt(socketDescriptor, ZMQ_LINGER, &value,sizeof(value))) {
-    //	Close();
+    // ZMQ_LINGER default is already -1 means no messages discarded.
+    // ZMQ_RCVHWM default is from zmqlib (1000). If not -1, calls
+    // SetReceiveHighWaterMark, also setting receive buffer size accordingly
     /** Constructor for a subscriber socket */
-    ZmqSocket(const char *const hostname_or_ip, const uint16_t portnumber);
+    ZmqSocket(const char *const hostname_or_ip, const uint16_t portnumber,
+              int hwm = -1);
 
-    /** Constructor for a publisher socket */
-    ZmqSocket(const uint16_t portnumber);
+    /** Constructor for a publisher socket with high water mark as -1 to mean
+     * default from zmqlib (1000). If hwm is not -1, it calls
+     * SetSendHighWaterMark, also setting send buffer size accordingly*/
+    ZmqSocket(const uint16_t portnumber, int hwm = -1);
 
     /** Returns high water mark for outbound messages */
     int GetSendHighWaterMark();
 
     /** Sets high water mark for outbound messages. Default 1000 (zmqlib). Also
-     * changes send buffer size depending on hwm. Must rebind.  */
+     * changes send buffer size depending on hwm. Must rebind or reconnect.  */
     void SetSendHighWaterMark(int limit);
 
     /** Returns high water mark for inbound messages */

slsSupportLib/include/sls/versionAPI.h

@@ -6,7 +6,7 @@
 #define APICTB        "0.0.0 0x260506"
 #define APIGOTTHARD2  "0.0.0 0x260427"
 #define APIMOENCH     "0.0.0 0x260424"
-#define APIEIGER      "0.0.0 0x260424"
+#define APIEIGER      "0.0.0 0x260807"
 #define APIXILINXCTB  "0.0.0 0x260506"
 #define APIJUNGFRAU   "0.0.0 0x260424"
 #define APIMYTHEN3    "0.0.0 0x260506"

slsSupportLib/src/ClientSocket.cpp

@@ -73,20 +73,23 @@ void ClientSocket::throwError(const std::string &msg) const {
     }
 }
 
-int ClientSocket::sendCommandThenRead(int fnum, const void *args,
-                                      size_t args_size, void *retval,
-                                      size_t retval_size) {
-    int ret = slsDetectorDefs::FAIL;
-    Send(&fnum, sizeof(fnum));
-    setFnum(fnum);
-    Send(args, args_size);
-    readReply(ret, retval, retval_size);
-    return ret;
+void ClientSocket::sendCommandThenRead(int fnum, const void *args,
+                                       size_t args_size, void *retval,
+                                       size_t retval_size) {
+    sendCommand(fnum, args, args_size);
+    readReply(retval, retval_size);
 }
 
-void ClientSocket::readReply(int &ret, void *retval, size_t retval_size) {
+void ClientSocket::sendCommand(int fnum, const void *args, size_t args_size) {
+    Send(fnum);
+    setFnum(fnum);
+    Send(args, args_size);
+}
+
+void ClientSocket::readReply(void *retval, size_t retval_size) {
 
     try {
+        int ret = slsDetectorDefs::FAIL;
         Receive(&ret, sizeof(ret));
         if (ret == slsDetectorDefs::FAIL) {
             std::string mess = readErrorMessage();

slsSupportLib/src/ZmqSocket.cpp

@@ -15,10 +15,11 @@ namespace sls {
 
 using namespace rapidjson;
 ZmqSocket::ZmqSocket(const char *const hostname_or_ip,
-                     const uint16_t portnumber)
+                     const uint16_t portnumber, int hwm)
     : portno(portnumber), sockfd(false) {
     // Extra check that throws if conversion fails, could be removed
     auto ipstr = HostnameToIp(hostname_or_ip).str();
+
     std::ostringstream oss;
     oss << "tcp://" << ipstr << ":" << portno;
     sockfd.serverAddress = oss.str();
@@ -27,20 +28,23 @@ ZmqSocket::ZmqSocket(const char *const hostname_or_ip,
     // create context
     sockfd.contextDescriptor = zmq_ctx_new();
     if (sockfd.contextDescriptor == nullptr)
-        throw ZmqSocketError("Could not create contextDescriptor");
+        throw ZmqSocketError("Could not create contextDescriptor for " +
+                             sockfd.serverAddress);
 
     // create subscriber
     sockfd.socketDescriptor = zmq_socket(sockfd.contextDescriptor, ZMQ_SUB);
     if (sockfd.socketDescriptor == nullptr) {
         PrintError();
-        throw ZmqSocketError("Could not create socket");
+        throw ZmqSocketError("Could not create socket for " +
+                             sockfd.serverAddress);
     }
 
     // Socket Options provided above
     // an empty string implies receiving any messages
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_SUBSCRIBE, "", 0)) {
         PrintError();
-        throw ZmqSocketError("Could set socket opt");
+        throw ZmqSocketError("Could set socket opt for " +
+                             sockfd.serverAddress);
     }
     // ZMQ_LINGER default is already -1 means no messages discarded. use this
     // options if optimizing required ZMQ_SNDHWM default is 0 means no limit.
@@ -49,7 +53,8 @@ ZmqSocket::ZmqSocket(const char *const hostname_or_ip,
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_LINGER, &value,
                        sizeof(value))) {
         PrintError();
-        throw ZmqSocketError("Could not set ZMQ_LINGER");
+        throw ZmqSocketError("Could not set ZMQ_LINGER for " +
+                             sockfd.serverAddress);
     }
     LOG(logDEBUG) << "Default receive high water mark:"
                   << GetReceiveHighWaterMark();
@@ -59,38 +64,47 @@ ZmqSocket::ZmqSocket(const char *const hostname_or_ip,
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_IPV6, &ipv6,
                        sizeof(ipv6))) {
         PrintError();
-        throw ZmqSocketError("Could not set ZMQ_IPV6");
+        throw ZmqSocketError("Could not set ZMQ_IPV6 for " +
+                             sockfd.serverAddress);
+    }
+
+    // set hwm if not default
+    if (hwm >= 0) {
+        SetReceiveHighWaterMark(hwm);
     }
 }
 
-ZmqSocket::ZmqSocket(const uint16_t portnumber)
+ZmqSocket::ZmqSocket(const uint16_t portnumber, int hwm)
     : portno(portnumber), sockfd(true) {
 
-    // create context
-    sockfd.contextDescriptor = zmq_ctx_new();
-    if (sockfd.contextDescriptor == nullptr)
-        throw ZmqSocketError("Could not create contextDescriptor");
-
-    // create publisher
-    sockfd.socketDescriptor = zmq_socket(sockfd.contextDescriptor, ZMQ_PUB);
-    if (sockfd.socketDescriptor == nullptr) {
-        PrintError();
-        throw ZmqSocketError("Could not create socket");
-    }
-    LOG(logDEBUG) << "Default send high water mark:" << GetSendHighWaterMark();
-
     // construct address, can be refactored with libfmt
     std::ostringstream oss;
     oss << "tcp://" << ZMQ_PUBLISHER_IP << ":" << portno;
     sockfd.serverAddress = oss.str();
     LOG(logDEBUG) << "zmq address: " << sockfd.serverAddress;
 
+    // create context
+    sockfd.contextDescriptor = zmq_ctx_new();
+    if (sockfd.contextDescriptor == nullptr)
+        throw ZmqSocketError("Could not create contextDescriptor for " +
+                             sockfd.serverAddress);
+
+    // create publisher
+    sockfd.socketDescriptor = zmq_socket(sockfd.contextDescriptor, ZMQ_PUB);
+    if (sockfd.socketDescriptor == nullptr) {
+        PrintError();
+        throw ZmqSocketError("Could not create socket for " +
+                             sockfd.serverAddress);
+    }
+    LOG(logDEBUG) << "Default send high water mark:" << GetSendHighWaterMark();
+
     // enable IPv6 addresses
     int ipv6 = 1;
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_IPV6, &ipv6,
                        sizeof(ipv6))) {
         PrintError();
-        throw ZmqSocketError("Could not set ZMQ_IPV6");
+        throw ZmqSocketError("Could not set ZMQ_IPV6 for " +
+                             sockfd.serverAddress);
     }
 
     // Socket Options for keepalive
@@ -99,34 +113,45 @@ ZmqSocket::ZmqSocket(const uint16_t portnumber)
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE, &keepalive,
                        sizeof(keepalive))) {
         PrintError();
-        throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE");
+        throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE for " +
+                             sockfd.serverAddress);
     }
     // set the number of keepalives before death
     keepalive = 10;
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_CNT,
                        &keepalive, sizeof(keepalive))) {
         PrintError();
-        throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_CNT");
+        throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_CNT for " +
+                             sockfd.serverAddress);
     }
     // set the time before the first keepalive
     keepalive = 60;
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_IDLE,
                        &keepalive, sizeof(keepalive))) {
         PrintError();
-        throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_IDLE");
+        throw ZmqSocketError(
+            "Could set socket opt ZMQ_TCP_KEEPALIVE_IDLE for " +
+            sockfd.serverAddress);
     }
     // set the interval between keepalives
     keepalive = 1;
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_TCP_KEEPALIVE_INTVL,
                        &keepalive, sizeof(keepalive))) {
         PrintError();
-        throw ZmqSocketError("Could set socket opt ZMQ_TCP_KEEPALIVE_INTVL");
+        throw ZmqSocketError(
+            "Could set socket opt ZMQ_TCP_KEEPALIVE_INTVL for " +
+            sockfd.serverAddress);
+    }
+    // set hwm if not default
+    if (hwm >= 0) {
+        SetSendHighWaterMark(hwm);
     }
 
     // bind address
     if (zmq_bind(sockfd.socketDescriptor, sockfd.serverAddress.c_str())) {
         PrintError();
-        throw ZmqSocketError("Could not bind socket");
+        throw ZmqSocketError("Could not bind socket for " +
+                             sockfd.serverAddress);
     }
     // sleep to allow a slow-joiner
     std::this_thread::sleep_for(std::chrono::milliseconds(200));
@@ -138,7 +163,8 @@ int ZmqSocket::GetSendHighWaterMark() {
     if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_SNDHWM, &value,
                        &value_size)) {
         PrintError();
-        throw ZmqSocketError("Could not get ZMQ_SNDHWM");
+        throw ZmqSocketError("Could not get ZMQ_SNDHWM for " +
+                             sockfd.serverAddress);
     }
     return value;
 }
@@ -147,10 +173,13 @@ void ZmqSocket::SetSendHighWaterMark(int limit) {
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_SNDHWM, &limit,
                        sizeof(limit))) {
         PrintError();
-        throw ZmqSocketError("Could not set ZMQ_SNDHWM");
+        throw ZmqSocketError("Could not set ZMQ_SNDHWM for " +
+                             sockfd.serverAddress + " to " +
+                             std::to_string(limit));
     }
     if (GetSendHighWaterMark() != limit) {
-        throw ZmqSocketError("Could not set ZMQ_SNDHWM to " +
+        throw ZmqSocketError("Could not set ZMQ_SNDHWM for " +
+                             sockfd.serverAddress + " to " +
                              std::to_string(limit));
     }
 
@@ -167,7 +196,8 @@ int ZmqSocket::GetReceiveHighWaterMark() {
     if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_RCVHWM, &value,
                        &value_size)) {
         PrintError();
-        throw ZmqSocketError("Could not get ZMQ_RCVHWM");
+        throw ZmqSocketError("Could not get ZMQ_RCVHWM for " +
+                             sockfd.serverAddress);
     }
     return value;
 }
@@ -176,10 +206,13 @@ void ZmqSocket::SetReceiveHighWaterMark(int limit) {
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_RCVHWM, &limit,
                        sizeof(limit))) {
         PrintError();
-        throw ZmqSocketError("Could not set ZMQ_RCVHWM");
+        throw ZmqSocketError("Could not set ZMQ_RCVHWM for " +
+                             sockfd.serverAddress + " to " +
+                             std::to_string(limit));
     }
     if (GetReceiveHighWaterMark() != limit) {
-        throw ZmqSocketError("Could not set ZMQ_RCVHWM to " +
+        throw ZmqSocketError("Could not set ZMQ_RCVHWM for " +
+                             sockfd.serverAddress + " to " +
                              std::to_string(limit));
     }
     int bufsize = DEFAULT_ZMQ_BUFFERSIZE;
@@ -195,7 +228,8 @@ int ZmqSocket::GetSendBuffer() {
     if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_SNDBUF, &value,
                        &value_size)) {
         PrintError();
-        throw ZmqSocketError("Could not get ZMQ_SNDBUF");
+        throw ZmqSocketError("Could not get ZMQ_SNDBUF for " +
+                             sockfd.serverAddress);
     }
     return value;
 }
@@ -204,10 +238,13 @@ void ZmqSocket::SetSendBuffer(int limit) {
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_SNDBUF, &limit,
                        sizeof(limit))) {
         PrintError();
-        throw ZmqSocketError("Could not set ZMQ_SNDBUF");
+        throw ZmqSocketError("Could not set ZMQ_SNDBUF for " +
+                             sockfd.serverAddress + " to " +
+                             std::to_string(limit));
     }
     if (GetSendBuffer() != limit) {
-        throw ZmqSocketError("Could not set ZMQ_SNDBUF to " +
+        throw ZmqSocketError("Could not set ZMQ_SNDBUF for " +
+                             sockfd.serverAddress + " to " +
                              std::to_string(limit));
     }
 }
@@ -218,7 +255,8 @@ int ZmqSocket::GetReceiveBuffer() {
     if (zmq_getsockopt(sockfd.socketDescriptor, ZMQ_RCVBUF, &value,
                        &value_size)) {
         PrintError();
-        throw ZmqSocketError("Could not get ZMQ_RCVBUF");
+        throw ZmqSocketError("Could not get ZMQ_RCVBUF for " +
+                             sockfd.serverAddress);
     }
     return value;
 }
@@ -227,10 +265,13 @@ void ZmqSocket::SetReceiveBuffer(int limit) {
     if (zmq_setsockopt(sockfd.socketDescriptor, ZMQ_RCVBUF, &limit,
                        sizeof(limit))) {
         PrintError();
-        throw ZmqSocketError("Could not set ZMQ_RCVBUF");
+        throw ZmqSocketError("Could not set ZMQ_RCVBUF for " +
+                             sockfd.serverAddress + " to " +
+                             std::to_string(limit));
     }
     if (GetReceiveBuffer() != limit) {
-        throw ZmqSocketError("Could not set ZMQ_RCVBUF to " +
+        throw ZmqSocketError("Could not set ZMQ_RCVBUF for " +
+                             sockfd.serverAddress + " to " +
                              std::to_string(limit));
     }
 }
@@ -242,12 +283,14 @@ void ZmqSocket::Rebind() {
     //    unbbind
     if (zmq_unbind(sockfd.socketDescriptor, sockfd.serverAddress.c_str())) {
         PrintError();
-        throw ZmqSocketError("Could not unbind socket");
+        throw ZmqSocketError("Could not unbind socket for " +
+                             sockfd.serverAddress);
     }
     // bind address
     if (zmq_bind(sockfd.socketDescriptor, sockfd.serverAddress.c_str())) {
         PrintError();
-        throw ZmqSocketError("Could not bind socket");
+        throw ZmqSocketError("Could not bind socket for " +
+                             sockfd.serverAddress);
     }
 }
 

slsSupportLib/tests/test-Sockets.cpp

@@ -231,14 +231,13 @@ TEST_CASE("Using DetectorSocket to talk to a Server Socket", "[support]") {
 
     auto client = DetectorSocket("localhost", port);
     int retval = 0;
-    int ret = client.sendCommandThenRead(fnum, &arg, sizeof(arg), &retval,
-                                         sizeof(retval));
+    CHECK_NOTHROW(client.sendCommandThenRead(fnum, &arg, sizeof(arg), &retval,
+                                             sizeof(retval)));
     client.close();
 
     auto server_received = s.get();
 
     // Client got OK and the expected return value back from the server
-    CHECK(ret == slsDetectorDefs::OK);
     CHECK(retval == arg * 2);
     // Server received the function number and argument we sent
     CHECK(server_received.first == fnum);

tests/scripts/test_frame_synchronizer.py

@@ -8,7 +8,6 @@ import sys, time
 import traceback, json
 
 from slsdet import Detector
-from slsdet.defines import DEFAULT_TCP_RX_PORTNO
 
 from utils_for_test import (
     Log,
@@ -22,7 +21,8 @@ from utils_for_test import (
     loadBasicSettings,
     ParseArguments, 
     build_dir,
-    optional_file
+    optional_file,
+    RX_START_TCP_PORTNO
 )
 
 LOG_PREFIX_FNAME = '/tmp/slsFrameSynchronizer_test'
@@ -44,9 +44,7 @@ def startFrameSynchronizerPullSocket(name, fp, no_log_file = False, quiet_mode=F
 
 
 def startFrameSynchronizer(num_mods, fp, no_log_file = False, quiet_mode=False):
-    cmd = [str(build_dir / 'slsFrameSynchronizer'), str(DEFAULT_TCP_RX_PORTNO), str(num_mods)]
-    # in 10.0.0
-    #cmd = ['slsFrameSynchronizer', '-p', str(DEFAULT_TCP_RX_PORTNO), '-n', str(num_mods)]
+    cmd = [str(build_dir / 'slsFrameSynchronizer'), '-p', str(RX_START_TCP_PORTNO), '-n', str(num_mods)]
     fname = SYNCHRONIZER_SUFFIX_FNAME
     if no_log_file:
         fname = None

tests/scripts/test_simulators.py

@@ -14,7 +14,6 @@ import sys, subprocess, time, traceback
 from contextlib import contextmanager
 
 from slsdet import Detector
-from slsdet.defines import DEFAULT_TCP_RX_PORTNO
 
 
 from utils_for_test import (

tests/scripts/utils_for_test.py

@@ -12,8 +12,10 @@ from datetime import timedelta
 from contextlib import contextmanager
 
 from slsdet import Detector, Ctb, detectorSettings, burstMode
-from slsdet.defines import DEFAULT_TCP_RX_PORTNO, DEFAULT_UDP_DST_PORTNO
-SERVER_START_PORTNO=1900
+from slsdet.defines import DEFAULT_UDP_DST_PORTNO
+DET_START_TCP_PORTNO=1900
+RX_START_TCP_PORTNO=2000
+START_ZMQ_PORTNO=8000
 
 LOG_PREFIX_FNAME = "/tmp/slsDetectorPackage_"
 
@@ -230,9 +232,9 @@ def runProcess(name, cmd, fp, log_file_name = None, quiet_mode=False):
 
 def startDetectorVirtualServer(name :str, num_mods, fp, no_log_file = False, quiet_mode=False):
     for i in range(num_mods):
-        port_no = SERVER_START_PORTNO + (i * 2)
+        port_no = DET_START_TCP_PORTNO + (i * 2)
         cmd = [str(build_dir / (name + 'DetectorServer_virtual')), '-p', str(port_no)]
-        fname = LOG_PREFIX_FNAME + "virtual_det_" + name + "_" + str(SERVER_START_PORTNO) + ".txt"
+        fname = LOG_PREFIX_FNAME + "virtual_det_" + name + "_" + str(DET_START_TCP_PORTNO) + ".txt"
         if no_log_file: 
             fname = None
         startProcessInBackground(cmd, fp, fname, quiet_mode)
@@ -257,7 +259,7 @@ def connectToVirtualServers(name, num_mods, ctb_object=False):
     counts_sec = 5
     while (counts_sec != 0):
         try:
-            d.virtual = [num_mods, SERVER_START_PORTNO] # sets the hostnames 
+            d.virtual = [num_mods, DET_START_TCP_PORTNO] # sets the hostnames 
             break
         except Exception as e:
             # stop server still not up, wait a bit longer
@@ -272,13 +274,11 @@ def connectToVirtualServers(name, num_mods, ctb_object=False):
 
 def startReceiver(num_mods, fp, no_log_file = False, quiet_mode=False):
     if num_mods == 1:
-        cmd = [str(build_dir / 'slsReceiver')]
+        cmd = [str(build_dir / 'slsReceiver'), '-p', str(RX_START_TCP_PORTNO)]
         fname = LOG_PREFIX_FNAME + "slsReceiver.txt"
     else:
-        cmd = [str(build_dir / 'slsMultiReceiver'), str(DEFAULT_TCP_RX_PORTNO), str(num_mods)]
+        cmd = [str(build_dir / 'slsMultiReceiver'), '-p', str(RX_START_TCP_PORTNO), '-n', str(num_mods)]
         fname = LOG_PREFIX_FNAME + "slsMultiReceiver.txt"
-        # in 10.0.0
-        #cmd = ['slsMultiReceiver', '-p', str(DEFAULT_TCP_RX_PORTNO), '-n', str(num_mods)]
 
     if no_log_file: 
         fname = None
@@ -300,6 +300,7 @@ def loadConfig(name, rx_hostname = 'localhost', settingsdir = None, log_file_fp
         if d.numinterfaces == 2: 
             d.udp_dstport2 = DEFAULT_UDP_DST_PORTNO + 1
 
+        d.rx_tcpport = RX_START_TCP_PORTNO
         d.rx_hostname = rx_hostname
         d.udp_dstip = 'auto'
         if name != "eiger":
@@ -308,6 +309,9 @@ def loadConfig(name, rx_hostname = 'localhost', settingsdir = None, log_file_fp
         if num_interfaces == 2: 
             d.udp_dstip2 = 'auto'
 
+        d.zmqport = START_ZMQ_PORTNO
+        d.rx_zmqport = START_ZMQ_PORTNO
+
         if name == "jungfrau" or name == "moench":
             d.powerchip = 1