examples/StreamingExample/ZmqMain.cc

0001
0002 // Copyright 2020, Jefferson Science Associates, LLC.
0003 // Subject to the terms in the LICENSE file found in the top-level directory.
0004
0005
0006
0007 #include <JANA/JApplication.h>
0008 #include <JANA/JFactoryGenerator.h>
0009 #include <JANA/JEventSourceGeneratorT.h>
0010 #include <JANA/Utils/JBenchUtils.h>
0011 #include <JANA/Streaming/JEventBuilder.h>
0012 #include <JANA/Streaming/JStreamingEventSource.h>
0013
0014
0015 #include "ReadoutMessageAuto.h"
0016 #include "ZmqTransport.h"
0017 #include "AHitParser.h"
0018 #include "AHitAnomalyDetector.h"
0019
0020
0021 void dummy_publisher_loop() {
0022
0023     JBenchUtils bench_utils = JBenchUtils();
0024     bench_utils.set_seed(6, "ZmqMain.cc:dummy_publisher_loop");
0025     bench_utils.consume_cpu_ms(3000, 0, false);
0026
0027     auto transport = ZmqTransport("tcp://127.0.0.1:5555", true);
0028     transport.initialize();
0029
0030     for (size_t counter = 1; counter < 11; ++counter) {
0031
0032         ReadoutMessageAuto message(nullptr);
0033         message.run_number = 0;
0034         message.event_number = counter;
0035
0036         message.payload_size = 4;
0037         message.payload[0] = bench_utils.randfloat(0,1);
0038         message.payload[1] = bench_utils.randfloat(-100,100);
0039         message.payload[2] = bench_utils.randfloat(-100,100);
0040         message.payload[3] = bench_utils.randfloat(-100,100);
0041
0042         transport.send(message);
0043         std::cout << "Send: " << message << "(" << message.get_buffer_capacity() << " bytes)" << std::endl;
0044         bench_utils.consume_cpu_ms(1000, 0, false);
0045     }
0046
0047     // Send end-of-stream message so that JANA knows to shut down
0048     ReadoutMessageAuto message (nullptr);
0049     message.set_end_of_stream();
0050     transport.send(message);
0051 }
0052
0053
0054 extern "C"{
0055 void InitPlugin(JApplication *app) {
0056
0057     InitJANAPlugin(app);
0058
0059     auto transport = std::unique_ptr<ZmqTransport>(new ZmqTransport("tcp://127.0.0.1:5555"));
0060     app->Add(new JStreamingEventSource<ReadoutMessageAuto>(std::move(transport)));
0061
0062     app->Add(new AHitAnomalyDetector(app, 5000));
0063     app->Add(new JFactoryGeneratorT<AHitParser>());
0064
0065     app->SetParameterValue("jana:extended_report", 0);
0066
0067     new std::thread(dummy_publisher_loop);
0068 }
0069 } // "C"
0070
0071