use only one backend and adapt to the new communication structure

avoid empty messages
This commit is contained in:
Sven Czarnian
2021-11-18 16:02:41 +01:00
parent a65aa445fa
commit 689c2326c2
6 changed files with 63 additions and 250 deletions

View File

@@ -2,31 +2,30 @@
* Author:
* Sven Czarnian <devel@svcz.de>
* Brief:
* Implements the backend notification
* Implements the backend communication
* Copyright:
* 2021 Sven Czarnian
* License:
* GNU General Public License v3 (GPLv3)
*/
#include <aman/com/BackendNotification.h>
#include <aman/com/Backend.h>
#include <aman/helper/String.h>
#include "ZmqContext.h"
using namespace aman;
BackendNotification::BackendNotification() noexcept :
Backend::Backend() noexcept :
m_socket() { }
bool BackendNotification::initialize(const Communication& configuration) {
bool Backend::initialize(const Communication& configuration) {
if (nullptr != this->m_socket)
return true;
if (false == configuration.valid)
return false;
this->m_socket = std::make_unique<zmq::socket_t>(ZmqContext::instance().context(), zmq::socket_type::pub);
this->m_socket->set(zmq::sockopt::immediate, true);
this->m_socket = std::make_unique<zmq::socket_t>(ZmqContext::instance().context(), zmq::socket_type::req);
/* configure the encryption */
if (false == this->setSocketKey(configuration.serverPublicIdentifier, zmq::sockopt::curve_serverkey))
@@ -48,7 +47,7 @@ bool BackendNotification::initialize(const Communication& configuration) {
return true;
}
bool BackendNotification::deinitialize() {
bool Backend::deinitialize() {
if (nullptr == this->m_socket)
return true;
@@ -58,13 +57,33 @@ bool BackendNotification::deinitialize() {
return true;
}
bool BackendNotification::initialized() const noexcept {
bool Backend::initialized() const noexcept {
return nullptr != this->m_socket;
}
bool BackendNotification::send(aman::AircraftUpdate& report) {
bool retval = false;
std::shared_ptr<aman::AircraftSequence> Backend::receiveSequence() {
zmq::message_t message;
if (nullptr == this->m_socket)
return nullptr;
try {
//auto result = this->m_socket->recv(message, zmq::recv_flags::dontwait);
auto result = this->m_socket->recv(message);
if (false == result.has_value() || 0 == result.value())
return nullptr;
}
catch (zmq::error_t&) {
return nullptr;
}
std::unique_ptr<aman::AircraftSequence> retval = std::make_unique<aman::AircraftSequence>();
retval->ParseFromString(reinterpret_cast<const char*>(message.data()));
return std::move(retval);
}
std::shared_ptr<aman::AircraftSequence> Backend::update(aman::AircraftUpdate& report) {
if (nullptr != this->m_socket) {
/* serialize the report */
std::string serialized = report.SerializeAsString();
@@ -72,19 +91,20 @@ bool BackendNotification::send(aman::AircraftUpdate& report) {
std::memcpy(message.data(), serialized.c_str(), serialized.size());
try {
auto size = message.size();
auto result = this->m_socket->send(message, zmq::send_flags::none);
retval = result.value() == size;
const auto size = message.size();
const auto result = this->m_socket->send(message, zmq::send_flags::none);
if (result.value() == size)
return this->receiveSequence();
}
catch (zmq::error_t&) {
return false;
return nullptr;
}
}
return retval;
return nullptr;
}
BackendNotification& BackendNotification::instance() noexcept {
static BackendNotification __instance;
Backend& Backend::instance() noexcept {
static Backend __instance;
return __instance;
}

View File

@@ -1,121 +0,0 @@
/*
* Author:
* Sven Czarnian <devel@svcz.de>
* Brief:
* Implements the backend receiver
* Copyright:
* 2021 Sven Czarnian
* License:
* GNU General Public License v3 (GPLv3)
*/
#include <aman/com/BackendReceiver.h>
#include "ZmqContext.h"
using namespace aman;
using namespace std::chrono;
BackendReceiver::BackendReceiver() noexcept :
m_socket(),
m_receiverThread(),
m_stopReceiver(false),
m_sequences(),
m_sequencesLock() { }
bool BackendReceiver::initialize(const Communication& configuration) {
if (nullptr != this->m_socket)
return true;
if (false == configuration.valid)
return false;
this->m_socket = std::make_unique<zmq::socket_t>(ZmqContext::instance().context(), zmq::socket_type::sub);
/* configure the encryption */
if (false == this->setSocketKey(configuration.serverPublicIdentifier, zmq::sockopt::curve_serverkey))
return false;
if (false == this->setSocketKey(configuration.clientPublicIdentifier, zmq::sockopt::curve_publickey))
return false;
if (false == this->setSocketKey(configuration.clientPrivateIdentifier, zmq::sockopt::curve_secretkey))
return false;
/* connect to the server */
try {
this->m_socket->connect("tcp://" + configuration.address + ":" + std::to_string(configuration.portNotification));
}
catch (zmq::error_t&) {
this->m_socket = std::unique_ptr<zmq::socket_t>();
return false;
}
this->m_stopReceiver = false;
this->m_receiverThread = std::thread(&BackendReceiver::run, this);
return true;
}
bool BackendReceiver::deinitialize() {
if (nullptr == this->m_socket)
return true;
this->m_stopReceiver = true;
this->m_receiverThread.join();
this->m_socket->close();
this->m_socket = std::make_unique<zmq::socket_t>();
return true;
}
bool BackendReceiver::initialized() const noexcept {
return nullptr != this->m_socket;
}
void BackendReceiver::receiveSequence() {
zmq::message_t message;
if (nullptr == this->m_socket)
return;
try {
auto result = this->m_socket->recv(message, zmq::recv_flags::dontwait);
if (false == result.has_value() || 0 == result.value())
return;
}
catch (zmq::error_t&) {
return;
}
std::unique_ptr<aman::AircraftSequence> retval = std::make_unique<aman::AircraftSequence>();
retval->ParseFromString(reinterpret_cast<const char*>(message.data()));
std::lock_guard guard(this->m_sequencesLock);
this->m_sequences.push_back(std::move(retval));
}
void BackendReceiver::run() {
while (false == this->m_stopReceiver) {
auto message = this->receive();
if (nullptr == message) {
std::this_thread::sleep_for(500ms);
continue;
}
this->receiveSequence();
}
}
std::shared_ptr<aman::AircraftSequence> BackendReceiver::receive() {
std::lock_guard guard(this->m_sequencesLock);
if (0 != this->m_sequences.size()) {
auto retval = this->m_sequences.front();
this->m_sequences.pop_front();
return retval;
}
return nullptr;
}
BackendReceiver& BackendReceiver::instance() noexcept {
static BackendReceiver __instance;
return __instance;
}