qsctpsocket.cpp

Go to the documentation of this file.

// Copyright (C) 2016 Alex Trotsenko <alex1973tr@gmail.com>
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
 
//#define QSCTPSOCKET_DEBUG
 
#include "qsctpsocket.h"
#include "qsctpsocket_p.h"
 
#include "qabstractsocketengine_p.h"
#include "private/qbytearray_p.h"
 
#ifdef QSCTPSOCKET_DEBUG
#include <qdebug.h>
#endif
 
QT_BEGIN_NAMESPACE
 
QSctpSocketPrivate::QSctpSocketPrivate()
    : maximumChannelCount(0)
{
}
 
QSctpSocketPrivate::~QSctpSocketPrivate()
{
}
 
bool QSctpSocketPrivate::canReadNotification()
{
    Q_Q(QSctpSocket);
#if defined (QSCTPSOCKET_DEBUG)
    qDebug("QSctpSocketPrivate::canReadNotification()");
#endif
 
    // Handle TCP emulation mode in the base implementation.
    if (!q->isInDatagramMode())
        return QTcpSocketPrivate::canReadNotification();
 
    const int savedCurrentChannel = currentReadChannel;
    bool currentChannelRead = false;
    do {
        qsizetype datagramSize = incomingDatagram.size();
        QIpPacketHeader header;
 
        do {
            // Determine the size of the pending datagram.
            qint64 bytesToRead = socketEngine->bytesAvailable();
            if (bytesToRead == 0) {
                // As a corner case, if we can't determine the size of the pending datagram,
                // try to read 4K of data from the socket. Subsequent ::recvmsg call either
                // fails or returns the actual length of the datagram.
                bytesToRead = 4096;
            }
 
            Q_ASSERT((datagramSize + qsizetype(bytesToRead)) < MaxByteArraySize);
            incomingDatagram.resize(datagramSize + int(bytesToRead));
 
#if defined (QSCTPSOCKET_DEBUG)
            qDebug("QSctpSocketPrivate::canReadNotification() about to read %lli bytes",
                   bytesToRead);
#endif
            qint64 readBytes = socketEngine->readDatagram(
                        incomingDatagram.data() + datagramSize, bytesToRead, &header,
                        QAbstractSocketEngine::WantAll);
            if (readBytes <= 0) {
                if (readBytes == -2) { // no data available for reading
                    incomingDatagram.resize(datagramSize);
                    return currentChannelRead;
                }
 
                socketEngine->close();
                if (readBytes == 0) {
                    setErrorAndEmit(QAbstractSocket::RemoteHostClosedError,
                                    QSctpSocket::tr("The remote host closed the connection"));
                } else {
#if defined (QSCTPSOCKET_DEBUG)
                    qDebug("QSctpSocketPrivate::canReadNotification() read failed: %s",
                           socketEngine->errorString().toLatin1().constData());
#endif
                    setErrorAndEmit(socketEngine->error(), socketEngine->errorString());
                }
 
#if defined (QSCTPSOCKET_DEBUG)
                qDebug("QSctpSocketPrivate::canReadNotification() disconnecting socket");
#endif
                q->disconnectFromHost();
                return currentChannelRead;
            }
            datagramSize += int(readBytes); // update datagram size
        } while (!header.endOfRecord);
 
#if defined (QSCTPSOCKET_DEBUG)
        qDebug("QSctpSocketPrivate::canReadNotification() got datagram from channel %i, size = %i",
               header.streamNumber, datagramSize);
#endif
 
        // Drop the datagram, if opened only for writing
        if (!q->isReadable()) {
            incomingDatagram.clear();
            continue;
        }
 
        // Store datagram in the channel buffer
        Q_ASSERT(header.streamNumber < readBuffers.size());
        incomingDatagram.resize(datagramSize);
        readBuffers[header.streamNumber].setChunkSize(0); // set packet mode on channel buffer
        readBuffers[header.streamNumber].append(incomingDatagram);
        incomingDatagram = QByteArray();
 
        if (readHeaders.size() != readBuffers.size())
            readHeaders.resize(readBuffers.size());
        readHeaders[header.streamNumber].push_back(header);
 
        // Emit notifications.
        if (header.streamNumber == savedCurrentChannel)
            currentChannelRead = true;
        emitReadyRead(header.streamNumber);
 
    } while (state == QAbstractSocket::ConnectedState);
 
    return currentChannelRead;
}
 
bool QSctpSocketPrivate::writeToSocket()
{
    Q_Q(QSctpSocket);
#if defined (QSCTPSOCKET_DEBUG)
    qDebug("QSctpSocketPrivate::writeToSocket()");
#endif
 
    // Handle TCP emulation mode in the base implementation.
    if (!q->isInDatagramMode())
        return QTcpSocketPrivate::writeToSocket();
 
    if (!socketEngine)
        return false;
 
    QIpPacketHeader defaultHeader;
    const int savedCurrentChannel = currentWriteChannel;
    bool currentChannelWritten = false;
    bool transmitting;
    do {
        transmitting = false;
 
        for (int channel = 0; channel < writeBuffers.size(); ++channel) {
            QRingBuffer &channelBuffer = writeBuffers[channel];
 
            if (channelBuffer.isEmpty())
                continue;
 
            const bool hasHeader = (channel < writeHeaders.size())
                                   && !writeHeaders[channel].empty();
            QIpPacketHeader &header = hasHeader ? writeHeaders[channel].front() : defaultHeader;
            header.streamNumber = channel;
            qint64 sent = socketEngine->writeDatagram(channelBuffer.readPointer(),
                                                      channelBuffer.nextDataBlockSize(),
                                                      header);
            if (sent < 0) {
                if (sent == -2) // temporary error in writeDatagram
                    return currentChannelWritten;
 
                socketEngine->close();
#if defined (QSCTPSOCKET_DEBUG)
                qDebug("QSctpSocketPrivate::writeToSocket() write error, aborting. %s",
                       socketEngine->errorString().toLatin1().constData());
#endif
                setErrorAndEmit(socketEngine->error(), socketEngine->errorString());
                // An unexpected error so close the socket.
                q->disconnectFromHost();
                return currentChannelWritten;
            }
            Q_ASSERT(sent == channelBuffer.nextDataBlockSize());
#if defined (QSCTPSOCKET_DEBUG)
            qDebug("QSctpSocketPrivate::writeToSocket() sent datagram of size %lli to channel %i",
                   sent, channel);
#endif
            transmitting = true;
 
            // Remove datagram from the buffer
            channelBuffer.read();
            if (hasHeader)
                writeHeaders[channel].pop_front();
 
            // Emit notifications.
            if (channel == savedCurrentChannel)
                currentChannelWritten = true;
            emitBytesWritten(sent, channel);
 
            // If we were closed as a result of the bytesWritten() signal, return.
            if (state == QAbstractSocket::UnconnectedState) {
#if defined (QSCTPSOCKET_DEBUG)
                qDebug("QSctpSocketPrivate::writeToSocket() socket is closing - returning");
#endif
                return currentChannelWritten;
            }
        }
    } while (transmitting);
 
    // At this point socket is either in Connected or Closing state,
    // write buffers are empty.
    if (state == QAbstractSocket::ClosingState)
        q->disconnectFromHost();
    else
        socketEngine->setWriteNotificationEnabled(false);
 
    return currentChannelWritten;
}
 
void QSctpSocketPrivate::configureCreatedSocket()
{
    if (socketEngine)
        socketEngine->setOption(QAbstractSocketEngine::MaxStreamsSocketOption,
                                maximumChannelCount < 0 ? 1 : maximumChannelCount);
}
 
QSctpSocket::QSctpSocket(QObject *parent)
    : QTcpSocket(SctpSocket, *new QSctpSocketPrivate, parent)
{
#if defined(QSCTPSOCKET_DEBUG)
    qDebug("QSctpSocket::QSctpSocket()");
#endif
    d_func()->isBuffered = true;
}
 
QSctpSocket::~QSctpSocket()
{
#if defined(QSCTPSOCKET_DEBUG)
    qDebug("QSctpSocket::~QSctpSocket()");
#endif
}
 
qint64 QSctpSocket::readData(char *data, qint64 maxSize)
{
    Q_D(QSctpSocket);
 
    // Cleanup headers, if the user calls the standard QIODevice functions
    if (d->currentReadChannel < d->readHeaders.size())
        d->readHeaders[d->currentReadChannel].clear();
 
    return QTcpSocket::readData(data, maxSize);
}
 
qint64 QSctpSocket::readLineData(char *data, qint64 maxlen)
{
    Q_D(QSctpSocket);
 
    // Cleanup headers, if the user calls the standard QIODevice functions
    if (d->currentReadChannel < d->readHeaders.size())
        d->readHeaders[d->currentReadChannel].clear();
 
    return QTcpSocket::readLineData(data, maxlen);
}
 
void QSctpSocket::close()
{
    QTcpSocket::close();
    d_func()->readHeaders.clear();
}
 
void QSctpSocket::disconnectFromHost()
{
    Q_D(QSctpSocket);
 
    QTcpSocket::disconnectFromHost();
    if (d->state == QAbstractSocket::UnconnectedState) {
        d->incomingDatagram.clear();
        d->writeHeaders.clear();
    }
}
 
void QSctpSocket::setMaximumChannelCount(int count)
{
    Q_D(QSctpSocket);
    if (d->state != QAbstractSocket::UnconnectedState) {
        qWarning("QSctpSocket::setMaximumChannelCount() is only allowed in UnconnectedState");
        return;
    }
#if defined(QSCTPSOCKET_DEBUG)
    qDebug("QSctpSocket::setMaximumChannelCount(%i)", count);
#endif
    d->maximumChannelCount = qMax(count, -1);
}
 
int QSctpSocket::maximumChannelCount() const
{
    return d_func()->maximumChannelCount;
}
 
bool QSctpSocket::isInDatagramMode() const
{
    Q_D(const QSctpSocket);
    return d->maximumChannelCount != -1 && d->isBuffered;
}
 
QNetworkDatagram QSctpSocket::readDatagram()
{
    Q_D(QSctpSocket);
 
    if (!isReadable() || !isInDatagramMode()) {
        qWarning("QSctpSocket::readDatagram(): operation is not permitted");
        return QNetworkDatagram();
    }
 
    if (d->currentReadChannel >= d->readHeaders.size()
        || d->readHeaders[d->currentReadChannel].size() == 0) {
        Q_ASSERT(d->buffer.isEmpty());
        return QNetworkDatagram();
    }
 
    QNetworkDatagram result(*new QNetworkDatagramPrivate(d->buffer.read(),
                                     d->readHeaders[d->currentReadChannel].front()));
    d->readHeaders[d->currentReadChannel].pop_front();
 
#if defined (QSCTPSOCKET_DEBUG)
    qDebug("QSctpSocket::readDatagram() returning datagram (%p, %i, \"%s\", %i)",
           result.d->data.constData(),
           result.d->data.size(),
           result.senderAddress().toString().toLatin1().constData(),
           result.senderPort());
#endif
 
    return result;
}
 
bool QSctpSocket::writeDatagram(const QNetworkDatagram &datagram)
{
    Q_D(QSctpSocket);
 
    if (!isWritable() || d->state != QAbstractSocket::ConnectedState || !d->socketEngine
        || !d->socketEngine->isValid() || !isInDatagramMode()) {
        qWarning("QSctpSocket::writeDatagram(): operation is not permitted");
        return false;
    }
 
    if (datagram.d->data.isEmpty()) {
        qWarning("QSctpSocket::writeDatagram() is called with empty datagram");
        return false;
    }
 
 
#if defined QSCTPSOCKET_DEBUG
    qDebug("QSctpSocket::writeDatagram(%p, %i, \"%s\", %i)",
           datagram.d->data.constData(),
           datagram.d->data.size(),
           datagram.destinationAddress().toString().toLatin1().constData(),
           datagram.destinationPort());
#endif
 
    if (d->writeHeaders.size() != d->writeBuffers.size())
        d->writeHeaders.resize(d->writeBuffers.size());
    Q_ASSERT(d->currentWriteChannel < d->writeHeaders.size());
    d->writeHeaders[d->currentWriteChannel].push_back(datagram.d->header);
    d->writeBuffer.setChunkSize(0); // set packet mode on channel buffer
    d->writeBuffer.append(datagram.d->data);
 
    d->socketEngine->setWriteNotificationEnabled(true);
    return true;
}
 
QT_END_NAMESPACE
 
#include "moc_qsctpsocket.cpp"