Examples: change the main network-chat protocol to CBOR
This complements the previous commit, which changed the broadcast datagram to CBOR. This commit changes the TCP protocol too. The protocol is an infinite array of commands, each of which is a map from an integer (the DataType enum) to either a string or null. The entire state machine for the connection is rewritten, relying on QCborStreamReader's ability to deal with incomplete data. Change-Id: Ic38ec929fc3f4bb795dafffd150ac674c32fac87 Reviewed-by: Mårten Nordheim <marten.nordheim@qt.io> Reviewed-by: Edward Welbourne <edward.welbourne@qt.io>bb10
Thiago Macieira
parent
c901cdadc0
commit
bfcb8c6dca
|
|
@ -1,6 +1,7 @@
|
|||
/****************************************************************************
|
||||
**
|
||||
** Copyright (C) 2016 The Qt Company Ltd.
|
||||
** Copyright (C) 2018 Intel Corporation.
|
||||
** Contact: https://www.qt.io/licensing/
|
||||
**
|
||||
** This file is part of the examples of the Qt Toolkit.
|
||||
|
|
@ -55,17 +56,29 @@
|
|||
static const int TransferTimeout = 30 * 1000;
|
||||
static const int PongTimeout = 60 * 1000;
|
||||
static const int PingInterval = 5 * 1000;
|
||||
static const char SeparatorToken = ' ';
|
||||
|
||||
/*
|
||||
* Protocol is defined as follows, using the CBOR Data Definition Language:
|
||||
*
|
||||
* protocol = [
|
||||
* greeting, ; must start with a greeting command
|
||||
* * command ; zero or more regular commands after
|
||||
* ]
|
||||
* command = plaintext / ping / pong / greeting
|
||||
* plaintext = { 0 => text }
|
||||
* ping = { 1 => null }
|
||||
* pong = { 2 => null }
|
||||
* greeting = { 3 => text }
|
||||
*/
|
||||
|
||||
Connection::Connection(QObject *parent)
|
||||
: QTcpSocket(parent)
|
||||
: QTcpSocket(parent), writer(this)
|
||||
{
|
||||
greetingMessage = tr("undefined");
|
||||
username = tr("unknown");
|
||||
state = WaitingForGreeting;
|
||||
currentDataType = Undefined;
|
||||
numBytesForCurrentDataType = -1;
|
||||
transferTimerId = 0;
|
||||
transferTimerId = -1;
|
||||
isGreetingMessageSent = false;
|
||||
pingTimer.setInterval(PingInterval);
|
||||
|
||||
|
|
@ -76,6 +89,22 @@ Connection::Connection(QObject *parent)
|
|||
this, SLOT(sendGreetingMessage()));
|
||||
}
|
||||
|
||||
Connection::Connection(qintptr socketDescriptor, QObject *parent)
|
||||
: Connection(parent)
|
||||
{
|
||||
setSocketDescriptor(socketDescriptor);
|
||||
reader.setDevice(this);
|
||||
}
|
||||
|
||||
Connection::~Connection()
|
||||
{
|
||||
if (isGreetingMessageSent) {
|
||||
// Indicate clean shutdown.
|
||||
writer.endArray();
|
||||
waitForBytesWritten(2000);
|
||||
}
|
||||
}
|
||||
|
||||
QString Connection::name() const
|
||||
{
|
||||
return username;
|
||||
|
|
@ -91,9 +120,11 @@ bool Connection::sendMessage(const QString &message)
|
|||
if (message.isEmpty())
|
||||
return false;
|
||||
|
||||
QByteArray msg = message.toUtf8();
|
||||
QByteArray data = "MESSAGE " + QByteArray::number(msg.size()) + ' ' + msg;
|
||||
return write(data) == data.size();
|
||||
writer.startMap(1);
|
||||
writer.append(PlainText);
|
||||
writer.append(message);
|
||||
writer.endMap();
|
||||
return true;
|
||||
}
|
||||
|
||||
void Connection::timerEvent(QTimerEvent *timerEvent)
|
||||
|
|
@ -101,61 +132,75 @@ void Connection::timerEvent(QTimerEvent *timerEvent)
|
|||
if (timerEvent->timerId() == transferTimerId) {
|
||||
abort();
|
||||
killTimer(transferTimerId);
|
||||
transferTimerId = 0;
|
||||
transferTimerId = -1;
|
||||
}
|
||||
}
|
||||
|
||||
void Connection::processReadyRead()
|
||||
{
|
||||
if (state == WaitingForGreeting) {
|
||||
if (!readProtocolHeader())
|
||||
return;
|
||||
if (currentDataType != Greeting) {
|
||||
abort();
|
||||
return;
|
||||
}
|
||||
state = ReadingGreeting;
|
||||
}
|
||||
// we've got more data, let's parse
|
||||
reader.reparse();
|
||||
while (reader.lastError() == QCborError::NoError) {
|
||||
if (state == WaitingForGreeting) {
|
||||
if (!reader.isArray())
|
||||
break; // protocol error
|
||||
|
||||
if (state == ReadingGreeting) {
|
||||
if (!hasEnoughData())
|
||||
return;
|
||||
|
||||
buffer = read(numBytesForCurrentDataType);
|
||||
if (buffer.size() != numBytesForCurrentDataType) {
|
||||
abort();
|
||||
return;
|
||||
}
|
||||
|
||||
username = QString(buffer) + '@' + peerAddress().toString() + ':'
|
||||
+ QString::number(peerPort());
|
||||
currentDataType = Undefined;
|
||||
numBytesForCurrentDataType = 0;
|
||||
buffer.clear();
|
||||
|
||||
if (!isValid()) {
|
||||
abort();
|
||||
return;
|
||||
}
|
||||
|
||||
if (!isGreetingMessageSent)
|
||||
sendGreetingMessage();
|
||||
|
||||
pingTimer.start();
|
||||
pongTime.start();
|
||||
state = ReadyForUse;
|
||||
emit readyForUse();
|
||||
}
|
||||
|
||||
do {
|
||||
if (currentDataType == Undefined) {
|
||||
if (!readProtocolHeader())
|
||||
reader.enterContainer(); // we'll be in this array forever
|
||||
state = ReadingGreeting;
|
||||
} else if (reader.containerDepth() == 1) {
|
||||
// Current state: no command read
|
||||
// Next state: read command ID
|
||||
if (!reader.hasNext()) {
|
||||
reader.leaveContainer();
|
||||
disconnectFromHost();
|
||||
return;
|
||||
}
|
||||
|
||||
if (!reader.isMap() || !reader.isLengthKnown() || reader.length() != 1)
|
||||
break; // protocol error
|
||||
reader.enterContainer();
|
||||
} else if (currentDataType == Undefined) {
|
||||
// Current state: read command ID
|
||||
// Next state: read command payload
|
||||
if (!reader.isInteger())
|
||||
break; // protocol error
|
||||
currentDataType = DataType(reader.toInteger());
|
||||
reader.next();
|
||||
} else {
|
||||
// Current state: read command payload
|
||||
if (reader.isString()) {
|
||||
auto r = reader.readString();
|
||||
buffer += r.data;
|
||||
if (r.status != QCborStreamReader::EndOfString)
|
||||
continue;
|
||||
} else if (reader.isNull()) {
|
||||
reader.next();
|
||||
} else {
|
||||
break; // protocol error
|
||||
}
|
||||
|
||||
// Next state: no command read
|
||||
reader.leaveContainer();
|
||||
if (transferTimerId != -1) {
|
||||
killTimer(transferTimerId);
|
||||
transferTimerId = -1;
|
||||
}
|
||||
|
||||
if (state == ReadingGreeting) {
|
||||
if (currentDataType != Greeting)
|
||||
break; // protocol error
|
||||
processGreeting();
|
||||
} else {
|
||||
processData();
|
||||
}
|
||||
}
|
||||
if (!hasEnoughData())
|
||||
return;
|
||||
processData();
|
||||
} while (bytesAvailable() > 0);
|
||||
}
|
||||
|
||||
if (reader.lastError() != QCborError::EndOfFile)
|
||||
abort(); // parse error
|
||||
|
||||
if (transferTimerId != -1 && reader.containerDepth() > 1)
|
||||
transferTimerId = startTimer(TransferTimeout);
|
||||
}
|
||||
|
||||
void Connection::sendPing()
|
||||
|
|
@ -165,112 +210,58 @@ void Connection::sendPing()
|
|||
return;
|
||||
}
|
||||
|
||||
write("PING 1 p");
|
||||
writer.startMap(1);
|
||||
writer.append(Ping);
|
||||
writer.append(nullptr); // no payload
|
||||
writer.endMap();
|
||||
}
|
||||
|
||||
void Connection::sendGreetingMessage()
|
||||
{
|
||||
QByteArray greeting = greetingMessage.toUtf8();
|
||||
QByteArray data = "GREETING " + QByteArray::number(greeting.size()) + ' ' + greeting;
|
||||
if (write(data) == data.size())
|
||||
isGreetingMessageSent = true;
|
||||
writer.startArray(); // this array never ends
|
||||
|
||||
writer.startMap(1);
|
||||
writer.append(Greeting);
|
||||
writer.append(greetingMessage);
|
||||
writer.endMap();
|
||||
isGreetingMessageSent = true;
|
||||
|
||||
if (!reader.device())
|
||||
reader.setDevice(this);
|
||||
}
|
||||
|
||||
int Connection::readDataIntoBuffer(int maxSize)
|
||||
void Connection::processGreeting()
|
||||
{
|
||||
if (maxSize > MaxBufferSize)
|
||||
return 0;
|
||||
|
||||
int numBytesBeforeRead = buffer.size();
|
||||
if (numBytesBeforeRead == MaxBufferSize) {
|
||||
abort();
|
||||
return 0;
|
||||
}
|
||||
|
||||
while (bytesAvailable() > 0 && buffer.size() < maxSize) {
|
||||
buffer.append(read(1));
|
||||
if (buffer.endsWith(SeparatorToken))
|
||||
break;
|
||||
}
|
||||
return buffer.size() - numBytesBeforeRead;
|
||||
}
|
||||
|
||||
int Connection::dataLengthForCurrentDataType()
|
||||
{
|
||||
if (bytesAvailable() <= 0 || readDataIntoBuffer() <= 0
|
||||
|| !buffer.endsWith(SeparatorToken))
|
||||
return 0;
|
||||
|
||||
buffer.chop(1);
|
||||
int number = buffer.toInt();
|
||||
username = buffer + '@' + peerAddress().toString() + ':'
|
||||
+ QString::number(peerPort());
|
||||
currentDataType = Undefined;
|
||||
buffer.clear();
|
||||
return number;
|
||||
}
|
||||
|
||||
bool Connection::readProtocolHeader()
|
||||
{
|
||||
if (transferTimerId) {
|
||||
killTimer(transferTimerId);
|
||||
transferTimerId = 0;
|
||||
}
|
||||
|
||||
if (readDataIntoBuffer() <= 0) {
|
||||
transferTimerId = startTimer(TransferTimeout);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (buffer == "PING ") {
|
||||
currentDataType = Ping;
|
||||
} else if (buffer == "PONG ") {
|
||||
currentDataType = Pong;
|
||||
} else if (buffer == "MESSAGE ") {
|
||||
currentDataType = PlainText;
|
||||
} else if (buffer == "GREETING ") {
|
||||
currentDataType = Greeting;
|
||||
} else {
|
||||
currentDataType = Undefined;
|
||||
abort();
|
||||
return false;
|
||||
}
|
||||
|
||||
buffer.clear();
|
||||
numBytesForCurrentDataType = dataLengthForCurrentDataType();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Connection::hasEnoughData()
|
||||
{
|
||||
if (transferTimerId) {
|
||||
QObject::killTimer(transferTimerId);
|
||||
transferTimerId = 0;
|
||||
}
|
||||
|
||||
if (numBytesForCurrentDataType <= 0)
|
||||
numBytesForCurrentDataType = dataLengthForCurrentDataType();
|
||||
|
||||
if (bytesAvailable() < numBytesForCurrentDataType
|
||||
|| numBytesForCurrentDataType <= 0) {
|
||||
transferTimerId = startTimer(TransferTimeout);
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
void Connection::processData()
|
||||
{
|
||||
buffer = read(numBytesForCurrentDataType);
|
||||
if (buffer.size() != numBytesForCurrentDataType) {
|
||||
if (!isValid()) {
|
||||
abort();
|
||||
return;
|
||||
}
|
||||
|
||||
if (!isGreetingMessageSent)
|
||||
sendGreetingMessage();
|
||||
|
||||
pingTimer.start();
|
||||
pongTime.start();
|
||||
state = ReadyForUse;
|
||||
emit readyForUse();
|
||||
}
|
||||
|
||||
void Connection::processData()
|
||||
{
|
||||
switch (currentDataType) {
|
||||
case PlainText:
|
||||
emit newMessage(username, QString::fromUtf8(buffer));
|
||||
emit newMessage(username, buffer);
|
||||
break;
|
||||
case Ping:
|
||||
write("PONG 1 p");
|
||||
writer.startMap(1);
|
||||
writer.append(Pong);
|
||||
writer.append(nullptr); // no payload
|
||||
writer.endMap();
|
||||
break;
|
||||
case Pong:
|
||||
pongTime.restart();
|
||||
|
|
@ -280,6 +271,5 @@ void Connection::processData()
|
|||
}
|
||||
|
||||
currentDataType = Undefined;
|
||||
numBytesForCurrentDataType = 0;
|
||||
buffer.clear();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -51,10 +51,12 @@
|
|||
#ifndef CONNECTION_H
|
||||
#define CONNECTION_H
|
||||
|
||||
#include <QCborStreamReader>
|
||||
#include <QCborStreamWriter>
|
||||
#include <QElapsedTimer>
|
||||
#include <QHostAddress>
|
||||
#include <QString>
|
||||
#include <QTcpSocket>
|
||||
#include <QTime>
|
||||
#include <QTimer>
|
||||
|
||||
static const int MaxBufferSize = 1024000;
|
||||
|
|
@ -78,6 +80,8 @@ public:
|
|||
};
|
||||
|
||||
Connection(QObject *parent = 0);
|
||||
Connection(qintptr socketDescriptor, QObject *parent = 0);
|
||||
~Connection();
|
||||
|
||||
QString name() const;
|
||||
void setGreetingMessage(const QString &message);
|
||||
|
|
@ -96,20 +100,19 @@ private slots:
|
|||
void sendGreetingMessage();
|
||||
|
||||
private:
|
||||
int readDataIntoBuffer(int maxSize = MaxBufferSize);
|
||||
int dataLengthForCurrentDataType();
|
||||
bool readProtocolHeader();
|
||||
bool hasEnoughData();
|
||||
void processGreeting();
|
||||
void processData();
|
||||
|
||||
QCborStreamReader reader;
|
||||
QCborStreamWriter writer;
|
||||
QString greetingMessage;
|
||||
QString username;
|
||||
QTimer pingTimer;
|
||||
QTime pongTime;
|
||||
QByteArray buffer;
|
||||
QElapsedTimer pongTime;
|
||||
QString buffer;
|
||||
ConnectionState state;
|
||||
DataType currentDataType;
|
||||
int numBytesForCurrentDataType;
|
||||
int transferTimerId;
|
||||
bool isGreetingMessageSent;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -61,7 +61,6 @@ Server::Server(QObject *parent)
|
|||
|
||||
void Server::incomingConnection(qintptr socketDescriptor)
|
||||
{
|
||||
Connection *connection = new Connection(this);
|
||||
connection->setSocketDescriptor(socketDescriptor);
|
||||
Connection *connection = new Connection(socketDescriptor, this);
|
||||
emit newConnection(connection);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue