introduce the aircraft scheduler to receive new messages from the backend
threading-concepts are used to avoid unnecessary main thread action
This commit is contained in:
75
include/aman/com/AircraftScheduler.h
Normal file
75
include/aman/com/AircraftScheduler.h
Normal file
@@ -0,0 +1,75 @@
|
||||
/*
|
||||
* @brief Defines the aircraft scheduling notifier module to receive plans from the backend
|
||||
* @file aman/com/AircraftScheduler.h
|
||||
* @author Sven Czarnian <devel@svcz.de>
|
||||
* @copyright Copyright 2021 Sven Czarnian
|
||||
* @license This project is published under the GNU General Public License v3 (GPLv3)
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <thread>
|
||||
#include <zmq.hpp>
|
||||
|
||||
#include <aman/types/Communication.h>
|
||||
|
||||
#include "protobuf/AircraftSchedule.pb.h"
|
||||
|
||||
namespace aman {
|
||||
/**
|
||||
* @brief Defines the aircraft scheduling notification class to receive scheduling sequences
|
||||
* @ingroup com
|
||||
*/
|
||||
class AircraftScheduler {
|
||||
private:
|
||||
std::unique_ptr<zmq::socket_t> m_socket;
|
||||
std::thread m_receiverThread;
|
||||
std::atomic_bool m_stopReceiver;
|
||||
|
||||
AircraftScheduler() noexcept;
|
||||
|
||||
template <typename T>
|
||||
bool setSocketKey(const std::string& key, T entry) {
|
||||
try {
|
||||
this->m_socket->set(entry, key);
|
||||
return true;
|
||||
}
|
||||
catch (std::exception&) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
void run();
|
||||
|
||||
public:
|
||||
AircraftScheduler(const AircraftScheduler&) = delete;
|
||||
AircraftScheduler(AircraftScheduler&&) = delete;
|
||||
AircraftScheduler& operator=(const AircraftScheduler&) = delete;
|
||||
AircraftScheduler& operator=(AircraftScheduler&&) = delete;
|
||||
|
||||
/**
|
||||
* @brief Initializes the aircraft scheduler
|
||||
* @param[in] configuration The current AMAM communication configuration
|
||||
* @return True if the initialization is done, else false
|
||||
*/
|
||||
bool initialize(const Communication& configuration);
|
||||
/**
|
||||
* @brief Terminates the scheduler connection
|
||||
*/
|
||||
bool deinitialize();
|
||||
/**
|
||||
* @brief Checks if the scheduler is initialized
|
||||
* @return True if it is initialized, else false
|
||||
*/
|
||||
bool initialized() const noexcept;
|
||||
/**
|
||||
* @brief Receives a new scheduling message
|
||||
* @return The new scheduling message or a nullptr
|
||||
*/
|
||||
std::unique_ptr<aman::AircraftSchedule> receive();
|
||||
/**
|
||||
* @brief Returns the scheduling instance
|
||||
* @return The system-wide instance
|
||||
*/
|
||||
static AircraftScheduler& instance() noexcept;
|
||||
};
|
||||
}
|
||||
@@ -32,6 +32,7 @@ SET(SOURCE_FILES
|
||||
|
||||
SET(SOURCE_COM_FILES
|
||||
com/AircraftReporter.cpp
|
||||
com/AircraftScheduler.cpp
|
||||
com/ZmqContext.cpp
|
||||
com/ZmqContext.h
|
||||
)
|
||||
@@ -50,6 +51,7 @@ SET(SOURCE_FILES_RES
|
||||
|
||||
SET(INCLUDE_COM_FILES
|
||||
${CMAKE_SOURCE_DIR}/include/aman/com/AircraftReporter.h
|
||||
${CMAKE_SOURCE_DIR}/include/aman/com/AircraftScheduler.h
|
||||
)
|
||||
|
||||
SET(INCLUDE_CONFIG_FILES
|
||||
|
||||
105
src/com/AircraftScheduler.cpp
Normal file
105
src/com/AircraftScheduler.cpp
Normal file
@@ -0,0 +1,105 @@
|
||||
/*
|
||||
* Author:
|
||||
* Sven Czarnian <devel@svcz.de>
|
||||
* Brief:
|
||||
* Implements the scheduling notification
|
||||
* Copyright:
|
||||
* 2021 Sven Czarnian
|
||||
* License:
|
||||
* GNU General Public License v3 (GPLv3)
|
||||
*/
|
||||
|
||||
#include <aman/com/AircraftScheduler.h>
|
||||
|
||||
#include "ZmqContext.h"
|
||||
|
||||
using namespace aman;
|
||||
using namespace std::chrono;
|
||||
|
||||
AircraftScheduler::AircraftScheduler() noexcept :
|
||||
m_socket(),
|
||||
m_receiverThread(),
|
||||
m_stopReceiver(false) { }
|
||||
|
||||
bool AircraftScheduler::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(&AircraftScheduler::run, this);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool AircraftScheduler::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 AircraftScheduler::initialized() const noexcept {
|
||||
return nullptr != this->m_socket;
|
||||
}
|
||||
|
||||
std::unique_ptr<aman::AircraftSchedule> AircraftScheduler::receive() {
|
||||
zmq::message_t message;
|
||||
|
||||
if (nullptr == this->m_socket)
|
||||
return std::unique_ptr<aman::AircraftSchedule>();
|
||||
|
||||
try {
|
||||
auto result = this->m_socket->recv(message, zmq::recv_flags::dontwait);
|
||||
if (false == result.has_value() || 0 == result.value())
|
||||
return std::unique_ptr<aman::AircraftSchedule>();
|
||||
}
|
||||
catch (zmq::error_t&) {
|
||||
return std::unique_ptr<aman::AircraftSchedule>();
|
||||
}
|
||||
|
||||
std::unique_ptr<aman::AircraftSchedule> retval = std::make_unique<aman::AircraftSchedule>();
|
||||
retval->ParseFromString(reinterpret_cast<const char*>(message.data()));
|
||||
|
||||
return retval;
|
||||
}
|
||||
|
||||
void AircraftScheduler::run() {
|
||||
while (false == this->m_stopReceiver) {
|
||||
auto message = this->receive();
|
||||
if (nullptr == message) {
|
||||
std::this_thread::sleep_for(500ms);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
AircraftScheduler& AircraftScheduler::instance() noexcept {
|
||||
static AircraftScheduler __instance;
|
||||
return __instance;
|
||||
}
|
||||
Reference in New Issue
Block a user