TransferSessionImpl.cpp

Go to the documentation of this file.

/*

Copyright (C) 2006-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 "TransferSessionImpl.h"
#include "Imports.cpp"


TransferSessionImpl::TransferSessionImpl (
        Session *session,
        const Handshake &handshake,
        IntegerResetStopwatch *downloadStopwatch,
        IntegerResetStopwatch *uploadStopwatch,
        TransferSessionStatus *status)
 :  session_ (session), handshake_ (handshake), status_ (status),
    inPacketDispatcher_(), outPacketDispatcher_(),
    areWeChoked_ (true), areWeInterested_ (false), areWeSeeder_ (false),
    isPeerChoked_ (true), isPeerInterested_ (false), isPeerSeeder_ (false),
    totalDownloaded_ (0), totalUploaded_ (0),
    downloadStopwatch_ (downloadStopwatch), uploadStopwatch_ (uploadStopwatch),
    downloadHistory_(), uploadHistory_(),
    peerPiecesAvailable_ (handshake.numberPieces),
    wePiecesAvailable_ (handshake.numberPieces)
{
    inPacketDispatcher_.registerTypeHandler (
        Callable <void (const Choke &)> (
            this, &TransferSessionImpl::receivedChoke));
    inPacketDispatcher_.registerTypeHandler (
        Callable <void (const Unchoke &)> (
            this, &TransferSessionImpl::receivedUnchoke));
    inPacketDispatcher_.registerTypeHandler (
        Callable <void (const Interested &)> (
            this, &TransferSessionImpl::receivedInterested));
    inPacketDispatcher_.registerTypeHandler (
        Callable <void (const NotInterested &)> (
            this, &TransferSessionImpl::receivedNotInterested));
    inPacketDispatcher_.registerTypeHandler (
        Callable <void (const Piece &)> (
            this, &TransferSessionImpl::receivedPiece));
    inPacketDispatcher_.registerTypeHandler (
        Callable <void (const BitField &)> (
            this, &TransferSessionImpl::receivedBitField));
    inPacketDispatcher_.registerTypeHandler (
        Callable <void (const Have &)> (
            this, &TransferSessionImpl::receivedHave));
    if (handshake_.hasFastExtension) {
        inPacketDispatcher_.registerTypeHandler (
            Callable <void (const HaveAll &)> (
                this, &TransferSessionImpl::receivedHaveAll));
        inPacketDispatcher_.registerTypeHandler (
            Callable <void (const HaveNone &)> (
                this, &TransferSessionImpl::receivedHaveNone));
    }
    inPacketDispatcher_.registerNotHandledHandler (
        Callable <void (const PacketBase &)> (
            this, &TransferSessionImpl::receivedUnhandled));

    outPacketDispatcher_.registerTypeHandler (
        Callable <void (const Choke &)> (
            this, &TransferSessionImpl::sentChoke));
    outPacketDispatcher_.registerTypeHandler (
        Callable <void (const Unchoke &)> (
            this, &TransferSessionImpl::sentUnchoke));
    outPacketDispatcher_.registerTypeHandler (
        Callable <void (const Interested &)> (
            this, &TransferSessionImpl::sentInterested));
    outPacketDispatcher_.registerTypeHandler (
        Callable <void (const NotInterested &)> (
            this, &TransferSessionImpl::sentNotInterested));
    outPacketDispatcher_.registerTypeHandler (
        Callable <void (const Piece &)> (
            this, &TransferSessionImpl::sentPiece));
    outPacketDispatcher_.registerTypeHandler (
        Callable <void (const BitField &)> (
            this, &TransferSessionImpl::sentBitField));
    outPacketDispatcher_.registerTypeHandler (
        Callable <void (const Have &)> (
            this, &TransferSessionImpl::sentHave));
    if (handshake_.hasFastExtension) {
        outPacketDispatcher_.registerTypeHandler (
            Callable <void (const HaveAll &)> (
                this, &TransferSessionImpl::sentHaveAll));
        outPacketDispatcher_.registerTypeHandler (
            Callable <void (const HaveNone &)> (
                this, &TransferSessionImpl::sentHaveNone));
    }
    outPacketDispatcher_.registerNotHandledHandler (
        Callable <void (const PacketBase &)> (
            this, &TransferSessionImpl::sentUnhandled));

    for (int i = SpeedMovingAverageDuration; i > 0; --i) {
        downloadHistory_.append (0);
        uploadHistory_.append (0);
    }
}


void TransferSessionImpl::sendPacket (const Packet &packet)
{
    status_->transferSessionSendingPacket (this, packet);
    outPacketDispatcher_.dispatch (*packet);
    session_->send (packet);
}

void TransferSessionImpl::close()
{
    session_->close();
}

bool TransferSessionImpl::areWeChoked() const
{
    return areWeChoked_;
}

bool TransferSessionImpl::areWeInterested() const
{
    return areWeInterested_;
}

bool TransferSessionImpl::areWeSnubbed() const
{
    return downloadStopwatch_->elapsedSeconds() > SnubbedTimeout
            && !isPeerChoked();
}

bool TransferSessionImpl::areWeSeeder() const
{
    return areWeSeeder_;
}

bool TransferSessionImpl::isPeerChoked() const
{
    return isPeerChoked_;
}

bool TransferSessionImpl::isPeerInterested() const
{
    return isPeerInterested_;
}

bool TransferSessionImpl::isPeerSnubbed() const
{
    return uploadStopwatch_->elapsedSeconds() > SnubbedTimeout
            && !areWeChoked();
}

bool TransferSessionImpl::isPeerSeeder() const
{
    return isPeerSeeder_;
}

qint64 TransferSessionImpl::totalDownloaded() const
{
    return totalDownloaded_;
}

qint64 TransferSessionImpl::totalUploaded() const
{
    return totalUploaded_;
}

qint64 TransferSessionImpl::currentDownloadSpeed() const
{
    updateDownloadSpeed (0);
    return calculateAverage (downloadHistory_);
}

qint64 TransferSessionImpl::currentUploadSpeed() const
{
    updateUploadSpeed (0);
    return calculateAverage (uploadHistory_);
}

QBitArray TransferSessionImpl::peerPiecesAvailable() const
{
    return peerPiecesAvailable_;
}

QBitArray TransferSessionImpl::wePiecesAvailable() const
{
    return wePiecesAvailable_;
}


void TransferSessionImpl::sessionReceivedData (Session *, const Data &data)
{
    Packet packet = data;
    if (inPacketDispatcher_.dispatch (*packet))
        status_->transferSessionReceivedPacket (this, packet);
}


void TransferSessionImpl::sessionSendingData (Session *, const Data &)
{
    // Nothing to do here.
}


void TransferSessionImpl::sessionClosing (Session *)
{
    status_->transferSessionClosing (this);
}


void TransferSessionImpl::sessionClosed (Session *)
{
    status_->transferSessionClosed (this);
}

void TransferSessionImpl::receivedChoke (const Choke &)
{
    areWeChoked_ = true;
}

void TransferSessionImpl::receivedUnchoke (const Unchoke &)
{
    areWeChoked_ = false;
}

void TransferSessionImpl::receivedInterested (const Interested &)
{
    isPeerInterested_ = true;
}

void TransferSessionImpl::receivedNotInterested (const NotInterested &)
{
    isPeerInterested_ = false;
}

void TransferSessionImpl::receivedPiece (const Piece &piece)
{
    int length = piece.pieceBytes().length();
    totalDownloaded_ += length;
    updateDownloadSpeed (length);
}

void TransferSessionImpl::receivedBitField (const BitField &bitField)
{
    setPiecesAvailable (peerPiecesAvailable_, bitField.bitField());
    unregisterBitFieldHaveAllHaveNoneReceiveHandlers();
}

void TransferSessionImpl::receivedHaveAll (const HaveAll &)
{
    peerPiecesAvailable_.fill (true);
    isPeerSeeder_ = true;
    unregisterBitFieldHaveAllHaveNoneReceiveHandlers();
}

void TransferSessionImpl::receivedHaveNone (const HaveNone &)
{
    peerPiecesAvailable_.fill (false);
    isPeerSeeder_ = false;
    unregisterBitFieldHaveAllHaveNoneReceiveHandlers();
}

void TransferSessionImpl::receivedHave (const Have &have)
{
    peerPiecesAvailable_.setBit (have.pieceIndex(), true);
    if (peerPiecesAvailable_.size() == peerPiecesAvailable_.count (true))
        isPeerSeeder_ = true;
}


void TransferSessionImpl::receivedUnhandled (const PacketBase &)
{
    session_->close(); // It is not allowed to send this type of packet.
}

void TransferSessionImpl::sentChoke (const Choke &)
{
    isPeerChoked_ = true;
}

void TransferSessionImpl::sentUnchoke (const Unchoke &)
{
    isPeerChoked_ = false;
}

void TransferSessionImpl::sentInterested (const Interested &)
{
    areWeInterested_ = true;
}

void TransferSessionImpl::sentNotInterested (const NotInterested &)
{
    areWeInterested_ = false;
}

void TransferSessionImpl::sentPiece (const Piece &piece)
{
    int length = piece.pieceBytes().length();
    totalUploaded_ += length;
    updateUploadSpeed (length);
}

void TransferSessionImpl::sentBitField (const BitField &bitField)
{
    setPiecesAvailable (wePiecesAvailable_, bitField.bitField());
    unregisterBitFieldHaveAllHaveNoneSendHandlers();
}

void TransferSessionImpl::sentHaveAll (const HaveAll &)
{
    wePiecesAvailable_.fill (true);
    areWeSeeder_ = true;
    unregisterBitFieldHaveAllHaveNoneSendHandlers();
}

void TransferSessionImpl::sentHaveNone (const HaveNone &)
{
    wePiecesAvailable_.fill (false);
    areWeSeeder_ = false;
    unregisterBitFieldHaveAllHaveNoneSendHandlers();
}

void TransferSessionImpl::sentHave (const Have &have)
{
    wePiecesAvailable_.setBit (have.pieceIndex(), true);
    if (wePiecesAvailable_.size() == wePiecesAvailable_.count (true))
        areWeSeeder_ = true;
}


void TransferSessionImpl::sentUnhandled (const PacketBase &)
{
    Q_ASSERT (false); // It is not allowed to send this packet type!
}


void TransferSessionImpl::updateDownloadSpeed (int pieceLength) const
{
    qint64 delay = downloadStopwatch_->elapsedSeconds();
    shiftList (downloadHistory_, delay);
    downloadHistory_.front() += pieceLength;
    downloadStopwatch_->reset();
}


void TransferSessionImpl::updateUploadSpeed (int pieceLength) const
{
    qint64 delay = uploadStopwatch_->elapsedSeconds();
    shiftList (uploadHistory_, delay);
    uploadHistory_.front() += pieceLength;
    uploadStopwatch_->reset();
}

void TransferSessionImpl::shiftList (QList <qint64> &list, qint64 count)
{
    count = qMin (count, static_cast <qint64> (SpeedMovingAverageDuration));
    for (qint64 i = count; i > 0; --i) {
        list.prepend (0);
        list.takeLast();
    }
}

qint64 TransferSessionImpl::calculateAverage (const QList <qint64> &list)
{
    Q_ASSERT (list.size() != 0);
    qint64 sum = std::accumulate (list.begin(), list.end(), 0);
    return sum / list.size();
}

void TransferSessionImpl::setPiecesAvailable (QBitArray &piecesAvailable,
                                              const QByteArray &bitField)
{
    int index = 0;
    int last = piecesAvailable.size();
    foreach (const char &ch, bitField)
        for (uchar mask = 0x80; mask != 0 && index < last; mask >>= 1, ++index)
            piecesAvailable.setBit (index, ch & mask);
}


void TransferSessionImpl::unregisterBitFieldHaveAllHaveNoneReceiveHandlers()
{
    inPacketDispatcher_.unregisterTypeHandler (
        Callable <void (const BitField &)> (
            this, &TransferSessionImpl::receivedBitField));
    inPacketDispatcher_.unregisterTypeHandler (
        Callable <void (const HaveAll &)> (
            this, &TransferSessionImpl::receivedHaveAll));
    inPacketDispatcher_.unregisterTypeHandler (
        Callable <void (const HaveNone &)> (
            this, &TransferSessionImpl::receivedHaveNone));
}


void TransferSessionImpl::unregisterBitFieldHaveAllHaveNoneSendHandlers()
{
    outPacketDispatcher_.unregisterTypeHandler (
        Callable <void (const BitField &)> (
            this, &TransferSessionImpl::sentBitField));
    outPacketDispatcher_.unregisterTypeHandler (
        Callable <void (const HaveAll &)> (
            this, &TransferSessionImpl::sentHaveAll));
    outPacketDispatcher_.unregisterTypeHandler (
        Callable <void (const HaveNone &)> (
            this, &TransferSessionImpl::sentHaveNone));
}

Copyright © 2006 Peter Dimov.
Powered by Doxygen.