diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 8348574e..9bef85f2 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -15,7 +15,7 @@ add_subdirectory(multipart) add_subdirectory(multiple-channels) if(BUILD_NANOMSG_TRANSPORT) add_subdirectory(multiple-transports) - endif() +endif() add_subdirectory(qc) add_subdirectory(readout) add_subdirectory(region) diff --git a/examples/qc/fairmq-start-ex-qc.sh.in b/examples/qc/fairmq-start-ex-qc.sh.in index 92e0c1aa..a64466eb 100755 --- a/examples/qc/fairmq-start-ex-qc.sh.in +++ b/examples/qc/fairmq-start-ex-qc.sh.in @@ -52,9 +52,11 @@ fairmq-dds-command-ui -c b fairmq-dds-command-ui -c x fairmq-dds-command-ui -c j fairmq-dds-command-ui -c r -sampler_and_sink="main/(Sampler|Sink).*" -fairmq-dds-command-ui -w "RUNNING->READY" -p $sampler_and_sink -echo "...$sampler_and_sink are READY, sending shutdown..." +qcconsumer="main/QCConsumer.*" +qcproducer="main/QCProducer.*" +fairmq-dds-command-ui -c p --property-key qc --property-value active -p $qcproducer +fairmq-dds-command-ui -w "RUNNING->READY" -p $qcconsumer +echo "...$qcconsumer received data and transitioned to READY, sending shutdown..." fairmq-dds-command-ui -c s fairmq-dds-command-ui -c t fairmq-dds-command-ui -c d diff --git a/examples/qc/runQCConsumer.cxx b/examples/qc/runQCConsumer.cxx index af31808a..fb337593 100644 --- a/examples/qc/runQCConsumer.cxx +++ b/examples/qc/runQCConsumer.cxx @@ -14,9 +14,9 @@ class QCConsumer : public FairMQDevice public: QCConsumer() { - OnData("qc", [](FairMQMessagePtr& /*msg*/, int){ + OnData("qc", [](FairMQMessagePtr& /*msg*/, int) { LOG(info) << "received data"; - return true; + return false; }); } }; diff --git a/examples/qc/runQCProducer.cxx b/examples/qc/runQCProducer.cxx index e3e5536f..b17ab833 100644 --- a/examples/qc/runQCProducer.cxx +++ b/examples/qc/runQCProducer.cxx @@ -14,15 +14,13 @@ class QCProducer : public FairMQDevice public: QCProducer() : fDoQC(false) - , fCounter(0) - , fInterval(100) { OnData("data1", &QCProducer::HandleData); } void InitTask() override { - GetConfig()->Subscribe("qc", [&](const std::string& key, std::string value) { + GetConfig()->Subscribe("qcdevice", [&](const std::string& key, std::string value) { if (key == "qc") { if (value == "active") { fDoQC.store(true); @@ -37,13 +35,10 @@ class QCProducer : public FairMQDevice bool HandleData(FairMQMessagePtr& msg, int) { if (fDoQC.load() == true) { - if (++fCounter == fInterval) { - fCounter = 0; - FairMQMessagePtr msgCopy(NewMessage()); - msgCopy->Copy(*msg); - if (Send(msg, "qc") < 0) { - return false; - } + FairMQMessagePtr msgCopy(NewMessage()); + msgCopy->Copy(*msg); + if (Send(msg, "qc") < 0) { + return false; } } @@ -54,12 +49,10 @@ class QCProducer : public FairMQDevice return true; } - void ResetTask() override { GetConfig()->Unsubscribe("qc"); } + void ResetTask() override { GetConfig()->Unsubscribe("qcdevice"); } private: std::atomic fDoQC; - int fCounter; - int fInterval; }; void addCustomOptions(boost::program_options::options_description& /*options*/) {} diff --git a/examples/qc/runSampler.cxx b/examples/qc/runSampler.cxx index 403c3356..b0faad57 100644 --- a/examples/qc/runSampler.cxx +++ b/examples/qc/runSampler.cxx @@ -15,41 +15,22 @@ class Sampler : public FairMQDevice { public: - Sampler() - : fMaxIterations(0) - , fNumIterations(0) - {} + Sampler() {} protected: - uint64_t fMaxIterations; - uint64_t fNumIterations; - - virtual void InitTask() - { - fMaxIterations = fConfig->GetProperty("max-iterations"); - } - virtual bool ConditionalRun() { FairMQMessagePtr msg(NewMessage(1000)); if (Send(msg, "data1") < 0) { return false; - } else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) { - LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state."; - return false; } - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - + std::this_thread::sleep_for(std::chrono::milliseconds(10)); return true; } }; namespace bpo = boost::program_options; -void addCustomOptions(bpo::options_description& options) -{ - options.add_options() - ("max-iterations", bpo::value()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)"); -} +void addCustomOptions(bpo::options_description& options) {} FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new Sampler(); } diff --git a/examples/qc/runSink.cxx b/examples/qc/runSink.cxx index c5920669..281d63f5 100644 --- a/examples/qc/runSink.cxx +++ b/examples/qc/runSink.cxx @@ -14,38 +14,12 @@ class Sink : public FairMQDevice { public: - Sink() - : fMaxIterations(0) - , fNumIterations(0) - { - OnData("data2", &Sink::HandleData); - } + Sink() { OnData("data2", &Sink::HandleData); } protected: - virtual void InitTask() - { - fMaxIterations = fConfig->GetProperty("max-iterations"); - } - - bool HandleData(FairMQMessagePtr& /*msg*/, int /*index*/) - { - if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) { - LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state."; - return false; - } - - return true; - } - - private: - uint64_t fMaxIterations; - uint64_t fNumIterations; + bool HandleData(FairMQMessagePtr& /*msg*/, int /*index*/) { return true; } }; namespace bpo = boost::program_options; -void addCustomOptions(bpo::options_description& options) -{ - options.add_options() - ("max-iterations", bpo::value()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)"); -} +void addCustomOptions(bpo::options_description& options) {} FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/) { return new Sink(); } diff --git a/fairmq/plugins/DDS/CMakeLists.txt b/fairmq/plugins/DDS/CMakeLists.txt index 731a4724..f9b07501 100644 --- a/fairmq/plugins/DDS/CMakeLists.txt +++ b/fairmq/plugins/DDS/CMakeLists.txt @@ -17,12 +17,12 @@ set_target_properties(${plugin} PROPERTIES LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/fairmq ) -set(exe1 fairmq-dds-command-ui) -add_executable(${exe1} ${CMAKE_CURRENT_SOURCE_DIR}/runDDSCommandUI.cxx) -target_link_libraries(${exe1} FairMQ Commands SDK StateMachine DDS::dds_intercom_lib DDS::dds_protocol_lib) -target_include_directories(${exe1} PRIVATE ${CMAKE_CURRENT_BINARY_DIR}) +set(exe fairmq-dds-command-ui) +add_executable(${exe} ${CMAKE_CURRENT_SOURCE_DIR}/runDDSCommandUI.cxx) +target_link_libraries(${exe} FairMQ Commands SDK StateMachine DDS::dds_intercom_lib DDS::dds_protocol_lib) +target_include_directories(${exe} PRIVATE ${CMAKE_CURRENT_BINARY_DIR}) -install(TARGETS ${plugin} ${exe1} +install(TARGETS ${plugin} ${exe} EXPORT ${PROJECT_EXPORT_SET} LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR} RUNTIME DESTINATION ${PROJECT_INSTALL_BINDIR} diff --git a/fairmq/plugins/DDS/runDDSCommandUI.cxx b/fairmq/plugins/DDS/runDDSCommandUI.cxx index 20820ca3..de11c89f 100644 --- a/fairmq/plugins/DDS/runDDSCommandUI.cxx +++ b/fairmq/plugins/DDS/runDDSCommandUI.cxx @@ -52,11 +52,11 @@ struct TerminalConfig void printControlsHelp() { cout << "Use keys to control the devices:" << endl; - cout << "[c] check states, [o] dump config, [h] help, [r] run, [s] stop, [t] reset task, [d] reset device, [q] end, [j] init task, [i] init device, [k] complete init, [b] bind, [x] connect" << endl; + cout << "[c] check states, [o] dump config, [h] help, [r] run, [s] stop, [t] reset task, [d] reset device, [q] end, [j] init task, [i] init device, [k] complete init, [b] bind, [x] connect, [p] set property" << endl; cout << "To quit press Ctrl+C" << endl; } -void handleCommand(const string& command, const string& path, unsigned int timeout, Topology& topo) +void handleCommand(const string& command, const string& path, unsigned int timeout, Topology& topo, const string& pKey, const string& pVal) { if (command == "c") { cout << "> checking state of the devices" << endl; @@ -73,6 +73,16 @@ void handleCommand(const string& command, const string& path, unsigned int timeo cout << d.first << ": " << p.first << " : " << p.second << endl; } } + } else if (command == "p") { + if (pKey == "" || pVal == "") { + cout << "cannot send property with empty key and/or value! given key: '" << pKey << "', value: '" << pVal << "'." << endl; + return; + } + const DeviceProperties props{{pKey, pVal}}; + cout << "> sending property" << endl; + topo.SetProperties(props, path); + // give dds time to complete request + this_thread::sleep_for(chrono::milliseconds(100)); } else if (command == "i") { cout << "> init devices" << endl; topo.ChangeState(TopologyTransition::InitDevice, std::chrono::milliseconds(timeout)); @@ -112,10 +122,10 @@ void handleCommand(const string& command, const string& path, unsigned int timeo } } -void sendCommand(const string& commandIn, const string& path, unsigned int timeout, Topology& topo) +void sendCommand(const string& commandIn, const string& path, unsigned int timeout, Topology& topo, const string& pKey, const string& pVal) { if (commandIn != "") { - handleCommand(commandIn, path, timeout, topo); + handleCommand(commandIn, path, timeout, topo, pKey, pVal); return; } @@ -128,7 +138,7 @@ void sendCommand(const string& commandIn, const string& path, unsigned int timeo command = c; while (true) { - handleCommand(command, path, timeout, topo); + handleCommand(command, path, timeout, topo, pKey, pVal); cin >> c; command = c; } @@ -142,6 +152,8 @@ try { string command; string path; string targetState; + string pKey; + string pVal; unsigned int timeout; fair::Logger::SetConsoleSeverity("debug"); @@ -166,6 +178,8 @@ try { options.add_options() ("command,c", bpo::value(&command)->default_value(""), "Command character") ("path,p", bpo::value(&path)->default_value(""), "DDS Topology path to send command to (empty - send to all tasks)") + ("property-key", bpo::value(&pKey)->default_value(""), "property key to be used with 'p' command") + ("property-value", bpo::value(&pVal)->default_value(""), "property value to be used with 'p' command") ("wait-for-state,w", bpo::value(&targetState)->default_value(""), "Wait until targeted FairMQ devices reach the given state") ("timeout,t", bpo::value(&timeout)->default_value(0), "Timeout in milliseconds when waiting for a device state (0 - wait infinitely)") ("help,h", "Produce help message"); @@ -175,7 +189,7 @@ try { if (vm.count("help")) { cout << "FairMQ DDS Command UI" << endl << options << endl; - cout << "Commands: [c] check state, [o] dump config, [h] help, [r] run, [s] stop, [t] reset task, [d] reset device, [q] end, [j] init task, [i] init device, [k] complete init, [b] bind, [x] connect" << endl; + cout << "Commands: [c] check state, [o] dump config, [h] help, [r] run, [s] stop, [t] reset task, [d] reset device, [q] end, [j] init task, [i] init device, [k] complete init, [b] bind, [x] connect, [p] set property" << endl; return EXIT_SUCCESS; } @@ -189,7 +203,7 @@ try { if (targetState != "") { if (command != "") { - sendCommand(command, path, timeout, topo); + sendCommand(command, path, timeout, topo, pKey, pVal); } size_t pos = targetState.find("->"); if (pos == string::npos) { @@ -200,7 +214,7 @@ try { // cout << "WaitForState(" << targetState << ") result: " << ec.message() << endl; } } else { - sendCommand(command, path, timeout, topo); + sendCommand(command, path, timeout, topo, pKey, pVal); } return EXIT_SUCCESS;