hyperion.ng/libsrc/flatbufserver/FlatBufferClient.cpp
Paulchen-Panther ce2f0a17d5
Proto- and Flatbuffer now share their input to all instances
Signed-off-by: Paulchen-Panther <Paulchen-Panter@protonmail.com>
2019-08-03 19:14:52 +02:00

218 lines
5.7 KiB
C++

#include "FlatBufferClient.h"
// qt
#include <QTcpSocket>
#include <QHostAddress>
#include <QTimer>
#include <QRgb>
#include <hyperion/Hyperion.h>
#include <hyperion/HyperionIManager.h>
FlatBufferClient::FlatBufferClient(QTcpSocket* socket, const int &timeout, QObject *parent)
: QObject(parent)
, _log(Logger::getInstance("FLATBUFSERVER"))
, _socket(socket)
, _clientAddress("@"+socket->peerAddress().toString())
, _timeoutTimer(new QTimer(this))
, _timeout(timeout * 1000)
, _priority()
{
// timer setup
_timeoutTimer->setSingleShot(true);
_timeoutTimer->setInterval(_timeout);
connect(_timeoutTimer, &QTimer::timeout, this, &FlatBufferClient::forceClose);
// connect socket signals
connect(_socket, &QTcpSocket::readyRead, this, &FlatBufferClient::readyRead);
connect(_socket, &QTcpSocket::disconnected, this, &FlatBufferClient::disconnected);
}
void FlatBufferClient::readyRead()
{
_timeoutTimer->start();
_receiveBuffer += _socket->readAll();
// check if we can read a header
while(_receiveBuffer.size() >= 4)
{
uint32_t messageSize =
((_receiveBuffer[0]<<24) & 0xFF000000) |
((_receiveBuffer[1]<<16) & 0x00FF0000) |
((_receiveBuffer[2]<< 8) & 0x0000FF00) |
((_receiveBuffer[3] ) & 0x000000FF);
// check if we can read a complete message
if((uint32_t) _receiveBuffer.size() < messageSize + 4) return;
// extract message only and remove header + msg from buffer :: QByteArray::remove() does not return the removed data
const QByteArray msg = _receiveBuffer.right(messageSize);
_receiveBuffer.remove(0, messageSize + 4);
const auto* msgData = reinterpret_cast<const uint8_t*>(msg.constData());
flatbuffers::Verifier verifier(msgData, messageSize);
if (hyperionnet::VerifyRequestBuffer(verifier))
{
auto message = hyperionnet::GetRequest(msgData);
handleMessage(message);
continue;
}
sendErrorReply("Unable to parse message");
}
}
void FlatBufferClient::forceClose()
{
_socket->close();
}
void FlatBufferClient::disconnected()
{
Debug(_log, "Socket Closed");
_socket->deleteLater();
emit clearGlobalInput(_priority);
emit clientDisconnected();
}
void FlatBufferClient::handleMessage(const hyperionnet::Request * req)
{
const void* reqPtr;
if ((reqPtr = req->command_as_Color()) != nullptr) {
handleColorCommand(static_cast<const hyperionnet::Color*>(reqPtr));
} else if ((reqPtr = req->command_as_Image()) != nullptr) {
handleImageCommand(static_cast<const hyperionnet::Image*>(reqPtr));
} else if ((reqPtr = req->command_as_Clear()) != nullptr) {
handleClearCommand(static_cast<const hyperionnet::Clear*>(reqPtr));
} else if ((reqPtr = req->command_as_Register()) != nullptr) {
handleRegisterCommand(static_cast<const hyperionnet::Register*>(reqPtr));
} else {
sendErrorReply("Received invalid packet.");
}
}
void FlatBufferClient::handleColorCommand(const hyperionnet::Color *colorReq)
{
// extract parameters
const int32_t rgbData = colorReq->data();
ColorRgb color;
color.red = qRed(rgbData);
color.green = qGreen(rgbData);
color.blue = qBlue(rgbData);
// set output
emit setGlobalInputColor(_priority, color, colorReq->duration());
// send reply
sendSuccessReply();
}
void FlatBufferClient::registationRequired(const int priority)
{
if (_priority == priority)
{
auto reply = hyperionnet::CreateReplyDirect(_builder, nullptr, -1, -1);
_builder.Finish(reply);
// send reply
sendMessage();
}
}
void FlatBufferClient::handleRegisterCommand(const hyperionnet::Register *regReq)
{
_priority = regReq->priority();
emit registerGlobalInput(_priority, hyperion::COMP_FLATBUFSERVER, regReq->origin()->c_str()+_clientAddress);
auto reply = hyperionnet::CreateReplyDirect(_builder, nullptr, -1, (_priority ? _priority : -1));
_builder.Finish(reply);
// send reply
sendMessage();
}
void FlatBufferClient::handleImageCommand(const hyperionnet::Image *image)
{
// extract parameters
int duration = image->duration();
const void* reqPtr;
if ((reqPtr = image->data_as_RawImage()) != nullptr)
{
const auto *img = static_cast<const hyperionnet::RawImage*>(reqPtr);
const auto & imageData = img->data();
const int width = img->width();
const int height = img->height();
if ((int) imageData->size() != width*height*3)
{
sendErrorReply("Size of image data does not match with the width and height");
return;
}
Image<ColorRgb> imageDest(width, height);
memmove(imageDest.memptr(), imageData->data(), imageData->size());
emit setGlobalInputImage(_priority, imageDest, duration);
}
// send reply
sendSuccessReply();
}
void FlatBufferClient::handleClearCommand(const hyperionnet::Clear *clear)
{
// extract parameters
const int priority = clear->priority();
if (priority == -1) {
emit clearAllGlobalInput();
}
else {
// Check if we are clearing ourselves.
if (priority == _priority) {
_priority = -1;
}
emit clearGlobalInput(priority);
}
sendSuccessReply();
}
void FlatBufferClient::handleNotImplemented()
{
sendErrorReply("Command not implemented");
}
void FlatBufferClient::sendMessage()
{
auto size = _builder.GetSize();
const uint8_t* buffer = _builder.GetBufferPointer();
uint8_t sizeData[] = {uint8_t(size >> 24), uint8_t(size >> 16), uint8_t(size >> 8), uint8_t(size)};
_socket->write((const char *) sizeData, sizeof(sizeData));
_socket->write((const char *)buffer, size);
_socket->flush();
_builder.Clear();
}
void FlatBufferClient::sendSuccessReply()
{
auto reply = hyperionnet::CreateReplyDirect(_builder);
_builder.Finish(reply);
// send reply
sendMessage();
}
void FlatBufferClient::sendErrorReply(const std::string &error)
{
// create reply
auto reply = hyperionnet::CreateReplyDirect(_builder, error.c_str());
_builder.Finish(reply);
// send reply
sendMessage();
}