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