1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
|
// Copyright (C) 2017 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
#ifndef QMODBUSTCPCLIENT_P_H
#define QMODBUSTCPCLIENT_P_H
#include <QtCore/qloggingcategory.h>
#include <QtNetwork/qhostaddress.h>
#include <QtNetwork/qtcpsocket.h>
#include "QtSerialBus/qmodbustcpclient.h"
#include "private/qmodbusclient_p.h"
//
// W A R N I N G
// -------------
//
// This file is not part of the Qt API. It exists purely as an
// implementation detail. This header file may change from version to
// version without notice, or even be removed.
//
// We mean it.
//
QT_BEGIN_NAMESPACE
Q_DECLARE_LOGGING_CATEGORY(QT_MODBUS)
Q_DECLARE_LOGGING_CATEGORY(QT_MODBUS_LOW)
class QModbusTcpClientPrivate : public QModbusClientPrivate
{
Q_DECLARE_PUBLIC(QModbusTcpClient)
public:
void setupTcpSocket()
{
Q_Q(QModbusTcpClient);
m_socket = new QTcpSocket(q);
QObject::connect(m_socket, &QAbstractSocket::connected, q, [this]() {
qCDebug(QT_MODBUS) << "(TCP client) Connected to" << m_socket->peerAddress()
<< "on port" << m_socket->peerPort();
Q_Q(QModbusTcpClient);
responseBuffer.clear();
q->setState(QModbusDevice::ConnectedState);
});
QObject::connect(m_socket, &QAbstractSocket::disconnected, q, [this]() {
qCDebug(QT_MODBUS) << "(TCP client) Connection closed.";
Q_Q(QModbusTcpClient);
q->setState(QModbusDevice::UnconnectedState);
cleanupTransactionStore();
});
QObject::connect(m_socket, &QAbstractSocket::errorOccurred, q,
[this](QAbstractSocket::SocketError /*error*/)
{
Q_Q(QModbusTcpClient);
if (m_socket->state() == QAbstractSocket::UnconnectedState) {
cleanupTransactionStore();
q->setState(QModbusDevice::UnconnectedState);
}
q->setError(QModbusClient::tr("TCP socket error (%1).").arg(m_socket->errorString()),
QModbusDevice::ConnectionError);
});
QObject::connect(m_socket, &QIODevice::readyRead, q, [this](){
responseBuffer += m_socket->read(m_socket->bytesAvailable());
qCDebug(QT_MODBUS_LOW) << "(TCP client) Response buffer:" << responseBuffer.toHex();
while (!responseBuffer.isEmpty()) {
// can we read enough for Modbus ADU header?
if (responseBuffer.size() < mbpaHeaderSize) {
qCDebug(QT_MODBUS_LOW) << "(TCP client) MBPA header too short. Waiting for more data.";
return;
}
quint8 serverAddress;
quint16 transactionId, bytesPdu, protocolId;
QDataStream input(responseBuffer);
input >> transactionId >> protocolId >> bytesPdu >> serverAddress;
// stop the timer as soon as we know enough about the transaction
const bool knownTransaction = m_transactionStore.contains(transactionId);
if (knownTransaction && m_transactionStore[transactionId].timer)
m_transactionStore[transactionId].timer->stop();
qCDebug(QT_MODBUS) << "(TCP client) tid:" << Qt::hex << transactionId << "size:"
<< bytesPdu << "server address:" << serverAddress;
// The length field is the byte count of the following fields, including the Unit
// Identifier and the PDU, so we remove on byte.
bytesPdu--;
int tcpAduSize = mbpaHeaderSize + bytesPdu;
if (responseBuffer.size() < tcpAduSize) {
qCDebug(QT_MODBUS) << "(TCP client) PDU too short. Waiting for more data";
return;
}
QModbusResponse responsePdu;
input >> responsePdu;
qCDebug(QT_MODBUS) << "(TCP client) Received PDU:" << responsePdu.functionCode()
<< responsePdu.data().toHex();
responseBuffer.remove(0, tcpAduSize);
if (!knownTransaction) {
qCDebug(QT_MODBUS) << "(TCP client) No pending request for response with "
"given transaction ID, ignoring response message.";
} else {
processQueueElement(responsePdu, m_transactionStore[transactionId]);
}
}
});
}
QModbusReply *enqueueRequest(const QModbusRequest &request, int serverAddress,
const QModbusDataUnit &unit,
QModbusReply::ReplyType type) override
{
auto writeToSocket = [this](quint16 tId, const QModbusRequest &request, int address) {
QByteArray buffer;
QDataStream output(&buffer, QIODevice::WriteOnly);
output << tId << quint16(0) << quint16(request.size() + 1) << quint8(address) << request;
int writtenBytes = m_socket->write(buffer);
if (writtenBytes == -1 || writtenBytes < buffer.size()) {
Q_Q(QModbusTcpClient);
qCDebug(QT_MODBUS) << "(TCP client) Cannot write request to socket.";
q->setError(QModbusTcpClient::tr("Could not write request to socket."),
QModbusDevice::WriteError);
return false;
}
qCDebug(QT_MODBUS_LOW) << "(TCP client) Sent TCP ADU:" << buffer.toHex();
qCDebug(QT_MODBUS) << "(TCP client) Sent TCP PDU:" << request << "with tId:" <<Qt:: hex
<< tId;
return true;
};
const int tId = transactionId();
if (!writeToSocket(tId, request, serverAddress))
return nullptr;
Q_Q(QModbusTcpClient);
auto reply = new QModbusReply(type, serverAddress, q);
const auto element = QueueElement{ reply, request, unit, m_numberOfRetries,
m_responseTimeoutDuration };
m_transactionStore.insert(tId, element);
q->connect(reply, &QObject::destroyed, q, [this, tId](QObject *) {
if (!m_transactionStore.contains(tId))
return;
const QueueElement element = m_transactionStore.take(tId);
if (element.timer)
element.timer->stop();
});
if (element.timer) {
q->connect(q, &QModbusClient::timeoutChanged,
element.timer.data(), QOverload<int>::of(&QTimer::setInterval));
QObject::connect(element.timer.data(), &QTimer::timeout, q, [this, writeToSocket, tId]() {
if (!m_transactionStore.contains(tId))
return;
QueueElement elem = m_transactionStore.take(tId);
if (elem.reply.isNull())
return;
if (elem.numberOfRetries > 0) {
elem.numberOfRetries--;
if (!writeToSocket(tId, elem.requestPdu, elem.reply->serverAddress()))
return;
m_transactionStore.insert(tId, elem);
elem.timer->start();
qCDebug(QT_MODBUS) << "(TCP client) Resend request with tId:" << Qt::hex << tId;
} else {
qCDebug(QT_MODBUS) << "(TCP client) Timeout of request with tId:" <<Qt::hex << tId;
elem.reply->setError(QModbusDevice::TimeoutError,
QModbusClient::tr("Request timeout."));
}
});
element.timer->start();
} else {
qCWarning(QT_MODBUS) << "(TCP client) No response timeout timer for request with tId:"
<< Qt::hex << tId << ". Expected timeout:" << m_responseTimeoutDuration;
}
incrementTransactionId();
return reply;
}
// TODO: Review once we have a transport layer in place.
bool isOpen() const override
{
if (m_socket)
return m_socket->isOpen();
return false;
}
void cleanupTransactionStore()
{
if (m_transactionStore.isEmpty())
return;
qCDebug(QT_MODBUS) << "(TCP client) Cleanup of pending requests";
for (const auto &elem : std::as_const(m_transactionStore)) {
if (elem.reply.isNull())
continue;
elem.reply->setError(QModbusDevice::ReplyAbortedError,
QModbusClient::tr("Reply aborted due to connection closure."));
}
m_transactionStore.clear();
}
// This doesn't overflow, it rather "wraps around". Expected.
inline void incrementTransactionId() { m_transactionId++; }
inline int transactionId() const { return m_transactionId; }
QIODevice *device() const override { return m_socket; }
QTcpSocket *m_socket = nullptr;
QByteArray responseBuffer;
QHash<quint16, QueueElement> m_transactionStore;
int mbpaHeaderSize = 7;
private: // Private to avoid using the wrong id inside the timer lambda,
quint16 m_transactionId = 0; // capturing 'this' will not copy the id.
};
QT_END_NAMESPACE
#endif // QMODBUSTCPCLIENT_P_H
|
