src/main/connection.cpp

   1 /***************************************************************************
   2  *                                                                         *
   3  *   Copyright (C) 2006 David Brodsky                                      *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or         *
   6  *   modify it under the terms of the GNU General Public License as        *
   7  *   published by the Free Software Foundation and appearing               *
   8  *   in the file LICENSE.GPL included in the packaging of this file.       *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
  13  *   General Public License for more details.                              *
  14  *                                                                         *
  15  ***************************************************************************/
  16
  17 #include <arpa/inet.h>
  18
  19 #include <tairon/core/log.h>
  20 #include <tairon/core/thread.h>
  21 #include <tairon/net/ireader.h>
  22 #include <tairon/net/limiter.h>
  23 #include <tairon/net/socket.h>
  24 #include <tairon/net/swriter.h>
  25 #include <tairon/net/timer.h>
  26
  27 #include "connection.h"
  28
  29 #include "core/bencode.h"
  30 #include "core/bitfield.h"
  31 #include "ratemeasurer.h"
  32 #include "storage.h"
  33 #include "torrentclient.h"
  34 #include "torrentmanager.h"
  35 #include "torrentserver.h"
  36
  37 namespace Tairent
  38 {
  39
  40 namespace Main
  41 {
  42
  43 const unsigned int handshakeLength = 48; // without peer id
  44 const char *protocolName = "BitTorrent protocol";
  45
  46 Tairon::Net::Limiter *rlimiter = 0;
  47 Tairon::Net::Limiter *wlimiter = 0;
  48
  49 /* {{{ Connection::Connection(int) */
  50 Connection::Connection(int fd) : bitfield(0), choked(true), client(0), commandReader(0), deleteReader(true), interested(false), lastPieceTime(0), lengthReader(0), peerChoked(true), peerInterested(false), pieceReaders(0), pieceWriters(0), reader(0)
  51 {
  52         socket = new Tairon::Net::Socket(fd);
  53         init();
  54
  55         incomingRate = new RateMeasurer(10);
  56         outgoingRate = new RateMeasurer(10);
  57 }
  58 /* }}} */
  59
  60 /* {{{ Connection::Connection(const String &, uint16_t, const String &, TorrentClient *) */
  61 Connection::Connection(const String &ip, uint16_t port, const String &pID, TorrentClient *c) : bitfield(0), choked(true), client(c), commandReader(0), deleteReader(true), interested(false), lastPieceTime(0), lengthReader(0), peerChoked(true), peerID(pID), peerInterested(false), pieceReaders(0), pieceWriters(0), reader(0)
  62 {
  63         Tairon::Core::Thread *current = Tairon::Core::Thread::current();
  64
  65         socket = new Tairon::Net::Socket(Tairon::Net::Socket::IPv4, Tairon::Net::Socket::Stream);
  66         socket->connectedSignal.connect(Tairon::Core::threadMethodDFunctor(current, this, &Connection::socketConnected));
  67         socket->errorSignal.connect(Tairon::Core::threadMethodDFunctor(current, this, &Connection::connectingError));
  68
  69         incomingRate = new RateMeasurer(10);
  70         outgoingRate = new RateMeasurer(10);
  71
  72         socket->connect(ip, port);
  73 }
  74 /* }}} */
  75
  76 /* {{{ Connection::~Connection() */
  77 Connection::~Connection()
  78 {
  79         delete incomingRate;
  80         delete outgoingRate;
  81         delete bitfield;
  82         if (deleteReader)
  83                 delete reader;
  84         delete commandReader;
  85         delete lengthReader;
  86
  87         for (std::list<Tairon::Net::Writer *>::const_iterator it = commandQueue.begin(); it != commandQueue.end(); ++it)
  88                 delete *it;
  89 }
  90 /* }}} */
  91
  92 /* {{{ Connection::readBitField(Tairon::Net::Reader *) */
  93 void Connection::readBitField(Tairon::Net::Reader *)
  94 {
  95         DEBUG("bitfield read");
  96         bitfield = new Tairent::Core::BitField(reader->getBuffer(), client->getMetaInfo()["info"]["pieces"].asString().length() / 20);
  97         delete reader;
  98         reader = lengthReader;
  99         deleteReader = false;
 100
 101         client->gotBitField(this);
 102 }
 103 /* }}} */
 104
 105 /* {{{ Connection::readCancel(Tairon::Net::Reader *) */
 106 void Connection::readCancel(Tairon::Net::Reader *)
 107 {
 108         uint32_t index = htonl(*(uint32_t *) reader->getBuffer());
 109         uint32_t start = htonl(*(uint32_t *) (reader->getBuffer() + 4));
 110         uint32_t length = htonl(*(uint32_t *) (reader->getBuffer() + 8));
 111
 112         for (std::list<PieceRequestStruct>::iterator it = peersRequests.begin(); it != peersRequests.end(); ++it)
 113                 if ((it->index == index) && (it->start == start) && (it->length == length)) {
 114                         peersRequests.erase(it);
 115                         break;
 116                 }
 117
 118         delete reader;
 119         reader = lengthReader;
 120         deleteReader = false;
 121 }
 122 /* }}} */
 123
 124 /* {{{ Connection::readCommand(Tairon::Net::Reader *) */
 125 void Connection::readCommand(Tairon::Net::Reader *)
 126 {
 127         commandReader->reset();
 128         switch (*reader->getBuffer()) {
 129                 case 0:
 130                         if (commandLength != 1) {
 131                                 DEBUG("Invalid length of choke command");
 132                                 close();
 133                                 closedSignal.emit(this);
 134                                 return;
 135                         }
 136                         DEBUG("choke");
 137                         reader = lengthReader;
 138                         if (choked) // we're already choked
 139                                 break;
 140                         choked = true;
 141                         client->gotChoke(this);
 142                         break;
 143                 case 1:
 144                         if (commandLength != 1) {
 145                                 DEBUG("Invalid length of unchoke command");
 146                                 close();
 147                                 closedSignal.emit(this);
 148                                 return;
 149                         }
 150                         DEBUG("unchoke");
 151                         reader = lengthReader;
 152                         if (!choked) // we're already unchoked
 153                                 break;
 154                         choked = false;
 155                         client->gotUnchoke(this);
 156                         break;
 157                 case 2:
 158                         if (commandLength != 1) {
 159                                 DEBUG("Invalid length of interested command");
 160                                 close();
 161                                 closedSignal.emit(this);
 162                                 return;
 163                         }
 164                         DEBUG("interested");
 165                         reader = lengthReader;
 166                         peerInterested = true;
 167                         client->gotInterestedChange(this);
 168                         break;
 169                 case 3:
 170                         if (commandLength != 1) {
 171                                 DEBUG("Invalid length of not interested command");
 172                                 close();
 173                                 closedSignal.emit(this);
 174                                 return;
 175                         }
 176                         DEBUG("not interested");
 177                         reader = lengthReader;
 178                         peerInterested = false;
 179                         client->gotInterestedChange(this);
 180                         break;
 181                 case 4:
 182                         if (commandLength != 5) {
 183                                 DEBUG("Invalid length of have command");
 184                                 close();
 185                                 closedSignal.emit(this);
 186                                 return;
 187                         }
 188                         DEBUG("have");
 189                         reader = new Tairon::Net::IReader(4, socket, rlimiter);
 190                         reader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readHave));
 191                         reader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 192                         deleteReader = true;
 193                         break;
 194                 case 5:
 195                         DEBUG("bitfield");
 196                         close();
 197                         closedSignal.emit(this);
 198                         return;
 199                 case 6:
 200                         if (commandLength != 13) {
 201                                 DEBUG("Invalid length of request command");
 202                                 close();
 203                                 closedSignal.emit(this);
 204                                 return;
 205                         }
 206                         DEBUG("request");
 207                         reader = new Tairon::Net::IReader(12, socket, rlimiter);
 208                         reader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readRequest));
 209                         reader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 210                         deleteReader = true;
 211                         break;
 212                 case 7:
 213                         DEBUG("piece");
 214                         reader = new Tairon::Net::IReader(8, socket, rlimiter);
 215                         reader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readPiece));
 216                         reader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 217                         lastPieceTime = Tairon::Net::Timer::currentTime();
 218                         deleteReader = true;
 219                         break;
 220                 case 8:
 221                         if (commandLength != 13) {
 222                                 DEBUG("Invalid length of cancel command");
 223                                 close();
 224                                 closedSignal.emit(this);
 225                                 return;
 226                         }
 227                         DEBUG("cancel");
 228                         reader = new Tairon::Net::IReader(12, socket, rlimiter);
 229                         reader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readCancel));
 230                         reader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 231                         deleteReader = true;
 232                         break;
 233                 default:
 234                         DEBUG((const char *) String("unknown message" + String::number(*reader->getBuffer())));
 235                         close();
 236                         closedSignal.emit(this);
 237         }
 238 }
 239 /* }}} */
 240
 241 /* {{{ Connection::readFirstCommand(Tairon::Net::Reader *) */
 242 void Connection::readFirstCommand(Tairon::Net::Reader *)
 243 {
 244         commandReader->bufferFullSignal.clear();
 245         commandReader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readCommand));
 246         commandReader->reset();
 247
 248         if (*reader->getBuffer() == 5) { // bitfield
 249                 DEBUG("bitfield");
 250
 251                 // check if the peer isn't sending fake data
 252                 uint32_t len = client->getMetaInfo()["info"]["pieces"].asString().length() / 20;
 253                 if ((commandLength - 1) != (len / 8 + (len % 8 ? 1 : 0))) {
 254                         ERROR("Peer sent an invalid bitfield length");
 255                         socket->close();
 256                         closedSignal.emit(this);
 257                         return;
 258                 }
 259
 260                 reader = new Tairon::Net::IReader(commandLength - 1, socket, rlimiter);
 261                 reader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readBitField));
 262                 reader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 263
 264                 deleteReader = true;
 265         } else {
 266                 bitfield = new Tairent::Core::BitField(client->getMetaInfo()["info"]["pieces"].asString().length() / 20);
 267                 readCommand(reader);
 268         }
 269 }
 270 /* }}} */
 271
 272 /* {{{ Connection::readHandshake(Tairon::Net::Reader *) */
 273 void Connection::readHandshake(Tairon::Net::Reader *)
 274 {
 275         DEBUG("got handshake 1");
 276
 277         const char *buffer = reader->getBuffer();
 278
 279         // check the protocol length (in the case that reader emitted
 280         // bufferFullSignal instead of dataReadSignal.
 281         if (*buffer != 19) {
 282                 WARNING("Invalid protocol length");
 283                 close();
 284                 closedSignal.emit(this);
 285                 return;
 286         }
 287
 288         if (strstr(buffer + 1, protocolName) != (buffer + 1)) { // invalid protocol name
 289                 WARNING("Invalid protocol name");
 290                 close();
 291                 closedSignal.emit(this);
 292                 return;
 293         }
 294
 295         DEBUG("Protocol ok");
 296
 297         if (!client) {
 298                 String infoHash(buffer + 28, 20);
 299                 DEBUG("sending handshake");
 300                 client = TorrentManager::self()->getClient(infoHash);
 301                 if (!client) {
 302                         WARNING("Got unknown infohash");
 303                         close();
 304                         closedSignal.emit(this);
 305                         return;
 306                 }
 307                 sendHandshake(infoHash);
 308         }
 309
 310         delete reader;
 311
 312         reader = new Tairon::Net::IReader(20, socket, rlimiter);
 313         reader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readPeerID));
 314         reader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 315 }
 316 /* }}} */
 317
 318 /* {{{ Connection::readHave(Tairon::Net::Reader *) */
 319 void Connection::readHave(Tairon::Net::Reader *)
 320 {
 321         uint32_t index = htonl(*(uint32_t *) reader->getBuffer());
 322
 323         if (index > bitfield->getLength()) {
 324                 WARNING("Peer sent invalid have index");
 325                 close();
 326                 closedSignal.emit(this);
 327                 return;
 328         }
 329
 330         delete reader;
 331         reader = lengthReader;
 332         deleteReader = false;
 333
 334         bitfield->setBit(index);
 335         client->gotHave(index, this);
 336 }
 337 /* }}} */
 338
 339 /* {{{ Connection::readLength(Tairon::Net::Reader *) */
 340 void Connection::readLength(Tairon::Net::Reader *)
 341 {
 342         commandLength = ntohl(*(uint32_t *) lengthReader->getBuffer());
 343         DEBUG((const char *) String("Got message of length " + String::number(commandLength)));
 344
 345         lengthReader->reset();
 346
 347         if (commandLength)
 348                 reader = commandReader;
 349         // else keepalive
 350 }
 351 /* }}} */
 352
 353 /* {{{ Connection::readPeerID(Tairon::Net::Reader *) */
 354 void Connection::readPeerID(Tairon::Net::Reader *)
 355 {
 356         DEBUG("got peer id");
 357         String pID(reader->getBuffer(), 20);
 358
 359         if ((peerID.length() == 20) && (peerID != pID)) { // we are connecting to a peer and it's ID doesn't match
 360                 WARNING("Peer's ID doesn't match");
 361                 close();
 362                 closedSignal.emit(this);
 363                 return;
 364         }
 365
 366         String oldPeerID = peerID;
 367         peerID = pID;
 368
 369         delete reader;
 370
 371         lengthReader = new Tairon::Net::IReader(4, socket, rlimiter);
 372         lengthReader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readLength));
 373         lengthReader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 374
 375         commandReader = new Tairon::Net::IReader(1, socket, rlimiter);
 376         commandReader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readFirstCommand));
 377         commandReader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 378
 379         reader = lengthReader;
 380         deleteReader = false;
 381
 382         TorrentServer::self()->connectionCompleted(this);
 383         client->newConnection(this, oldPeerID);
 384 }
 385 /* }}} */
 386
 387 /* {{{ Connection::readPiece(Tairon::Net::Reader *) */
 388 void Connection::readPiece(Tairon::Net::Reader *)
 389 {
 390         pieceIndex = ntohl(*(uint32_t *) reader->getBuffer());
 391         pieceStart = ntohl(*(uint32_t *) (reader->getBuffer() + 4));
 392         delete reader;
 393
 394         std::set<uint32_t> &r = requests[pieceIndex];
 395         r.erase(pieceStart);
 396         if (!r.size())
 397                 requests.erase(pieceIndex);
 398
 399         deleteReader = false;
 400         client->gotPiece(pieceIndex, pieceStart, this);
 401 }
 402 /* }}} */
 403
 404 /* {{{ Connection::readProtocolLength(Tairon::Net::Reader *, size_t) */
 405 void Connection::readProtocolLength(Tairon::Net::Reader *, size_t)
 406 {
 407         if (*reader->getBuffer() != 19) { // length of the protocol name
 408                 WARNING("Invalid protocol length");
 409                 close();
 410                 closedSignal.emit(this);
 411                 return;
 412         }
 413
 414         reader->dataReadSignal.clear();
 415 }
 416 /* }}} */
 417
 418 /* {{{ Connection::readRequest(Tairon::Net::Reader *) */
 419 void Connection::readRequest(Tairon::Net::Reader *)
 420 {
 421         if (peerChoked) { // don't send anything to the choked peer
 422                 delete reader;
 423                 reader = lengthReader;
 424                 deleteReader = false;
 425                 return;
 426         }
 427
 428         PieceRequestStruct rs;
 429         rs.index = htonl(*(uint32_t *) reader->getBuffer());
 430         rs.start = htonl(*(uint32_t *) (reader->getBuffer() + 4));
 431         rs.length = htonl(*(uint32_t *) (reader->getBuffer() + 8));
 432
 433         if (rs.length > 16384) {
 434                 WARNING((const char *) String("Client requested too big chunk: " + String::number(rs.length)));
 435                 close();
 436                 closedSignal.emit(this);
 437                 return;
 438         }
 439
 440         delete reader;
 441         reader = lengthReader;
 442         deleteReader = false;
 443
 444         if (pieceWriters || commandQueue.size()) // we are sending something
 445                 peersRequests.push_back(rs);
 446         else
 447                 client->getStorage()->gotRequest(rs.index, rs.start, rs.length, this);
 448 }
 449 /* }}} */
 450
 451 /* {{{ Connection::readyRead(Tairon::Net::Socket *) */
 452 void Connection::readyRead(Tairon::Net::Socket *)
 453 {
 454         try {
 455                 reader->read();
 456         } catch (const Tairon::Net::SocketException &) { // connection closed or other error
 457                 close();
 458                 closedSignal.emit(this);
 459         }
 460 }
 461 /* }}} */
 462
 463 /* {{{ Connection::readyWrite(Tairon::Net::Socket *) */
 464 void Connection::readyWrite(Tairon::Net::Socket *)
 465 {
 466         if (pieceWriters)
 467                 pieceWriters->writers.front()->write();
 468         else
 469                 commandQueue.front()->write();
 470 }
 471 /* }}} */
 472
 473 /* {{{ Connection::addCommandWriter(Tairon::Net::Writer *) */
 474 void Connection::addCommandWriter(Tairon::Net::Writer *writer)
 475 {
 476         commandQueue.push_back(writer);
 477         if ((commandQueue.size() == 1) && !pieceWriters) // don't send anything if we are sending something
 478                 writer->write();
 479 }
 480 /* }}} */
 481
 482 /* {{{ Connection::connectingError(Tairon::Net::Socket *, int) */
 483 void Connection::connectingError(Tairon::Net::Socket *, int)
 484 {
 485         socket->close();
 486         client->connectingFailed(this);
 487 }
 488 /* }}} */
 489
 490 /* {{{ Connection::close() */
 491 void Connection::close()
 492 {
 493         socket->close();
 494 }
 495 /* }}} */
 496
 497 /* {{{ Connection::clearRequested() */
 498 void Connection::clearRequested()
 499 {
 500         requests.clear();
 501 }
 502 /* }}} */
 503
 504 /* {{{ Connection::dataWritten(Tairon::Net::Writer *) */
 505 void Connection::dataWritten(Tairon::Net::Writer *writer)
 506 {
 507         DEBUG("dataWritten");
 508
 509         Tairon::Net::Writer *w = commandQueue.front();
 510         if (w == writer) // is this writer in the queue?
 511                 commandQueue.pop_front(); // remove it
 512         delete writer;
 513
 514         if (commandQueue.size()) // are there some commands to send?
 515                 commandQueue.front()->write();
 516         else if (peersRequests.size()) { // let's try to send some piece
 517                 PieceRequestStruct rs = peersRequests.front();
 518                 peersRequests.pop_front();
 519                 client->getStorage()->gotRequest(rs.index, rs.start, rs.length, this);
 520         }
 521 }
 522 /* }}} */
 523
 524 /* {{{ Connection::destroyPieceReaders() */
 525 void Connection::destroyPieceReaders()
 526 {
 527         if (!pieceReaders)
 528                 return;
 529
 530         for (std::list<Tairon::Net::Reader *>::iterator it = pieceReaders->readers.begin(); it != pieceReaders->readers.end(); ++it)
 531                 delete *it;
 532         delete pieceReaders;
 533         pieceReaders = 0;
 534 }
 535 /* }}} */
 536
 537 /* {{{ Connection::destroyPieceWriters() */
 538 void Connection::destroyPieceWriters()
 539 {
 540         if (!pieceWriters)
 541                 return;
 542
 543         for (std::list<Tairon::Net::Writer *>::iterator it = pieceWriters->writers.begin(); it != pieceWriters->writers.end(); ++it)
 544                 delete *it;
 545         delete pieceWriters;
 546         pieceWriters = 0;
 547 }
 548 /* }}} */
 549
 550 /* {{{ Connection::getIncomingRate() */
 551 double Connection::getIncomingRate()
 552 {
 553         return incomingRate->getRate();
 554 }
 555 /* }}} */
 556
 557 /* {{{ Connection::getOutgoingRate() */
 558 double Connection::getOutgoingRate()
 559 {
 560         return outgoingRate->getRate();
 561 }
 562 /* }}} */
 563
 564 /* {{{ Connection::getPieceLength() */
 565 uint32_t Connection::getPieceLength()
 566 {
 567         // 1 byte message type, 4 bytes index, 4 bytes start
 568         return commandLength - 9;
 569 }
 570 /* }}} */
 571
 572 /* {{{ Connection::getReadingLimiter() */
 573 Tairon::Net::Limiter *Connection::getReadingLimiter()
 574 {
 575         return rlimiter;
 576 }
 577 /* }}} */
 578
 579 /* {{{ Connection::getRequestsCount() */
 580 unsigned int Connection::getRequestsCount()
 581 {
 582         unsigned int ret = 0;
 583         for (std::map<uint32_t, std::set<uint32_t> >::const_iterator it = requests.begin(); it != requests.end(); ++it)
 584                 ret += it->second.size();
 585         return ret;
 586 }
 587 /* }}} */
 588
 589 /* {{{ Connection::getWritingLimiter() */
 590 Tairon::Net::Limiter *Connection::getWritingLimiter()
 591 {
 592         return wlimiter;
 593 }
 594 /* }}} */
 595
 596 /* {{{ Connection::init() */
 597 void Connection::init()
 598 {
 599         Tairon::Core::Thread *current = Tairon::Core::Thread::current();
 600
 601         socket->errorSignal.connect(Tairon::Core::threadMethodDFunctor(current, this, &Connection::socketError));
 602         socket->readyReadSignal.connect(Tairon::Core::threadMethodDFunctor(current, this, &Connection::readyRead));
 603         socket->readyWriteSignal.connect(Tairon::Core::threadMethodDFunctor(current, this, &Connection::readyWrite));
 604         socket->ready();
 605
 606         reader = new Tairon::Net::IReader(handshakeLength, socket, rlimiter);
 607         reader->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readHandshake));
 608         // it is safe to do this, we just want to check the first byte
 609         reader->dataReadSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readProtocolLength));
 610         reader->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 611 }
 612 /* }}} */
 613
 614 /* {{{ Connection::isSnubbed() */
 615 bool Connection::isSnubbed()
 616 {
 617         return Tairon::Net::Timer::currentTime() - lastPieceTime > 10000; // 10 seconds, should be configurable
 618 }
 619 /* }}} */
 620
 621 /* {{{ Connection::pieceDataWritten(Tairon::Net::Writer *) */
 622 void Connection::pieceDataWritten(Tairon::Net::Writer *writer)
 623 {
 624         pieceWriters->writers.pop_front();
 625
 626         if (pieceWriters->writers.size()) { // there is still something to send from the piece
 627                 delete writer; // delete the current writer
 628                 pieceWriters->writers.front()->write();
 629         } else { // whole piece has been sent
 630                 outgoingRate->update(pieceWriters->length);
 631                 client->pieceSent(pieceWriters->index, pieceWriters->length, this);
 632                 // the last piece's writer will be deleted in dataWritten() method
 633                 delete pieceWriters;
 634                 pieceWriters = 0;
 635
 636                 dataWritten(writer);
 637         }
 638 }
 639 /* }}} */
 640
 641 /* {{{ Connection::readerError(Tairon::Net::Reader *, Tairon::Net::Socket *) */
 642 void Connection::readerError(Tairon::Net::Reader *, Tairon::Net::Socket *)
 643 {
 644         closedSignal.emit(this);
 645 }
 646 /* }}} */
 647
 648 /* {{{ Connection::sendChoke() */
 649 void Connection::sendChoke()
 650 {
 651         DEBUG("sending choke");
 652
 653         if (peerChoked)
 654                 return;
 655
 656         peersRequests.clear();
 657
 658         peerChoked = true;
 659         Tairon::Net::SWriter *writer = new Tairon::Net::SWriter(String("\x00\x00\x00\x01\x00", 5), socket, wlimiter);
 660         writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 661         writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 662
 663         addCommandWriter(writer);
 664 }
 665 /* }}} */
 666
 667 /* {{{ Connection::sendHandshake(const String &) */
 668 void Connection::sendHandshake(const String &infoHash)
 669 {
 670         char handshake[68];
 671         handshake[0] = (char) 19;
 672         memcpy(handshake + 1, protocolName, 19);
 673         memset(handshake + 20, 0, 8);
 674         memcpy(handshake + 28, infoHash.data(), 20);
 675         memcpy(handshake + 48, TorrentManager::self()->getClientID().data(), 20);
 676
 677         String h(handshake, 68);
 678
 679         Tairon::Net::Writer *writer = new Tairon::Net::SWriter(h, socket, wlimiter);
 680         writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 681         writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 682
 683         addCommandWriter(writer);
 684
 685         if (client->getStorage()->hasSomething()) { // we have already downloaded some piece, send bitfield as well
 686                 DEBUG("sending bitfield");
 687                 // count length of the bitfield
 688                 size_t length = client->getStorage()->getBitField()->getLength();
 689                 length = length / 8 + (length % 8 ? 1 : 0);
 690
 691                 char *bf = new char[5 + length];
 692
 693                 uint32_t l = htonl(1 + length); // 1 byte message type, rest is bitfield
 694                 memcpy(bf, (const char *) (&l), 4);
 695                 bf[4] = 5; // bitfield message
 696                 memcpy(bf + 5, client->getStorage()->getBitField()->getData(), length);
 697
 698                 String b(bf, 5 + length);
 699                 writer = new Tairon::Net::SWriter(b, socket, wlimiter);
 700                 writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 701                 writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 702
 703                 addCommandWriter(writer);
 704
 705                 delete [] bf;
 706         }
 707 }
 708 /* }}} */
 709
 710 /* {{{ Connection::sendHave(uint32_t) */
 711 void Connection::sendHave(uint32_t index)
 712 {
 713         DEBUG("sending have");
 714
 715         uint32_t i = htonl(index);
 716         Tairon::Net::SWriter *writer = new Tairon::Net::SWriter(String("\x00\x00\x00\x05\x04", 5) + String((const char *) (&i), 4), socket, wlimiter);
 717         writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 718         writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 719
 720         addCommandWriter(writer);
 721 }
 722 /* }}} */
 723
 724 /* {{{ Connection::sendInterested() */
 725 void Connection::sendInterested()
 726 {
 727         DEBUG("sending interested");
 728
 729         if (interested)
 730                 return;
 731
 732         interested = true;
 733         Tairon::Net::SWriter *writer = new Tairon::Net::SWriter(String("\x00\x00\x00\x01\x02", 5), socket, wlimiter);
 734         writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 735         writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 736
 737         addCommandWriter(writer);
 738 }
 739 /* }}} */
 740
 741 /* {{{ Connection::sendNotInterested() */
 742 void Connection::sendNotInterested()
 743 {
 744         DEBUG("sending not interested");
 745
 746         if (!interested)
 747                 return;
 748
 749         interested = false;
 750         Tairon::Net::SWriter *writer = new Tairon::Net::SWriter(String("\x00\x00\x00\x01\x03", 5), socket, wlimiter);
 751         writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 752         writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 753
 754         addCommandWriter(writer);
 755 }
 756 /* }}} */
 757
 758 /* {{{ Connection::sendRequest(uint32_t, uint32_t, uint32_t) */
 759 void Connection::sendRequest(uint32_t index, uint32_t start, uint32_t length)
 760 {
 761         requests[index].insert(start);
 762
 763         char msg[17];
 764         memcpy(msg, "\x00\x00\x00\x0d\x06", 5);
 765         *(uint32_t *) (msg + 5) = htonl(index);
 766         *(uint32_t *) (msg + 9) = htonl(start);
 767         *(uint32_t *) (msg + 13) = htonl(length);
 768
 769         Tairon::Net::SWriter *writer = new Tairon::Net::SWriter(String(msg, 17), socket, wlimiter);
 770         writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 771         writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 772
 773         addCommandWriter(writer);
 774 }
 775 /* }}} */
 776
 777 /* {{{ Connection::sendUnchoke(int) */
 778 void Connection::sendUnchoke(int time)
 779 {
 780         DEBUG("sending unchoke");
 781
 782         if (!peerChoked)
 783                 return;
 784
 785         unchokeTime = time;
 786
 787         peerChoked = false;
 788         Tairon::Net::SWriter *writer = new Tairon::Net::SWriter(String("\x00\x00\x00\x01\x01", 5), socket, wlimiter);
 789         writer->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::dataWritten));
 790         writer->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 791
 792         addCommandWriter(writer);
 793 }
 794 /* }}} */
 795
 796 /* {{{ Connection::setNextReader(Tairon::Net::Reader *) */
 797 void Connection::setNextReader(Tairon::Net::Reader *)
 798 {
 799         delete pieceReaders->readers.front();
 800         pieceReaders->readers.pop_front();
 801
 802         if (pieceReaders->readers.size()) {
 803                 DEBUG("Setting next piece reader");
 804                 reader = pieceReaders->readers.front();
 805         } else {
 806                 DEBUG("Piece downloaded");
 807
 808                 incomingRate->update(pieceReaders->length);
 809
 810                 bool fake = pieceReaders->fake;
 811                 delete pieceReaders;
 812                 pieceReaders = 0;
 813                 reader = lengthReader;
 814                 deleteReader = false;
 815                 if (!fake)
 816                         client->pieceDownloaded(pieceIndex, pieceStart, this);
 817         }
 818 }
 819 /* }}} */
 820
 821 /* {{{ Connection::setReaders(PieceReadersStruct *) */
 822 void Connection::setReaders(PieceReadersStruct *readers)
 823 {
 824         for (std::list<Tairon::Net::Reader *>::iterator it = readers->readers.begin(); it != readers->readers.end(); ++it) {
 825                 (*it)->bufferFullSignal.connect(Tairon::Core::methodFunctor(this, &Connection::setNextReader));
 826                 (*it)->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::readerError));
 827         }
 828         pieceReaders = readers;
 829         reader = pieceReaders->readers.front();
 830 }
 831 /* }}} */
 832
 833 /* {{{ Connection::setWriters(PieceWritersStruct *) */
 834 void Connection::setWriters(PieceWritersStruct *writers)
 835 {
 836         for (std::list<Tairon::Net::Writer *>::iterator it = writers->writers.begin(); it != writers->writers.end(); ++it) {
 837                 (*it)->dataWrittenSignal.connect(Tairon::Core::methodFunctor(this, &Connection::pieceDataWritten));
 838                 (*it)->errorSignal.connect(Tairon::Core::methodFunctor(this, &Connection::writerError));
 839         }
 840
 841         pieceWriters = writers;
 842         writers->writers.front()->write();
 843 }
 844 /* }}} */
 845
 846 /* {{{ Connection::socketConnected(Tairon::Net::Socket *) */
 847 void Connection::socketConnected(Tairon::Net::Socket *)
 848 {
 849         socket->connectedSignal.clear();
 850         socket->errorSignal.clear();
 851
 852         init();
 853
 854         sendHandshake(client->getInfoHash());
 855 }
 856 /* }}} */
 857
 858 /* {{{ Connection::socketError(Tairon::Net::Socket *, int) */
 859 void Connection::socketError(Tairon::Net::Socket *, int)
 860 {
 861         socket->close();
 862         closedSignal.emit(this);
 863 }
 864 /* }}} */
 865
 866 /* {{{ Connection::writerError(Tairon::Net::Writer *, Tairon::Net::Socket *) */
 867 void Connection::writerError(Tairon::Net::Writer *, Tairon::Net::Socket *)
 868 {
 869         closedSignal.emit(this);
 870 }
 871 /* }}} */
 872
 873 }; // namespace Main
 874
 875 }; // namespace Tairent
 876
 877 // vim: ai sw=4 ts=4 noet fdm=marker