sf-daq/sf_daq_buffer
0
0
Fork 0
mirror of https://github.com/paulscherrerinstitute/sf_daq_buffer.git synced 2026-05-02 00:12:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Improve method separation in buffer

Browse Source
This commit is contained in:
babic_a
2020-06-04 10:46:45 +02:00
parent 1b49452e49
commit bd456f6fba
1 changed files with 28 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sf-buffer/src/main.cpp
+28 -22
View File
@@ -15,6 +15,33 @@ using namespace std;
using namespace chrono;
using namespace core_buffer;
void* get_live_stream_socket(const string& detector_name, const int source_id)
{
stringstream ipc_stream;
string LIVE_IPC_URL = BUFFER_LIVE_IPC_URL + detector_name + "-";
ipc_stream << LIVE_IPC_URL << source_id;
const auto ipc_address = ipc_stream.str();
void* ctx = zmq_ctx_new();
void* socket = zmq_socket(ctx, ZMQ_PUB);
const int sndhwm = BUFFER_ZMQ_SNDHWM;
if (zmq_setsockopt(socket, ZMQ_SNDHWM, &sndhwm, sizeof(sndhwm)) != 0) {
throw runtime_error(zmq_strerror(errno));
}
const int linger = 0;
if (zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger)) != 0) {
throw runtime_error(zmq_strerror(errno));
}
if (zmq_bind(socket, ipc_address.c_str()) != 0) {
throw runtime_error(zmq_strerror(errno));
}
return socket;
}
int main (int argc, char *argv[]) {
if (argc != 6) {
@@ -38,28 +65,6 @@ int main (int argc, char *argv[]) {
string root_folder = string(argv[4]);
int source_id = atoi(argv[5]);
stringstream ipc_stream;
string LIVE_IPC_URL = BUFFER_LIVE_IPC_URL + detector_name + "-";
ipc_stream << LIVE_IPC_URL << source_id;
const auto ipc_address = ipc_stream.str();
auto ctx = zmq_ctx_new();
auto socket = zmq_socket(ctx, ZMQ_PUB);
const int sndhwm = BUFFER_ZMQ_SNDHWM;
if (zmq_setsockopt(socket, ZMQ_SNDHWM, &sndhwm, sizeof(sndhwm)) != 0) {
throw runtime_error(zmq_strerror(errno));
}
const int linger = 0;
if (zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger)) != 0) {
throw runtime_error(zmq_strerror(errno));
}
if (zmq_bind(socket, ipc_address.c_str()) != 0) {
throw runtime_error(zmq_strerror(errno));
}
uint64_t stats_counter(0);
uint64_t n_missed_packets = 0;
uint64_t n_corrupted_frames = 0;
@@ -68,6 +73,7 @@ int main (int argc, char *argv[]) {
BufferUdpReceiver receiver(udp_port, source_id);
BufferBinaryFormat* binary_buffer = new BufferBinaryFormat();
auto socket = get_live_stream_socket(detector_name, source_id);
size_t write_total_us = 0;
size_t write_max_us = 0;