QTcpSocket: reading and writing

Question

I know some similar questions may have been asked already, but the answers to those I found covered very specific problems and I still haven't figured it out.

In my program I'm creating a QObject (called QPeer) that uses a QTcpSocket to communicate with another such object over a network. QPeer has a slot that accepts a QByteArray with data (sendData(QByteArray)). The entire contents of that array are considered to be one 'message' and they are written to the socket. I want to do the following: every time a message is written, I want the receiving QPeer to emit its signal dataReceived(QByteArray) exactly once, that QByteArray containing the entire message. (NOTE: all signals/slots, both private ones connecting the QPeer with its socket and the public ones such as sendData(QByteArray) are serialized by using Qt::QueuedConnection whenever necessary.)

I use the signal QTcpSocket::readyRead() for asynchronous reading from the socket. Now I know I can't just call QTcpSocket::write() once in sendData and then assume that for every write I do, the QTcpSocket on the other side produces exactly one readyRead signal. So what should I do?

This is my idea, please tell me if this will work:

WRITING:

void QPeer::sendData(QByteArray data)
{
    // TODO: write data.size() as raw int of exactly 4 bytes to socket
    const char *bytes = data.constData();
    int bytesWritten = 0;
    while (bytesWritten < data.size())
        bytesWritten += _socket->write(bytes + bytesWritten);
}

READING:

now I want the read function (connected to QTcpSocket::readyRead()) to use the header (the 4 byte int specifying the length of the message) and then read that amount of bytes; next emit dataReceived with exactly those bytes. I'm having serious trouble trying to do this. For example: what to do if readyRead is emitted and I can read the header of a message, but not the amount of bytes specified? Or what if a header has only been received partially?

1. How do I correctly write the header (4 byte int) to the socket?

2. How do I correctly implement the read function so that it does what I want?

Any tips are welcome. Thanks!

Solution 1

I worked on a project that does what you expect, see here the solution that I developed to our problems, simplified to be easier to understand:

Edited, added support to the server deal with multiple clients.

Client.h:

#include <QtCore>
#include <QtNetwork>

class Client : public QObject
{
    Q_OBJECT
public:
    explicit Client(QObject *parent = 0);

public slots:
    bool connectToHost(QString host);
    bool writeData(QByteArray data);

private:
    QTcpSocket *socket;
};

Client.cpp:

#include "client.h"

static inline QByteArray IntToArray(qint32 source);

Client::Client(QObject *parent) : QObject(parent)
{
    socket = new QTcpSocket(this);
}

bool Client::connectToHost(QString host)
{
    socket->connectToHost(host, 1024);
    return socket->waitForConnected();
}

bool Client::writeData(QByteArray data)
{
    if(socket->state() == QAbstractSocket::ConnectedState)
    {
        socket->write(IntToArray(data.size())); //write size of data
        socket->write(data); //write the data itself
        return socket->waitForBytesWritten();
    }
    else
        return false;
}

QByteArray IntToArray(qint32 source) //Use qint32 to ensure that the number have 4 bytes
{
    //Avoid use of cast, this is the Qt way to serialize objects
    QByteArray temp;
    QDataStream data(&temp, QIODevice::ReadWrite);
    data << source;
    return temp;
}

Server.h:

#include <QtCore>
#include <QtNetwork>

class Server : public QObject
{
    Q_OBJECT
public:
    explicit Server(QObject *parent = 0);

signals:
    void dataReceived(QByteArray);

private slots:
    void newConnection();
    void disconnected();
    void readyRead();

private:
    QTcpServer *server;
    QHash<QTcpSocket*, QByteArray*> buffers; //We need a buffer to store data until block has completely received
    QHash<QTcpSocket*, qint32*> sizes; //We need to store the size to verify if a block has received completely
};

Server.cpp:

#include "server.h"

static inline qint32 ArrayToInt(QByteArray source);

Server::Server(QObject *parent) : QObject(parent)
{
    server = new QTcpServer(this);
    connect(server, SIGNAL(newConnection()), SLOT(newConnection()));
    qDebug() << "Listening:" << server->listen(QHostAddress::Any, 1024);
}

void Server::newConnection()
{
    while (server->hasPendingConnections())
    {
        QTcpSocket *socket = server->nextPendingConnection();
        connect(socket, SIGNAL(readyRead()), SLOT(readyRead()));
        connect(socket, SIGNAL(disconnected()), SLOT(disconnected()));
        QByteArray *buffer = new QByteArray();
        qint32 *s = new qint32(0);
        buffers.insert(socket, buffer);
        sizes.insert(socket, s);
    }
}

void Server::disconnected()
{
    QTcpSocket *socket = static_cast<QTcpSocket*>(sender());
    QByteArray *buffer = buffers.value(socket);
    qint32 *s = sizes.value(socket);
    socket->deleteLater();
    delete buffer;
    delete s;
}

void Server::readyRead()
{
    QTcpSocket *socket = static_cast<QTcpSocket*>(sender());
    QByteArray *buffer = buffers.value(socket);
    qint32 *s = sizes.value(socket);
    qint32 size = *s;
    while (socket->bytesAvailable() > 0)
    {
        buffer->append(socket->readAll());
        while ((size == 0 && buffer->size() >= 4) || (size > 0 && buffer->size() >= size)) //While can process data, process it
        {
            if (size == 0 && buffer->size() >= 4) //if size of data has received completely, then store it on our global variable
            {
                size = ArrayToInt(buffer->mid(0, 4));
                *s = size;
                buffer->remove(0, 4);
            }
            if (size > 0 && buffer->size() >= size) // If data has received completely, then emit our SIGNAL with the data
            {
                QByteArray data = buffer->mid(0, size);
                buffer->remove(0, size);
                size = 0;
                *s = size;
                emit dataReceived(data);
            }
        }
    }
}

qint32 ArrayToInt(QByteArray source)
{
    qint32 temp;
    QDataStream data(&source, QIODevice::ReadWrite);
    data >> temp;
    return temp;
}

Note: Do not use this method to transfer large files because with this method the entire contents of the message is put inside the memory before sent and this causes a high memory usage. And because 32 bits signed INT has max value to 2,147,483,647, if your input data has a value higher than that in bytes it won't work. Take care.

Solution 2

As you said, you need to wait that your header is entirely sent, before reading it, and then read the good number of bytes and emit a signal for data availability.

Here an example (untested) :

//header file

class Peer {
//[...]
protected:
   bool m_headerRead; //initialize to false
   unsigned int m_size_of_data_to_read;
//[...]
};

//source file
void QPeer::sendData(QByteArray data)
{
  int size = data.size();
  _socket->write((const char*) &size, sizeof(int);
  //use directly QIODevice::write(QByteArray)
  _socket->write(data);
}

void QPeer::readData()
{
    int bytes = _socket->bytesAvailable();
    bool contains_enough_data = true;

    while (contains_enough_data) {
       if (! m_headerRead && _socket->bytesAvailable() >= sizeof(int)) {
         //read header only and update m_size_of_data_to_read
         m_headerRead = true;
        } else if (m_headerRead && _socket->bytesAvailable >= m_size_of_data_to_read) {
          //read data here
          m_headerRead = false;
          emit dataAvailable();
       } else {
           contains_enough_data = false; //wait that further data arrived
       }
    }
}

Comments

[DiscobarMolokai]

So you're saying that _socket->write(data) guarantees that the entire QByteArray is written at once? Because Qt documentation says this: qint64 QIODevice::write(const QByteArray & byteArray) This is an overloaded function. Writes the content of byteArray to the device. Returns the number of bytes that were actually written, or -1 if an error occurred. So it says it returns the amount of bytes actually written. Why would it do that if the entire array would be written all at once at all times?

[epsilon]

Good point, I always use this method when I have a QByteArray to write and check for error on -1 result. Your previous way to do was not bad, you can safely replace it with this one. I will search if a QByteArray is always completely written.

[DiscobarMolokai]

I don't check for -1 error; but I have connected the error signal of the socket to a slot in QPeer that handles errors. Is this enough? Thanks for the help! Your implementation works fine, I haven't tested it thoroughly but it's fine so far, thank you.

[epsilon]

I don't think you will have error signal for such error. The signal is more related on independant errors (such as DNS failure, physic disconnection, etc..), see AbstractSocket::Error. This is probably safer to handle -1 return code ;)

[DiscobarMolokai]

Many thanks! The previous answer worked, but this made me reconsider my class design. It was all becoming a bit... nasty.

[DiscobarMolokai]

A quick question, though: when your server gets its first client (newConnection is called for the first time), the socket member contains a pointer to the socket for that connection. But what if a few moments later a new incoming connection is encountered, and newConnection is called again. Then the previous socket is lost, right? Or will the signal/slot mechanism still work on that socket?

[Antonio Dias]

The signal will be emitted because the socket is a pointer, it's won't be destroyed at the end of the scope, except if you call delete or use a smart pointer, however you can no longer receive data from that connection because the readyRead() method will work only with the pointer stored on our variable(the last connected client) To work with multiple clients you will need create multiple instances of a class containing your own socket pointer and the respective buffer.

[Antonio Dias]

That is one of some ways to deal with multiple clients.

[Antonio Dias]

I've edited the answer to add support to multiple clients with no need to create multiple instances of the class.

[DiscobarMolokai]

Thanks. I basically did the same thing, but I instantiate a clienthandler for every new connection, give it the socket handle and store the handler in a QSet (key is simply the pointer to the clienthandle). In the clienthandler I have a buffer etc. Works fine now!

[DiscobarMolokai]

By the way, does your implementation guarantee that the handling of signals emitted by the various sockets is serialized? For example, what happens if two sockets receive data from their resp. peers at the same time?

[Antonio Dias]

My implementation is guaranteed to handle multiple sockets at the same time because for all readyRead() SIGNALS the program will wait until the thread is free to to handle that signal, sometimes you need to wait to receive incoming data, but it will come.

[DiscobarMolokai]

Oh right, it all works in one thread. Thanks, I understand everything now.

[ajeh]

You should be creating the buffers before connecting the readyRead slot! Otherwise your handler may be called before you have created the buffer and you would get an AV.

[Antonio Dias]

@ajeh: Firstly, thanks for supporting my answer! But I don't believe that will be a problem, because the thread are not able to process anything until the end of the function, and when the function goes to the end, the buffer was already created. If you think i'm wrong, please correct me!

[Agnel Kurian]

@user2014561 Why are you using a QHash<QTcpSocket*, qint32*> instead of a QHash<QTcpSocket*, qint32> ?

[Antonio Dias]

@Agnel Kurian: The reason is to prevent multiple lookups for each read/write operation, that way the lookup is done only once for each function call.