diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/DabMod.cpp | 21 | ||||
| -rw-r--r-- | src/OutputZeroMQ.cpp | 28 | ||||
| -rw-r--r-- | src/OutputZeroMQ.h | 5 |
3 files changed, 44 insertions, 10 deletions
diff --git a/src/DabMod.cpp b/src/DabMod.cpp index 75e76e0..304d252 100644 --- a/src/DabMod.cpp +++ b/src/DabMod.cpp @@ -47,6 +47,7 @@ #include "RemoteControl.h" #include <boost/shared_ptr.hpp> +#include <boost/make_shared.hpp> #include <boost/property_tree/ptree.hpp> #include <boost/property_tree/ini_parser.hpp> #include <complex> @@ -119,6 +120,7 @@ int main(int argc, char* argv[]) std::string outputName; int useZeroMQOutput = 0; + std::string zmqOutputSocketType = ""; int useFileOutput = 0; std::string fileOutputFormat = "complexf"; int useUHDOutput = 0; @@ -563,6 +565,7 @@ int main(int argc, char* argv[]) #if defined(HAVE_ZEROMQ) else if (output_selected == "zmq") { outputName = pt.get<std::string>("zmqoutput.listen"); + zmqOutputSocketType = pt.get<std::string>("zmqoutput.socket_type"); useZeroMQOutput = 1; } #endif @@ -676,8 +679,10 @@ int main(int argc, char* argv[]) #endif else if (useZeroMQOutput) { fprintf(stderr, " ZeroMQ\n" - " Listening on: %s\n", - outputName.c_str()); + " Listening on: %s\n" + " Socket type : %s\n", + outputName.c_str(), + zmqOutputSocketType.c_str()); } fprintf(stderr, " Sampling rate: "); @@ -744,7 +749,17 @@ int main(int argc, char* argv[]) else if (useZeroMQOutput) { /* We normalise the same way as for the UHD output */ normalise = 1.0f / normalise_factor; - output = shared_ptr<OutputZeroMQ>(new OutputZeroMQ(outputName)); + if (zmqOutputSocketType == "pub") { + output = make_shared<OutputZeroMQ>(outputName, ZMQ_PUB); + } + else if (zmqOutputSocketType == "rep") { + output = make_shared<OutputZeroMQ>(outputName, ZMQ_REP); + } + else { + std::stringstream ss; + ss << "ZeroMQ output socket type " << zmqOutputSocketType << " invalid"; + throw std::invalid_argument(ss.str()); + } } #endif diff --git a/src/OutputZeroMQ.cpp b/src/OutputZeroMQ.cpp index 793e473..da4473e 100644 --- a/src/OutputZeroMQ.cpp +++ b/src/OutputZeroMQ.cpp @@ -32,19 +32,31 @@ #if defined(HAVE_ZEROMQ) -OutputZeroMQ::OutputZeroMQ(std::string endpoint, Buffer* dataOut) +OutputZeroMQ::OutputZeroMQ(std::string endpoint, int type, Buffer* dataOut) : ModOutput(ModFormat(1), ModFormat(0)), + m_type(type), m_zmq_context(1), - m_zmq_pub_sock(m_zmq_context, ZMQ_PUB), + m_zmq_sock(m_zmq_context, type), m_endpoint(endpoint) { PDEBUG("OutputZeroMQ::OutputZeroMQ(%p) @ %p\n", dataOut, this); std::stringstream ss; - ss << "OutputZeroMQ(" << m_endpoint << ")"; + ss << "OutputZeroMQ(" << m_endpoint << " "; + + if (type == ZMQ_PUB) { + ss << "ZMQ_PUB"; + } + else if (type == ZMQ_REP) { + ss << "ZMQ_REP"; + } + else { + throw std::invalid_argument("ZMQ socket type unknown"); + } + ss << ")"; m_name = ss.str(); - m_zmq_pub_sock.bind(m_endpoint.c_str()); + m_zmq_sock.bind(m_endpoint.c_str()); } OutputZeroMQ::~OutputZeroMQ() @@ -58,7 +70,13 @@ int OutputZeroMQ::process(Buffer* dataIn, Buffer* dataOut) "(dataIn: %p, dataOut: %p)\n", dataIn, dataOut); - m_zmq_pub_sock.send(dataIn->getData(), dataIn->getLength()); + if (m_type == ZMQ_REP) { + // A ZMQ_REP socket requires a request first + zmq::message_t msg; + m_zmq_sock.recv(&msg); + } + + m_zmq_sock.send(dataIn->getData(), dataIn->getLength()); return dataIn->getLength(); } diff --git a/src/OutputZeroMQ.h b/src/OutputZeroMQ.h index a80eab4..85f85a7 100644 --- a/src/OutputZeroMQ.h +++ b/src/OutputZeroMQ.h @@ -39,14 +39,15 @@ class OutputZeroMQ : public ModOutput { public: - OutputZeroMQ(std::string endpoint, Buffer* dataOut = NULL); + OutputZeroMQ(std::string endpoint, int type, Buffer* dataOut = NULL); virtual ~OutputZeroMQ(); virtual int process(Buffer* dataIn, Buffer* dataOut); const char* name() { return m_name.c_str(); } protected: + int m_type; // zmq socket type zmq::context_t m_zmq_context; // handle for the zmq context - zmq::socket_t m_zmq_pub_sock; // handle for the zmq publisher socket + zmq::socket_t m_zmq_sock; // handle for the zmq publisher socket std::string m_endpoint; // On which port to listen: e.g. // tcp://*:58300 |