ETM-Schurig/etm-powersync-plugins
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

TCP Commander: Make the commander bidirectional

Browse Source
This commit is contained in:
Michael Zanetti 2020-11-23 23:31:36 +01:00
parent 1d06c6c001
commit 8dfad31006
9 changed files with 169 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
tcpcommander/README.md
View File

@ -4,18 +4,13 @@ This plugin is a generic approach to allow sending and receiving custom TCP pack
> Note: This plugin is ment to be combined with a rule.
## TCP output
## TCP client
The TCP output opens a TCP connection to the given host IPv4 address and port everytime the output trigger gets activated. As soon
as the command has been executed the socket will close again. The connected state is only stated as connected
as long as the connection is active.
The TCP output opens a TCP connection to the given host address and port. Once the connection is established, TCP packets can be sent both directions.
## TCP input
## TCP server
The TCP input creates a TCP server on the given port.
Be aware that only a single connection can be established simultaneously. The connected state is active as long as
a client is connected. It is up to the client to deside how long the connection stays active.
The TCP input creates a TCP server on the given port. Other applications may connect to this server and send messages to it which can be processed further within nymea. Also, TCP packets can be sent to all or individual clients. Use the address 0.0.0.0 (the default) to send the data to all connected clients.
## Example

121
tcpcommander/integrationplugintcpcommander.cpp
View File

@ -31,6 +31,7 @@
#include "integrationplugintcpcommander.h"
#include "plugininfo.h"
#include <QTimer>
IntegrationPluginTcpCommander::IntegrationPluginTcpCommander()
{
@ -41,33 +42,53 @@ void IntegrationPluginTcpCommander::setupThing(ThingSetupInfo *info)
{
Thing *thing = info->thing();
if (thing->thingClassId() == tcpOutputThingClassId) {
if (thing->thingClassId() == tcpClientThingClassId) {
quint16 port = thing->paramValue(tcpOutputThingPortParamTypeId).toUInt();
QHostAddress address= QHostAddress(thing->paramValue(tcpOutputThingIpv4addressParamTypeId).toString());
TcpSocket *tcpSocket = new TcpSocket(address, port, this);
m_tcpSockets.insert(tcpSocket, thing);
connect(tcpSocket, &TcpSocket::connectionChanged, this, &IntegrationPluginTcpCommander::onTcpSocketConnectionChanged);
quint16 port = thing->paramValue(tcpClientThingPortParamTypeId).toUInt();
QHostAddress address= QHostAddress(thing->paramValue(tcpClientThingIpv4addressParamTypeId).toString());
QTcpSocket *tcpSocket = m_tcpSockets.value(thing);
if (!tcpSocket) {
tcpSocket = new QTcpSocket(this);
m_tcpSockets.insert(thing, tcpSocket);
} else {
// In case of a reconfigure, make sure we reconnect
tcpSocket->disconnectFromHost();
}
connect(tcpSocket, &QTcpSocket::stateChanged, thing, [=](QAbstractSocket::SocketState state){
thing->setStateValue(tcpClientConnectedStateTypeId, state == QAbstractSocket::ConnectedState);
connect(tcpSocket, &TcpSocket::connectionTestFinished, info, [info] (bool status) {
if (status) {
info->finish(Thing::ThingErrorNoError);
} else {
info->finish(Thing::ThingErrorSetupFailed, QT_TR_NOOP("Error connecting to remote server."));
if (state == QAbstractSocket::UnconnectedState) {
QTimer::singleShot(10000, tcpSocket, [=](){
qCDebug(dcTCPCommander()) << "Reconnecting to server" << address << port;
tcpSocket->connectToHost(address, port);
});
}
});
connect(tcpSocket, &QTcpSocket::readyRead, thing, [=](){
QByteArray data = tcpSocket->readAll();
ParamList params;
params << Param(tcpClientTriggeredEventDataParamTypeId, data);
Event event(tcpClientTriggeredEventTypeId, thing->id(), params);
emitEvent(event);
});
// Test the socket, if a socket can be established the setup process was successfull
tcpSocket->connectionTest();
tcpSocket->connectToHost(address, port);
info->finish(Thing::ThingErrorNoError);
return;
}
if (thing->thingClassId() == tcpInputThingClassId) {
int port = thing->paramValue(tcpInputThingPortParamTypeId).toInt();
TcpServer *tcpServer = new TcpServer(port, this);
if (thing->thingClassId() == tcpServerThingClassId) {
int port = thing->paramValue(tcpServerThingPortParamTypeId).toInt();
TcpServer *tcpServer = m_tcpServers.value(thing);
if (tcpServer) {
// In case of reconfigure, make sure to re-setup the server
delete tcpServer;
}
tcpServer = new TcpServer(port, this);
if (tcpServer->isValid()) {
m_tcpServer.insert(tcpServer, thing);
m_tcpServers.insert(thing, tcpServer);
connect(tcpServer, &TcpServer::connectionCountChanged, this, &IntegrationPluginTcpCommander::onTcpServerConnectionCountChanged);
connect(tcpServer, &TcpServer::commandReceived, this, &IntegrationPluginTcpCommander::onTcpServerCommandReceived);
return info->finish(Thing::ThingErrorNoError);
@ -85,34 +106,39 @@ void IntegrationPluginTcpCommander::executeAction(ThingActionInfo *info)
Thing *thing = info->thing();
Action action = info->action();
Q_ASSERT_X(action.actionTypeId() == tcpOutputTriggerActionTypeId, "TcpCommander", "Invalid action type in executeAction");
TcpSocket *tcpSocket = m_tcpSockets.key(thing);
QByteArray data = action.param(tcpOutputTriggerActionOutputDataAreaParamTypeId).value().toByteArray();
tcpSocket->sendCommand(data);
connect(tcpSocket, &TcpSocket::commandSent, info, [info](bool success){
if (success) {
if (action.actionTypeId() == tcpClientTriggerActionTypeId) {
QTcpSocket *tcpSocket = m_tcpSockets.value(thing);
QByteArray data = action.param(tcpClientTriggerActionDataParamTypeId).value().toByteArray();
qint64 len = tcpSocket->write(data);
if (len == data.length()) {
info->finish(Thing::ThingErrorNoError);
} else {
info->finish(Thing::ThingErrorHardwareNotAvailable);
}
});
}
else if (action.actionTypeId() == tcpServerTriggerActionTypeId){
TcpServer *server = m_tcpServers.value(thing);
QByteArray data = action.param(tcpServerTriggerActionDataParamTypeId).value().toByteArray();
QString clientIp = action.param(tcpServerTriggerActionClientIpParamTypeId).value().toString();
bool success = server->sendCommand(clientIp, data);
if (success) {
info->finish(Thing::ThingErrorNoError);
} else {
info->finish(Thing::ThingErrorHardwareFailure, QT_TR_NOOP("Failed to send the command to the specified client(s)."));
}
}
}
void IntegrationPluginTcpCommander::thingRemoved(Thing *thing)
{
if(thing->thingClassId() == tcpOutputThingClassId){
TcpSocket *tcpSocket = m_tcpSockets.key(thing);
m_tcpSockets.remove(tcpSocket);
if(thing->thingClassId() == tcpClientThingClassId){
QTcpSocket *tcpSocket = m_tcpSockets.take(thing);
tcpSocket->deleteLater();
} else if(thing->thingClassId() == tcpInputThingClassId){
TcpServer *tcpServer = m_tcpServer.key(thing);
m_tcpServer.remove(tcpServer);
} else if(thing->thingClassId() == tcpServerThingClassId){
TcpServer *tcpServer = m_tcpServers.take(thing);
tcpServer->deleteLater();
}
}
@ -120,10 +146,10 @@ void IntegrationPluginTcpCommander::thingRemoved(Thing *thing)
void IntegrationPluginTcpCommander::onTcpSocketConnectionChanged(bool connected)
{
TcpSocket *tcpSocket = static_cast<TcpSocket *>(sender());
Thing *thing = m_tcpSockets.value(tcpSocket);
if (thing->thingClassId() == tcpOutputThingClassId) {
thing->setStateValue(tcpOutputConnectedStateTypeId, connected);
QTcpSocket *tcpSocket = static_cast<QTcpSocket *>(sender());
Thing *thing = m_tcpSockets.key(tcpSocket);
if (thing->thingClassId() == tcpClientThingClassId) {
thing->setStateValue(tcpClientConnectedStateTypeId, connected);
}
}
@ -131,29 +157,30 @@ void IntegrationPluginTcpCommander::onTcpSocketConnectionChanged(bool connected)
void IntegrationPluginTcpCommander::onTcpServerConnectionCountChanged(int connections)
{
TcpServer *tcpServer = static_cast<TcpServer *>(sender());
Thing *thing = m_tcpServer.value(tcpServer);
Thing *thing = m_tcpServers.key(tcpServer);
if (!thing)
return;
qDebug(dcTCPCommander()) << thing->name() << "Tcp Server Client connected";
if (thing->thingClassId() == tcpInputThingClassId) {
if (thing->thingClassId() == tcpServerThingClassId) {
if (connections > 0) {
thing->setStateValue(tcpInputConnectedStateTypeId, true);
thing->setStateValue(tcpServerConnectedStateTypeId, true);
} else {
thing->setStateValue(tcpInputConnectedStateTypeId, false);
thing->setStateValue(tcpServerConnectedStateTypeId, false);
}
thing->setStateValue(tcpInputConnectionCountStateTypeId, connections);
thing->setStateValue(tcpServerConnectionCountStateTypeId, connections);
}
}
void IntegrationPluginTcpCommander::onTcpServerCommandReceived(QByteArray data)
void IntegrationPluginTcpCommander::onTcpServerCommandReceived(const QString &clientIp, const QByteArray &data)
{
TcpServer *tcpServer = static_cast<TcpServer *>(sender());
Thing *thing = m_tcpServer.value(tcpServer);
Thing *thing = m_tcpServers.key(tcpServer);
qDebug(dcTCPCommander()) << thing->name() << "Message received" << data;
Event event = Event(tcpInputTriggeredEventTypeId, thing->id());
Event event = Event(tcpServerTriggeredEventTypeId, thing->id());
ParamList params;
params.append(Param(tcpInputTriggeredEventDataParamTypeId, data));
params.append(Param(tcpServerTriggeredEventDataParamTypeId, data));
params.append(Param(tcpServerTriggeredEventClientIpParamTypeId, clientIp));
event.setParams(params);
emitEvent(event);
}

7
tcpcommander/integrationplugintcpcommander.h
View File

@ -33,7 +33,6 @@
#include "integrations/integrationplugin.h"
#include "tcpserver.h"
#include "tcpsocket.h"
class IntegrationPluginTcpCommander : public IntegrationPlugin
{
@ -52,14 +51,14 @@ public:
void executeAction(ThingActionInfo *info) override;
private:
QHash<TcpSocket *, Thing *> m_tcpSockets;
QHash<TcpServer *, Thing *> m_tcpServer;
QHash<Thing*, QTcpSocket*> m_tcpSockets;
QHash<Thing*, TcpServer*> m_tcpServers;
private slots:
void onTcpSocketConnectionChanged(bool connected);
void onTcpServerConnectionCountChanged(int connections);
void onTcpServerCommandReceived(QByteArray message);
void onTcpServerCommandReceived(const QString &clientIp, const QByteArray &message);
};
#endif // INTEGRATIONPLUGINTCPCOMMANDER_H

59
tcpcommander/integrationplugintcpcommander.json
View File

@ -10,10 +10,10 @@
"thingClasses": [
{
"id": "c67d059f-694f-47cb-8e1d-9e3e6d014c1a",
"name": "tcpOutput",
"displayName": "TCP Output",
"name": "tcpClient",
"displayName": "TCP Client",
"createMethods": ["user"],
"interfaces": ["outputtrigger", "connectable"],
"interfaces": ["outputtrigger", "inputtrigger", "connectable"],
"paramTypes": [
{
"id": "2a3fcb23-931b-4ba1-b134-c49b656c76f7",
@ -49,7 +49,7 @@
"paramTypes": [
{
"id": "6604c852-6b24-4707-b8e5-1ddd8032efcc",
"name": "outputDataArea",
"name": "data",
"displayName": "Data",
"type": "QString",
"inputType": "TextArea",
@ -57,14 +57,29 @@
}
]
}
],
"eventTypes": [
{
"id": "d4cab21d-5877-465f-be3d-8fb20df0e087",
"name": "triggered",
"displayName": "Data received",
"paramTypes": [
{
"id": "9b1ae5d2-aeee-4cee-9695-e770e6f50a2c",
"name": "data",
"displayName": "Data",
"type": "QString"
}
]
}
]
},
{
"id": "bc40e84a-69c4-4fd2-beb0-bd65f19aa8ff",
"name": "tcpInput",
"displayName": "TCP Input",
"name": "tcpServer",
"displayName": "TCP Server",
"createMethods": ["user"],
"interfaces": ["inputtrigger", "connectable"],
"interfaces": ["inputtrigger", "outputtrigger", "connectable"],
"paramTypes": [
{
"id": "88eb227d-13f7-451c-babf-1b141c219fd4",
@ -105,6 +120,36 @@
"name": "data",
"displayName": "Data",
"type": "QString"
},
{
"id": "1f92bde3-4250-431c-abf4-1b3de8c23b27",
"name": "clientIp",
"displayName": "Client IP",
"type": "QString"
}
]
}
],
"actionTypes": [
{
"id": "be495c14-61a7-4292-b16f-0734b006d10f",
"name": "trigger",
"displayName": "Send Data",
"paramTypes": [
{
"id": "ae624855-0356-4859-a42b-1657543c0a5d",
"name": "data",
"displayName": "Data",
"type": "QString",
"inputType": "TextArea",
"defaultValue": ""
},
{
"id": "31182d99-26f1-48e1-8b47-38fe01822125",
"name": "clientIp",
"displayName": "Client IP",
"type": "QString",
"defaultValue": "0.0.0.0"
}
]
}

6
tcpcommander/tcpcommander.pro
View File

@ -6,10 +6,8 @@ TARGET = $$qtLibraryTarget(nymea_integrationplugintcpcommander)
SOURCES += \
integrationplugintcpcommander.cpp \
tcpserver.cpp \
tcpsocket.cpp
tcpserver.cpp
HEADERS += \
integrationplugintcpcommander.h \
tcpserver.h \
tcpsocket.h
tcpserver.h

35
tcpcommander/tcpserver.cpp
View File

@ -78,7 +78,24 @@ int TcpServer::serverPort()
int TcpServer::connectionCount()
{
return m_connectionCount;
return m_clients.count();
}
bool TcpServer::sendCommand(const QString &clientIp, const QByteArray &data)
{
bool success = false;
QHostAddress address(clientIp);
foreach (QTcpSocket *client, m_clients) {
if (address == QHostAddress(QHostAddress::AnyIPv4) || client->peerAddress() == address) {
qint64 len = client->write(data);
if (len == data.length()) {
success = true;
}
}
}
qCWarning(dcTCPCommander()) << "No client matching the destination IP" << address.toString();
return success;
}
void TcpServer::newConnection()
@ -87,8 +104,8 @@ void TcpServer::newConnection()
QTcpSocket *socket = m_tcpServer->nextPendingConnection();
socket->flush();
m_connectionCount++;
emit connectionCountChanged(m_connectionCount);
m_clients.append(socket);
emit connectionCountChanged(m_clients.count());
connect(socket, &QTcpSocket::disconnected, this, &TcpServer::onDisconnected);
connect(socket, &QTcpSocket::readyRead, this, &TcpServer::readData);
// Note: error signal will be interpreted as function, not as signal in C++11
@ -97,12 +114,10 @@ void TcpServer::newConnection()
void TcpServer::onDisconnected()
{
qDebug(dcTCPCommander()) << "TCP Server connection aborted";
m_connectionCount--;
if (m_connectionCount < 0)
m_connectionCount = 0;
emit connectionCountChanged(m_connectionCount);
QTcpSocket *client = qobject_cast<QTcpSocket*>(sender());
qDebug(dcTCPCommander()) << "TCP client disconnected";
m_clients.removeAll(client);
emit connectionCountChanged(m_clients.count());
}
void TcpServer::readData()
@ -111,7 +126,7 @@ void TcpServer::readData()
QByteArray data = socket->readAll();
qDebug(dcTCPCommander()) << "TCP Server data received: " << data;
socket->write("OK\n");
emit commandReceived(data);
emit commandReceived(socket->peerAddress().toString(), data);
}
void TcpServer::onError(QAbstractSocket::SocketError error)

16
tcpcommander/tcpserver.h
View File

@ -45,20 +45,16 @@ public:
bool isValid();
QHostAddress serverAddress();
int serverPort();
void setPort(int port);
void setServerAddress(const QHostAddress &address);
int connectionCount();
private:
QTcpServer *m_tcpServer = nullptr;
int m_connectionCount = 0;
bool sendCommand(const QString &clientIp, const QByteArray &data);
signals:
void newPendingConnection();
void commandReceived(QByteArray message);
void commandReceived(const QString &clientIp, const QByteArray &message);
void connectionCountChanged(int connections);
private slots:
@ -66,6 +62,12 @@ private slots:
void onDisconnected();
void readData();
void onError(QAbstractSocket::SocketError error);
private:
QTcpServer *m_tcpServer = nullptr;
QList<QTcpSocket*> m_clients;
};
#endif // TCPSERVER_H

123
tcpcommander/tcpsocket.cpp
View File

@ -1,123 +0,0 @@
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
*
* Copyright 2013 - 2020, nymea GmbH
* Contact: contact@nymea.io
*
* This file is part of nymea.
* This project including source code and documentation is protected by
* copyright law, and remains the property of nymea GmbH. All rights, including
* reproduction, publication, editing and translation, are reserved. The use of
* this project is subject to the terms of a license agreement to be concluded
* with nymea GmbH in accordance with the terms of use of nymea GmbH, available
* under https://nymea.io/license
*
* GNU Lesser General Public License Usage
* Alternatively, this project may be redistributed and/or modified under the
* terms of the GNU Lesser General Public License as published by the Free
* Software Foundation; version 3. This project is distributed in the hope that
* it will be useful, but WITHOUT ANY WARRANTY; without even the implied
* warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this project. If not, see <https://www.gnu.org/licenses/>.
*
* For any further details and any questions please contact us under
* contact@nymea.io or see our FAQ/Licensing Information on
* https://nymea.io/license/faq
*
* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
#include "tcpsocket.h"
#include "extern-plugininfo.h"
TcpSocket::TcpSocket(const QHostAddress address, const quint16 &port, QObject *parent) :
QObject(parent),
m_port(port),
m_address(address)
{
m_tcpSocket = new QTcpSocket(this);
connect(m_tcpSocket, &QTcpSocket::connected, this, &TcpSocket::onConnected);
connect(m_tcpSocket, &QTcpSocket::disconnected, this, &TcpSocket::onDisconnected);
connect(m_tcpSocket, &QTcpSocket::bytesWritten, this, &TcpSocket::onBytesWritten);
connect(m_tcpSocket, SIGNAL(error(QAbstractSocket::SocketError)), this, SLOT(onTcpSocketError(QAbstractSocket::SocketError)));
}
void TcpSocket::sendCommand(QByteArray command)
{
if (m_pendingCommands.isEmpty()) {
m_pendingCommands.append(command);
m_tcpSocket->abort();
m_tcpSocket->connectToHost(m_address, m_port);
} else {
m_pendingCommands.append(command);
}
}
void TcpSocket::connectionTest()
{
QTcpSocket *testSocket = new QTcpSocket(this);
connect(testSocket, &QTcpSocket::connected, this,[this, testSocket] {
emit connectionTestFinished(true);
testSocket->deleteLater();
});
connect(testSocket, static_cast<void (QTcpSocket::*) (QAbstractSocket::SocketError)>(&QTcpSocket::error), this, [this, testSocket] {
emit connectionTestFinished(false);
testSocket->deleteLater();
});
testSocket->connectToHost(m_address, m_port);
}
void TcpSocket::onConnected()
{
qDebug(dcTCPCommander()) << "Socket connected" ;
if (!m_pendingCommands.isEmpty()) {
QByteArray data = m_pendingCommands.takeLast();
qDebug(dcTCPCommander()) << "Writing data:" << data;
m_tcpSocket->write(data + "\n");
} else {
m_tcpSocket->disconnectFromHost();
}
emit connectionChanged(true);
}
void TcpSocket::onDisconnected()
{
qDebug(dcTCPCommander()) << "Socket disconnected" ;
emit connectionChanged(false);
}
void TcpSocket::onBytesWritten()
{
emit commandSent(true);
if (!m_pendingCommands.isEmpty()){
m_tcpSocket->write(m_pendingCommands.takeFirst());
} else {
m_tcpSocket->close();
}
}
void TcpSocket::onError(QAbstractSocket::SocketError error)
{
qWarning(dcTCPCommander()) << "Socket Error" << m_tcpSocket->errorString();
switch (error) {
case QAbstractSocket::RemoteHostClosedError:
break;
case QAbstractSocket::HostNotFoundError:
break;
case QAbstractSocket::ConnectionRefusedError:
break;
default:
;
}
emit commandSent(false);
emit connectionChanged(false);
m_pendingCommands.clear(); //undefined socket state needs to clear command buffer.
if (m_tcpSocket->isOpen()) {
m_tcpSocket->disconnectFromHost();
}
}

69
tcpcommander/tcpsocket.h
View File

@ -1,69 +0,0 @@
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
*
* Copyright 2013 - 2020, nymea GmbH
* Contact: contact@nymea.io
*
* This file is part of nymea.
* This project including source code and documentation is protected by
* copyright law, and remains the property of nymea GmbH. All rights, including
* reproduction, publication, editing and translation, are reserved. The use of
* this project is subject to the terms of a license agreement to be concluded
* with nymea GmbH in accordance with the terms of use of nymea GmbH, available
* under https://nymea.io/license
*
* GNU Lesser General Public License Usage
* Alternatively, this project may be redistributed and/or modified under the
* terms of the GNU Lesser General Public License as published by the Free
* Software Foundation; version 3. This project is distributed in the hope that
* it will be useful, but WITHOUT ANY WARRANTY; without even the implied
* warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this project. If not, see <https://www.gnu.org/licenses/>.
*
* For any further details and any questions please contact us under
* contact@nymea.io or see our FAQ/Licensing Information on
* https://nymea.io/license/faq
*
* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
#ifndef TCPSOCKET_H
#define TCPSOCKET_H
#include <QObject>
#include <QTcpSocket>
#include <QHostAddress>
class TcpSocket : public QObject
{
Q_OBJECT
public:
explicit TcpSocket(const QHostAddress address, const quint16 &port, QObject *parent = nullptr);
void sendCommand(QByteArray command);
void connectionTest();
private:
QTcpSocket *m_tcpSocket = nullptr;
quint16 m_port;
QHostAddress m_address;
QList<QByteArray> m_pendingCommands;
signals:
void connectionChanged(bool connected);
void commandSent(bool successfull);
void connectionTestFinished(bool successfull);
private slots:
void onConnected();
void onDisconnected();
void onBytesWritten();
void onError(QAbstractSocket::SocketError error);
};
#endif // TCPSOCKET_H