PacketSession.cpp

Go to the documentation of this file.

/*

Copyright (C) 2005-2007 by Peter Dimov.

This file is part of Calitko (http://www.calitko.org).

Calitko is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.

Calitko 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 General Public License for more details.

You should have received a copy of the GNU General Public License
along with Calitko; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

*/

#include "Qt.h"
#include "PacketSession.h"
#include "PacketProcessor.h"
#include "PacketReader.h"
#include "PacketWriter.h"
#include "Gnutella/PacketProcessing/QueryRouting/QrtReader.h"
#include "Gnutella/PacketProcessing/QueryRouting/QrtWriter.h"
#include "Gnutella/Packets/Packet.h"
#include "Gnutella/Packets/Bye.h"

using Gnutella::Packets::Packet;
using Gnutella::Packets::Bye;
using Gnutella::PacketProcessing::PacketSession;
using Gnutella::PacketProcessing::PacketSessionPrivate;
using Gnutella::PacketProcessing::PacketReader;
using Gnutella::PacketProcessing::PacketWriter;
using Gnutella::PacketProcessing::QueryRouting::QrtReader;
using Gnutella::PacketProcessing::QueryRouting::QrtWriter;
using Gnutella::PacketProcessing::QueryRouting::QueryRoutingTable;
using Gnutella::NodeInfo;
using Protocols::Transports::Connection;

enum Constants
{
    ReadTimeout = 60000
};

// \todo take it from some global configuration object:
static const QString SoftwareIdentification = "Calitko/0.5.3";

class PacketSessionPrivate
{
    REFERENCE_OBJECT (PacketSessionPrivate)

public:
    typedef QList <Packet *> PendingPackets;

    QrtReader           *qrtReader;
    QrtWriter           *qrtWriter;
    PacketWriter        *writer;
    PacketReader        *reader;
    Connection          *connection;
    NodeInfo            nodeInfo;
    PendingPackets      pendingPackets;
    QTimer              readTimer;
    bool                isPacketRead;
    bool                isClosing;

    PacketSessionPrivate() :    qrtReader (0), qrtWriter (0), writer (0),
                                reader (0), connection (0), nodeInfo(),
                                pendingPackets(), readTimer(),
                                isPacketRead (false), isClosing (false)
    {}
};

NodeInfo PacketSession::nodeInfo() const
{ return p->nodeInfo; }

const QueryRoutingTable & PacketSession::queryRoutingTable() const
{ return p->qrtReader->currentTable(); }

int PacketSession::readTimeout() const
{ return p->readTimer.interval(); }

void PacketSession::setReadTimeout (int readTimeout)
{ p->readTimer.setInterval (readTimeout); }

PacketSession::PacketSession (Connection *c, const NodeInfo &ni)
 :  p (new PacketSessionPrivate)
{
    Q_ASSERT (c->state() == Connection::ConnectedState);

    quint8 qrtEntryBits = 4;
    if (ni.type == Gnutella::TypeUltrapeer)
        qrtEntryBits = 1;

    p->qrtReader    = new QrtReader;
    p->qrtWriter    = new QrtWriter (this, qrtEntryBits);
    p->reader       = new PacketReader (c);
    p->writer       = new PacketWriter (c);
    p->connection   = c;
    p->nodeInfo     = ni;
    p->isPacketRead = false;
    p->isClosing    = false;

    connect (p->connection, SIGNAL (connectionClosed()),
             this, SLOT (connectionClosed()));
    connect (p->reader, SIGNAL (packetRead()), this, SLOT (packetRead()));
    connect (p->writer, SIGNAL (packetWritten()), this, SLOT (packetWritten()));
    connect (p->qrtReader, SIGNAL (qrtRead()), this, SLOT (qrtRead()));
    QObject::connect (p->reader,                SIGNAL  (readError()),
                      this,                     SLOT    (readError()));
    //QObject::connect (p->writer,  SIGNAL  (writeError()),
    //                this,                     SLOT    (writeError()));
    //QObject::connect (p->qrtReader,
    //                SIGNAL (qrtReadError (QrtReadError)),
    //                this,
    //                SLOT (qrtReadError (QrtReadError)));
    connect (&p->readTimer, SIGNAL (timeout()), this, SLOT (readTimeout()));

    p->readTimer.setInterval (ReadTimeout);
    p->readTimer.start();

    p->reader->startReading(); // Enable packet reading.
}

void PacketSession::deleteLater()
{
    p->connection->disconnect (this);
    p->reader->disconnect (p->connection);
    p->reader->disconnect (this);
    p->writer->disconnect (p->connection);
    p->writer->disconnect (this);
    p->readTimer.disconnect (this);
    p->readTimer.stop();

    QObject::deleteLater();
}

PacketSession::~PacketSession()
{
    foreach (Packet *packet, p->pendingPackets)
        delete packet;
    p->pendingPackets.clear();

    delete p->qrtReader;
    delete p->qrtWriter;
    delete p->reader;
    delete p->writer;
    delete p->connection;
    delete p;
}

void PacketSession::readTimeout()
{
    // \todo Set some error, send Bye packet?
    if (!p->isPacketRead)
        p->connection->disconnectFromNode();
    p->isPacketRead = false;
}

void PacketSession::connectionClosed()
{
    emit sessionClosed (this);
}

void PacketSession::packetRead()
{
    p->isPacketRead = true;
    bool isDropped  = false;
    Packet *packet  = p->reader->packet()->copy();

    // \todo Validate packet (ttl, hops, size)
    //isDropped = validatePacket (packet);
    isDropped = !packet->isValid();

    // \todo Keep packet traffic stats.
    //p->inStats->countPacket (packet, isDropped);

    // Special handling for QRP/QRT packets:
    if (!isDropped)
        isDropped = p->qrtReader->handlePacket (*packet);

    // \todo What about ttl=1 pings? Should we answer them hiere directly and
    // not forward them?
    if (!isDropped)
        emit packetRead (*packet, this);

    delete packet;
    p->reader->startReading();
}

void PacketSession::packetWritten()
{
    bool isDropped = true;
    Packet *packet = 0;

    do {
        if (p->pendingPackets.isEmpty()) {
            packet = 0;
            // We delayed actual disconnecting until all packets were written:
            if (p->isClosing &&
                p->connection->state() == Connection::ConnectedState) {
                p->connection->disconnectFromNode();
            }
            break;
        } else {
            packet = p->pendingPackets.takeFirst();
            /*  \todo When flow control is implemented, some  packets may be
                dropped. Until then: */
            isDropped = false;
        }
        // \todo Keep packet traffic stats.
        //p->outStats->countPacket (packet, isDropped);

        if (isDropped)
            delete packet;
    } while (isDropped);

    if (packet) {
        p->writer->write (*packet);
        delete packet;
    }
}

void PacketSession::closeSession (CloseCode closeCode)
{
    Q_ASSERT (!p->isClosing);

    quint16 code = 0;
    QString message;

    switch (closeCode)
    {
    case CloseNoBye:
        break;
    case ExittingNormally:
        message = "Application is terminating, closing all connections.";
        break;
    case ExplicitClose:
        message = "The user requested explicit close of this connection.";
        break;
    case TooBigPackets:
        message = "You send too big packets!";
        break;
    case TooManyDuplicates:
        message = "You are sending too many duplicate packets!";
        break;
    case BadQueries:
        message = "You are relaying bad queries!";
        break;
    case TooHighTtlHops:
        message = "You are relaying packets with excessive ttl + hops value!";
        break;
    case ManyUnknownPackets:
        message = "You are sending too many unknown packets!";
        break;
    case InactivityTimeout:
        message = "Your connection reached my inactivity timeout!";
        break;
    case NoTtl1PingReply:
        message = "You failed to reply to a ping with ttl=1!";
        break;
    case NotSharingEnough:
        message = "You are not sharing enough!";
        break;
    case InternalError:
        message = "Closing due to internal error! Sorry!";
        break;
    case ProtocolDesync:
        message = "Closing due to protocol desynchronization!";
        break;
    case SendQueueFull:
        message = "Send queue for your connection became full!";
        break;
    default:
        Q_ASSERT (false); // Bad CloseCode
    }
    // Remove all pending packets from send queue:
    foreach (Packet *packet, p->pendingPackets)
        delete packet;
    p->pendingPackets.clear();

    if (code != 0) {
        // Send a bye packet with HTTP style message:
        message += "\r\n";
        message += QString ("Server: %1\r\n\r\n").arg (SoftwareIdentification);
        Bye bye (code, message.toLatin1());
        sendPacket (bye);
    } else {
        // Directly close the connection:
        p->connection->disconnectFromNode();
    }
    // Disable sending any more packets, yet allow bye to be sent.
    p->isClosing = true;
}

void PacketSession::sendPacket (const Packet &packet)
{
    if (p->isClosing)
        return;
    // \todo Add flow control => use a kind of priority queue for output packets
    p->pendingPackets.append (packet.copy());
    if (p->writer->canWrite())
        packetWritten();
}

void PacketSession::qrtRead()
{
    emit qrtRead (this);
}

void PacketSession::sendQrt (const QueryRoutingTable &table)
{
    p->qrtWriter->write (table);
}

void PacketSession::readError()
{
    if (!p->isClosing)
        closeSession (ProtocolDesync);
}

Copyright © 2006 Peter Dimov.
Powered by Doxygen.