public interface PeerInformation {

  /**
   * @return {@link InetSocketAddress} of remote peer
   */
  InetSocketAddress getAddress();

  /**
   * @return id of current peer which the peers sent in the handshake
   */
  byte[] getId();

  /**
   * @return client identifier of current peer
   */
  String getClientIdentifier();

  /**
   * @return client version of current peer
   */
  int getClientVersion();

} public class Piece implements Comparable<Piece>, PieceInformation {

  private static final Logger logger =
          TorrentLoggerFactory.getLogger(Piece.class);

  private final PieceStorage pieceStorage;
  private final int index;
  private final long length;
  private final byte[] hash;

  private volatile boolean valid;
  private int seen;
  private ByteBuffer data;

  /**
   * Initialize a new piece in the byte bucket.
   *
   * @param pieceStorage  The underlying piece storage bucket.
   * @param index   This piece index in the torrent.
   * @param length  This piece length, in bytes.
   * @param hash    This piece 20-byte SHA1 hash sum.
   */
  public Piece(PieceStorage pieceStorage, int index, long length, byte[] hash) {
    this.pieceStorage = pieceStorage;
    this.index = index;
    this.length = length;
    this.hash = hash;

    // Piece is considered invalid until first check.
    this.valid = false;

    // Piece start unseen
    this.seen = 0;

    this.data = null;
  }

  @Override
  public int getSize() {
    return (int)length;
  }

  /**
   * Tells whether this piece's data is valid or not.
   */
  public boolean isValid() {
    return this.valid;
  }

  /**
   * Returns the index of this piece in the torrent.
   */
  public int getIndex() {
    return this.index;
  }

  /**
   * Returns the size, in bytes, of this piece.
   *
   * <p>
   * All pieces, except the last one, are expected to have the same size.
   * </p>
   */
  public long size() {
    return this.length;
  }

  /**
   * Tells whether this piece is available in the current connected peer swarm.
   */
  public boolean available() {
    return this.seen > 0;
  }

  /**
   * Mark this piece as being seen at the given peer.
   *
   * @param peer The sharing peer this piece has been seen available at.
   */
  public void seenAt(SharingPeer peer) {
    this.seen++;
  }

  /**
   * Mark this piece as no longer being available at the given peer.
   *
   * @param peer The sharing peer from which the piece is no longer available.
   */
  public void noLongerAt(SharingPeer peer) {
    this.seen--;
  }

  void setValid(boolean valid) {
    this.valid = valid;
  }

  /**
   * Validates this piece.
   *
   * @return Returns true if this piece, as stored in the underlying byte
   * storage, is valid, i.e. its SHA1 sum matches the one from the torrent
   * meta-info.
   */
  public boolean validate(SharedTorrent torrent, Piece piece) throws IOException {

    logger.trace("Validating {}...", this);

    // TODO: remove cast to int when large ByteBuffer support is
    // implemented in Java.
    byte[] pieceBytes = data.array();
    final byte[] calculatedHash = TorrentUtils.calculateSha1Hash(pieceBytes);
    this.valid = Arrays.equals(calculatedHash, this.hash);
    logger.trace("validating result of piece {} is {}", this.index, this.valid);

    return this.isValid();
  }

  /**
   * Internal piece data read function.
   *
   * <p>
   * This function will read the piece data without checking if the piece has
   * been validated. It is simply meant at factoring-in the common read code
   * from the validate and read functions.
   * </p>
   *
   * @param offset Offset inside this piece where to start reading.
   * @param length Number of bytes to read from the piece.
   * @return A byte buffer containing the piece data.
   * @throws IllegalArgumentException If <em>offset + length</em> goes over
   *                                  the piece boundary.
   * @throws IOException              If the read can't be completed (I/O error, or EOF
   *                                  reached, which can happen if the piece is not complete).
   */
  private ByteBuffer _read(long offset, long length, ByteBuffer buffer) throws IOException {
    if (offset + length > this.length) {
      throw new IllegalArgumentException("Piece#" + this.index +
              " overrun (" + offset + " + " + length + " > " +
              this.length + ") !");
    }

    // TODO: remove cast to int when large ByteBuffer support is
    // implemented in Java.
    int position = buffer.position();
    byte[] bytes = this.pieceStorage.readPiecePart(this.index, (int)offset, (int)length);
    buffer.put(bytes);
    buffer.rewind();
    buffer.limit(bytes.length + position);
    return buffer;
  }

  /**
   * Read a piece block from the underlying byte storage.
   *
   * <p>
   * This is the public method for reading this piece's data, and it will
   * only succeed if the piece is complete and valid on disk, thus ensuring
   * any data that comes out of this function is valid piece data we can send
   * to other peers.
   * </p>
   *
   * @param offset Offset inside this piece where to start reading.
   * @param length Number of bytes to read from the piece.
   * @return A byte buffer containing the piece data.
   * @throws IllegalArgumentException If <em>offset + length</em> goes over
   *                                  the piece boundary.
   * @throws IllegalStateException    If the piece is not valid when attempting
   *                                  to read it.
   * @throws IOException              If the read can't be completed (I/O error, or EOF
   *                                  reached, which can happen if the piece is not complete).
   */
  public ByteBuffer read(long offset, int length, ByteBuffer block)
          throws IllegalArgumentException, IllegalStateException, IOException {
    if (!this.valid) {
      throw new IllegalStateException("Attempting to read an " +
              "known-to-be invalid piece!");
    }

    return this._read(offset, length, block);
  }

  /**
   * Record the given block at the given offset in this piece.
   *
   * @param block  The ByteBuffer containing the block data.
   * @param offset The block offset in this piece.
   */
  public void record(ByteBuffer block, int offset) {
    if (this.data == null) {
      // TODO: remove cast to int when large ByteBuffer support is
      // implemented in Java.
      this.data = ByteBuffer.allocate((int) this.length);
    }

    int pos = block.position();
    this.data.position(offset);
    this.data.put(block);
    block.position(pos);
  }

  public void finish() throws IOException {
    this.data.rewind();
    logger.trace("Recording {}...", this);
    try {
      pieceStorage.savePiece(index, this.data.array());
    } finally {
      this.data = null;
    }
  }

  /**
   * Return a human-readable representation of this piece.
   */
  public String toString() {
    return String.format("piece#%4d%s",
            this.index,
            this.isValid() ? "+" : "-");
  }

  @Override
  public boolean equals(Object obj) {
    if (obj instanceof Piece) {
      return this.index == ((Piece) obj).index;
    } else {
      return false;
    }
  }

  /**
   * Piece comparison function for ordering pieces based on their
   * availability.
   *
   * @param other The piece to compare with, should not be <em>null</em>.
   */
  public int compareTo(Piece other) {
    // return true for the same pieces, otherwise sort by time seen, then by index;
    if (this.equals(other)) {
      return 0;
    } else if (this.seen == other.seen) {
      return new Integer(this.index).compareTo(other.index);
    } else if (this.seen < other.seen) {
      return -1;
    } else {
      return 1;
    }
  }

} public class SharedTorrent implements PeerActivityListener, TorrentMetadata, TorrentInfo {

  private static final Logger logger =
          TorrentLoggerFactory.getLogger(SharedTorrent.class);

  private final static RequestStrategy DEFAULT_REQUEST_STRATEGY = new RequestStrategyImplAnyInteresting();

  /**
   * End-game trigger ratio.
   *
   * <p>
   * Eng-game behavior (requesting already requested pieces from available
   * and ready peers to try to speed-up the end of the transfer) will only be
   * enabled when the ratio of completed pieces over total pieces in the
   * torrent is over this value.
   * </p>
   */
  private static final float ENG_GAME_COMPLETION_RATIO = 0.95f;
  private static final int END_GAME_STATIC_PIECES_COUNT = 20;
  private static final long END_GAME_INVOCATION_PERIOD_MS = 2000;

  private final TorrentStatistic myTorrentStatistic;

  private long myLastAnnounceTime = -1;
  private int mySeedersCount = 0;

  private final PieceStorage pieceStorage;
  private boolean isFileChannelOpen = false;
  private final Map<Integer, Future<?>> myValidationFutures;
  private final TorrentMetadata myTorrentMetadata;
  private final long myTorrentTotalSize;

  private final int pieceLength;
  private final ByteBuffer piecesHashes;

  private boolean initialized;
  private Piece[] pieces;
  private final BitSet completedPieces;
  private final BitSet requestedPieces;
  private final RequestStrategy myRequestStrategy;
  private final EventDispatcher eventDispatcher;

  private final List<SharingPeer> myDownloaders = new CopyOnWriteArrayList<SharingPeer>();
  private final EndGameStrategy endGameStrategy = new EndGameStrategyImpl(2);
  private volatile long endGameEnabledOn = -1;

  private volatile ClientState clientState = ClientState.WAITING;
  private static final int MAX_VALIDATION_TASK_COUNT = 200;
  private static final int MAX_REQUESTED_PIECES_PER_TORRENT = 100;

  /**
   * Create a new shared torrent from meta-info
   *
   * @param torrentMetadata The meta-info
   * @param eventDispatcher
   */
  public SharedTorrent(TorrentMetadata torrentMetadata, PieceStorage pieceStorage, RequestStrategy requestStrategy,
                       TorrentStatistic torrentStatistic, EventDispatcher eventDispatcher) {
    myTorrentMetadata = torrentMetadata;
    this.pieceStorage = pieceStorage;
    this.eventDispatcher = eventDispatcher;
    myTorrentStatistic = torrentStatistic;
    myValidationFutures = new HashMap<Integer, Future<?>>();
    long totalSize = 0;
    for (TorrentFile torrentFile : myTorrentMetadata.getFiles()) {
      totalSize += torrentFile.size;
    }
    myTorrentTotalSize = totalSize;
    this.myRequestStrategy = requestStrategy;

    this.pieceLength = myTorrentMetadata.getPieceLength();
    this.piecesHashes = ByteBuffer.wrap(myTorrentMetadata.getPiecesHashes());

    if (this.piecesHashes.capacity() / Constants.PIECE_HASH_SIZE *
            (long) this.pieceLength < myTorrentTotalSize) {
      throw new IllegalArgumentException("Torrent size does not " +
              "match the number of pieces and the piece size!");
    }

    this.initialized = false;
    this.pieces = new Piece[0];
    this.completedPieces = new BitSet(torrentMetadata.getPiecesCount());
    this.requestedPieces = new BitSet();
  }

  public static SharedTorrent fromFile(File source, PieceStorage pieceStorage, TorrentStatistic torrentStatistic)
          throws IOException {
    byte[] data = FileUtils.readFileToByteArray(source);
    TorrentMetadata torrentMetadata = new TorrentParser().parse(data);
    return new SharedTorrent(torrentMetadata, pieceStorage, DEFAULT_REQUEST_STRATEGY, torrentStatistic, new EventDispatcher());
  }

  private synchronized void closeFileChannelIfNecessary() throws IOException {
    if (isFileChannelOpen && myDownloaders.size() == 0) {
      logger.debug("Closing file  channel for {} if necessary. Downloaders: {}", getHexInfoHash(), myDownloaders.size());
      this.pieceStorage.close();
      isFileChannelOpen = false;
    }
  }

  /**
   * Get the number of bytes uploaded for this torrent.
   */
  public long getUploaded() {
    return myTorrentStatistic.getUploadedBytes();
  }

  /**
   * Get the number of bytes downloaded for this torrent.
   * <p/>
   * <p>
   * <b>Note:</b> this could be more than the torrent's length, and should
   * not be used to determine a completion percentage.
   * </p>
   */
  public long getDownloaded() {
    return myTorrentStatistic.getDownloadedBytes();
  }

  /**
   * Get the number of bytes left to download for this torrent.
   */
  public long getLeft() {
    return myTorrentStatistic.getLeftBytes();
  }

  public int getSeedersCount() {
    return mySeedersCount;
  }

  public void setSeedersCount(int seedersCount) {
    mySeedersCount = seedersCount;
  }

  public long getLastAnnounceTime() {
    return myLastAnnounceTime;
  }

  public void setLastAnnounceTime(long lastAnnounceTime) {
    myLastAnnounceTime = lastAnnounceTime;
  }

  /**
   * Tells whether this torrent has been fully initialized yet.
   */
  public boolean isInitialized() {
    return this.initialized;
  }

  /**
   * Stop the torrent initialization as soon as possible.
   */
  public void stop() {
  }

  /**
   * Build this torrent's pieces array.
   * <p/>
   * <p>
   * Hash and verify any potentially present local data and create this
   * torrent's pieces array from their respective hash provided in the
   * torrent meta-info.
   * </p>
   * <p/>
   * <p>
   * This function should be called soon after the constructor to initialize
   * the pieces array.
   * </p>
   */
  public synchronized void init() throws InterruptedException, IOException {
    setClientState(ClientState.VALIDATING);

    if (this.isInitialized()) {
      throw new IllegalStateException("Torrent was already initialized!");
    }

    hashSingleThread();

    this.initialized = true;
  }

  private void initPieces() {
    int nPieces = (int) (Math.ceil(
            (double) myTorrentTotalSize / this.pieceLength));
    this.pieces = new Piece[nPieces];
    this.piecesHashes.clear();
  }

  private void hashSingleThread() {
    initPieces();

    logger.debug("Analyzing local data for {} with {} threads...",
            myTorrentMetadata.getDirectoryName(), TorrentCreator.HASHING_THREADS_COUNT);
    for (int idx = 0; idx < this.pieces.length; idx++) {
      byte[] hash = new byte[Constants.PIECE_HASH_SIZE];
      this.piecesHashes.get(hash);

      // The last piece may be shorter than the torrent's global piece
      // length. Let's make sure we get the right piece length in any
      // situation.
      long off = ((long) idx) * this.pieceLength;
      long len = Math.min(
              myTorrentTotalSize - off,
              this.pieceLength);

      Piece piece = new Piece(this.pieceStorage, idx, len, hash
      );
      this.pieces[idx] = piece;
      piece.setValid(pieceStorage.getAvailablePieces().get(idx));

      if (piece.isValid()) {
        this.completedPieces.set(piece.getIndex());
      }
    }
  }

  public synchronized void close() {
    logger.trace("Closing torrent", myTorrentMetadata.getDirectoryName());
    try {
      this.pieceStorage.close();
      isFileChannelOpen = false;
    } catch (IOException ioe) {
      logger.error("Error closing torrent byte storage: {}",
              ioe.getMessage());
    }
  }

  public synchronized void closeFully() {
    logger.trace("Closing torrent", myTorrentMetadata.getDirectoryName());
    try {
      this.pieceStorage.closeFully();
      isFileChannelOpen = false;
    } catch (IOException ioe) {
      logger.error("Error closing torrent byte storage: {}",
              ioe.getMessage());
    }
  }

  /**
   * Retrieve a piece object by index.
   *
   * @param index The index of the piece in this torrent.
   */
  public Piece getPiece(int index) {
    if (this.pieces == null) {
      throw new IllegalStateException("Torrent not initialized yet.");
    }

    if (index >= this.pieces.length) {
      throw new IllegalArgumentException("Invalid piece index!");
    }

    return this.pieces[index];
  }

  /**
   * Return a copy of the bit field of available pieces for this torrent.
   * <p/>
   * <p>
   * Available pieces are pieces available in the swarm, and it does not
   * include our own pieces.
   * </p>
   */
  public BitSet getAvailablePieces() {
    if (!this.isInitialized()) {
      throw new IllegalStateException("Torrent not yet initialized!");
    }

    BitSet availablePieces = new BitSet(this.pieces.length);

    synchronized (this.pieces) {
      for (Piece piece : this.pieces) {
        if (piece.available()) {
          availablePieces.set(piece.getIndex());
        }
      }
    }

    return availablePieces;
  }

  /**
   * Return a copy of the completed pieces bitset.
   */
  public BitSet getCompletedPieces() {
    if (!this.isInitialized()) {
      throw new IllegalStateException("Torrent not yet initialized!");
    }

    return pieceStorage.getAvailablePieces();
  }

  /**
   * Tells whether this torrent has been fully downloaded, or is fully
   * available locally.
   */
  public synchronized boolean isComplete() {
    return this.pieces.length > 0
            && pieceStorage.getAvailablePieces().cardinality() == myTorrentMetadata.getPiecesCount();
  }

  /**
   * Finalize the download of this torrent.
   * <p/>
   * <p>
   * This realizes the final, pre-seeding phase actions on this torrent,
   * which usually consists in putting the torrent data in their final form
   * and at their target location.
   * </p>
   *
   * @see TorrentByteStorage#finish
   */
  public synchronized void finish() {
    if (!this.isInitialized()) {
      throw new IllegalStateException("Torrent not yet initialized!");
    }

    if (!this.isComplete()) {
      throw new IllegalStateException("Torrent download is not complete!");
    }

    eventDispatcher.multicaster().downloadComplete();
    setClientState(ClientState.SEEDING);
  }

  public boolean isFinished() {
    return pieceStorage.getAvailablePieces().cardinality() == myTorrentMetadata.getPiecesCount();
  }

  public ClientState getClientState() {
    return this.clientState;
  }

  public void setClientState(ClientState clientState) {
    this.clientState = clientState;
  }

  /**
   * Mark a piece as completed, decrementing the piece size in bytes from our
   * left bytes to download counter.
   */
  public synchronized void markCompleted(Piece piece) {
    if (this.completedPieces.get(piece.getIndex())) {
      return;
    }

    // A completed piece means that's that much data left to download for
    // this torrent.
    myTorrentStatistic.addLeft(-piece.size());
    this.completedPieces.set(piece.getIndex());
    if (completedPieces.cardinality() == getPiecesCount()) {
      logger.info("all pieces are received for torrent {}. Validating...", this);
    }
  }

  public synchronized void markUncompleted(Piece piece) {
    if (!this.completedPieces.get(piece.getIndex())) {
      return;
    }

    removeValidationFuture(piece);
    myTorrentStatistic.addLeft(piece.size());
    this.completedPieces.clear(piece.getIndex());
  }

  public synchronized void removeValidationFuture(Piece piece) {
    myValidationFutures.remove(piece.getIndex());
  }

  public void notifyPieceDownloaded(Piece piece, SharingPeer peer) {
    eventDispatcher.multicaster().pieceDownloaded(piece, peer);
  }

  /** PeerActivityListener handler(s). *************************************/

  /**
   * Peer choked handler.
   * <p/>
   * <p>
   * When a peer chokes, the requests made to it are canceled and we need to
   * mark the eventually piece we requested from it as available again for
   * download tentative from another peer.
   * </p>
   *
   * @param peer The peer that choked.
   */
  @Override
  public synchronized void handlePeerChoked(SharingPeer peer) {
    Set<Piece> pieces = peer.getRequestedPieces();

    if (pieces.size() > 0) {
      for (Piece piece : pieces) {
        this.requestedPieces.set(piece.getIndex(), false);
      }
    }

    logger.trace("Peer {} choked, we now have {} outstanding " +
                    "request(s): {}.",
            new Object[]{
                    peer,
                    this.requestedPieces.cardinality(),
                    this.requestedPieces
            });
  }

  /**
   * Peer ready handler.
   * <p/>
   * <p>
   * When a peer becomes ready to accept piece block requests, select a piece
   * to download and go for it.
   * </p>
   *
   * @param peer The peer that became ready.
   */
  @Override
  public void handlePeerReady(SharingPeer peer) {
    initIfNecessary(peer);

    RequestsCollection requestsCollection = getRequestsCollection(peer);
    requestsCollection.sendAllRequests();
  }

  @NotNull
  private synchronized RequestsCollection getRequestsCollection(final SharingPeer peer) {
    if (myValidationFutures.size() > MAX_VALIDATION_TASK_COUNT) return RequestsCollection.Empty.INSTANCE;

    if (this.requestedPieces.cardinality() > MAX_REQUESTED_PIECES_PER_TORRENT) return RequestsCollection.Empty.INSTANCE;

    int completedAndValidated = pieceStorage.getAvailablePieces().cardinality();

    boolean turnOnEndGame = completedAndValidated > getPiecesCount() * ENG_GAME_COMPLETION_RATIO ||
            completedAndValidated > getPiecesCount() - END_GAME_STATIC_PIECES_COUNT;
    if (turnOnEndGame) {
      long now = System.currentTimeMillis();
      if (now - END_GAME_INVOCATION_PERIOD_MS > endGameEnabledOn) {
        endGameEnabledOn = now;
        logger.info("Running end-game mode, currently available {}/{} pieces",
                pieceStorage.getAvailablePieces().cardinality(),
                getPieceCount());
        return endGameStrategy.collectRequests(pieces, myDownloaders);
      }
      return RequestsCollection.Empty.INSTANCE;
    }

    final BitSet interesting = peer.getAvailablePieces();
    interesting.andNot(this.completedPieces);
    interesting.andNot(this.requestedPieces);

    int maxRequestingPieces = Math.min(10, interesting.cardinality());
    int currentlyDownloading = peer.getDownloadingPiecesCount();
    Map<Piece, List<SharingPeer>> toRequest = new HashMap<Piece, List<SharingPeer>>();
    while (currentlyDownloading < maxRequestingPieces) {
      if (!peer.isConnected()) {
        break;
      }

      if (interesting.cardinality() == 0) {
        return RequestsCollection.Empty.INSTANCE;
      }

      Piece chosen = myRequestStrategy.choosePiece(interesting, pieces);
      if (chosen == null) {
        logger.info("chosen piece is null");
        break;
      }
      this.requestedPieces.set(chosen.getIndex());
      currentlyDownloading++;
      toRequest.put(chosen, Collections.singletonList(peer));
      interesting.clear(chosen.getIndex());
    }

    return new RequestsCollectionImpl(toRequest);
  }

  public synchronized void initIfNecessary(SharingPeer peer) {
    if (!isInitialized()) {
      try {
        init();
      } catch (InterruptedException e) {
        logger.info("Interrupted init", e);
        peer.unbind(true);
        return;
      } catch (IOException e) {
        logger.info("IOE during init", e);
        peer.unbind(true);
        return;
      }
    }
  }

  /**
   * Piece availability handler.
   * <p/>
   * <p>
   * Handle updates in piece availability from a peer's HAVE message. When
   * this happens, we need to mark that piece as available from the peer.
   * </p>
   *
   * @param peer  The peer we got the update from.
   * @param piece The piece that became available.
   */
  @Override
  public void handlePieceAvailability(SharingPeer peer, Piece piece) {
    boolean isPeerInteresting = !this.completedPieces.get(piece.getIndex()) &&
            !this.requestedPieces.get(piece.getIndex());
    if (isPeerInteresting) {
      peer.interesting();
    }

    piece.seenAt(peer);

    logger.trace("Peer {} contributes {} piece(s) [{}/{}/{}].",
            new Object[]{
                    peer,
                    peer.getAvailablePieces().cardinality(),
                    this.completedPieces.cardinality(),
                    this.getAvailablePieces().cardinality(),
                    this.pieces.length
            });

    if (!peer.isChoked() &&
            peer.isInteresting() &&
            !peer.isDownloading()) {
      this.handlePeerReady(peer);
    }
  }

  /**
   * Bit field availability handler.
   * <p/>
   * <p>
   * Handle updates in piece availability from a peer's BITFIELD message.
   * When this happens, we need to mark in all the pieces the peer has that
   * they can be reached through this peer, thus augmenting the global
   * availability of pieces.
   * </p>
   *
   * @param peer            The peer we got the update from.
   * @param availablePieces The pieces availability bit field of the peer.
   */
  @Override
  public void handleBitfieldAvailability(SharingPeer peer,
                                         BitSet availablePieces) {
    // Determine if the peer is interesting for us or not, and notify it.
    BitSet interesting = (BitSet) availablePieces.clone();
    synchronized (this) {
      interesting.andNot(this.completedPieces);
      interesting.andNot(this.requestedPieces);
    }
    // Record the peer has all the pieces it told us it had.
    for (int i = availablePieces.nextSetBit(0); i >= 0;
         i = availablePieces.nextSetBit(i + 1)) {
      this.pieces[i].seenAt(peer);
    }

    if (interesting.cardinality() == 0) {
      peer.notInteresting();
    } else {
      peer.interesting();
    }

    logger.debug("Peer {} contributes {} piece(s), total pieces count: {}.",
            new Object[]{
                    peer,
                    availablePieces.cardinality(),
                    myTorrentMetadata.getPiecesCount()
            });
  }

  public int getDownloadersCount() {
    return myDownloaders.size();
  }

  @Override
  public void afterPeerRemoved(SharingPeer peer) {

  }

  /**
   * Piece upload completion handler.
   * <p/>
   * <p>
   * When a piece has been sent to a peer, we just record that we sent that
   * many bytes. If the piece is valid on the peer's side, it will send us a
   * HAVE message and we'll record that the piece is available on the peer at
   * that moment (see <code>handlePieceAvailability()</code>).
   * </p>
   *
   * @param peer  The peer we got this piece from.
   * @param piece The piece in question.
   */
  @Override
  public void handlePieceSent(SharingPeer peer, Piece piece) {
    logger.trace("Completed upload of {} to {}.", piece, peer);
    myTorrentStatistic.addUploaded(piece.size());
  }

  /**
   * Piece download completion handler.
   * <p/>
   * <p>
   * If the complete piece downloaded is valid, we can record in the torrent
   * completedPieces bit field that we know have this piece.
   * </p>
   *
   * @param peer  The peer we got this piece from.
   * @param piece The piece in question.
   */
  @Override
  public void handlePieceCompleted(SharingPeer peer,
                                   Piece piece) throws IOException {
    // Regardless of validity, record the number of bytes downloaded and
    // mark the piece as not requested anymore
    myTorrentStatistic.addDownloaded(piece.size());
    this.requestedPieces.set(piece.getIndex(), false);

    logger.trace("We now have {} piece(s) and {} outstanding request(s): {}",
            new Object[]{
                    this.completedPieces.cardinality(),
                    this.requestedPieces.cardinality(),
                    this.requestedPieces
            });
  }

  /**
   * Peer disconnection handler.
   * <p/>
   * <p>
   * When a peer disconnects, we need to mark in all of the pieces it had
   * available that they can't be reached through this peer anymore.
   * </p>
   *
   * @param peer The peer we got this piece from.
   */
  @Override
  public synchronized void handlePeerDisconnected(SharingPeer peer) {
    BitSet availablePieces = peer.getAvailablePieces();

    for (int i = availablePieces.nextSetBit(0); i >= 0;
         i = availablePieces.nextSetBit(i + 1)) {
      this.pieces[i].noLongerAt(peer);
    }

    Set<Piece> requested = peer.getRequestedPieces();
    if (requested != null) {
      for (Piece piece : requested) {
        this.requestedPieces.set(piece.getIndex(), false);
      }
    }

    myDownloaders.remove(peer);

    try {
      closeFileChannelIfNecessary();
    } catch (IOException e) {
      logger.info("I/O error on attempt to close file storage: " + e.toString());
    }

    logger.debug("Peer {} went away with {} piece(s) [{}/{}].",
            new Object[]{
                    peer,
                    availablePieces.cardinality(),
                    this.completedPieces.cardinality(),
                    this.pieces.length
            });
    logger.trace("We now have {} piece(s) and {} outstanding request(s): {}",
            new Object[]{
                    this.completedPieces.cardinality(),
                    this.requestedPieces.cardinality(),
                    this.requestedPieces
            });
    eventDispatcher.multicaster().peerDisconnected(peer);
  }

  @Override
  public synchronized void handleIOException(SharingPeer peer,
                                             IOException ioe) {
    eventDispatcher.multicaster().downloadFailed(ioe);
  }

  @Override
  public synchronized void handleNewPeerConnected(SharingPeer peer) {
    initIfNecessary(peer);
    eventDispatcher.multicaster().peerConnected(peer);
  }

  @Override
  public String toString() {
    return "SharedTorrent{" +
            Arrays.toString(TorrentUtils.getTorrentFileNames(myTorrentMetadata).toArray()) +
            "}";
  }

  @Override
  public String getDirectoryName() {
    return myTorrentMetadata.getDirectoryName();
  }

  @Override
  public List<TorrentFile> getFiles() {
    return myTorrentMetadata.getFiles();
  }

  @Nullable
  @Override
  public List<List<String>> getAnnounceList() {
    return myTorrentMetadata.getAnnounceList();
  }

  @Nullable
  @Override
  public String getAnnounce() {
    return myTorrentMetadata.getAnnounce();
  }

  @Override
  public Optional<Long> getCreationDate() {
    return myTorrentMetadata.getCreationDate();
  }

  @Override
  public Optional<String> getComment() {
    return myTorrentMetadata.getComment();
  }

  @Override
  public Optional<String> getCreatedBy() {
    return myTorrentMetadata.getCreatedBy();
  }

  @Override
  public int getPieceLength() {
    return myTorrentMetadata.getPieceLength();
  }

  @Override
  public byte[] getPiecesHashes() {
    return myTorrentMetadata.getPiecesHashes();
  }

  @Override
  public boolean isPrivate() {
    return myTorrentMetadata.isPrivate();
  }

  @Override
  public int getPiecesCount() {
    return myTorrentMetadata.getPiecesCount();
  }

  @Override
  public byte[] getInfoHash() {
    return myTorrentMetadata.getInfoHash();
  }

  @Override
  public String getHexInfoHash() {
    return myTorrentMetadata.getHexInfoHash();
  }

  @Override
  public int getPieceCount() {
    return getPiecesCount();
  }

  @Override
  public long getPieceSize(int pieceIdx) {
    return getPieceLength();
  }

  public synchronized void savePieceAndValidate(Piece p) throws IOException {
//    p.finish();
  }

  public synchronized void markCompletedAndAddValidationFuture(Piece piece, Future<?> validationFuture) {
    this.markCompleted(piece);
    myValidationFutures.put(piece.getIndex(), validationFuture);
  }

  public synchronized boolean isAllPiecesOfPeerCompletedAndValidated(SharingPeer peer) {
    final BitSet availablePieces = peer.getAvailablePieces();
    for (Piece piece : pieces) {
      final boolean peerHaveCurrentPiece = availablePieces.get(piece.getIndex());
      if (!peerHaveCurrentPiece) continue;
      if (!completedPieces.get(piece.getIndex())) return false;
      if (myValidationFutures.get(piece.getIndex()) != null) return false;
    }
    return true;
  }

  public void addConnectedPeer(SharingPeer sharingPeer) {
    myDownloaders.add(sharingPeer);
  }
} public final class LoggerUtils {

  public static void warnAndDebugDetails(Logger logger, String message, Throwable t) {
    logger.warn(message);
    logger.debug("", t);
  }

  public static void warnAndDebugDetails(Logger logger, String message, Object arg, Throwable t) {
    logger.warn(message, arg);
    logger.debug("", t);
  }

  public static void warnWithMessageAndDebugDetails(Logger logger, String message, Object arg, Throwable t) {
    logger.warn(message + ": " + (t.getMessage() != null ? t.getMessage() : t.getClass().getName()), arg);
    logger.debug("", t);
  }

  public static void errorAndDebugDetails(Logger logger, String message, Object arg, Throwable t) {
    logger.error(message, arg);
    logger.debug("", t);
  }

  public static void errorAndDebugDetails(Logger logger, String message, Throwable t) {
    logger.error(message);
    logger.debug("", t);
  }

} public class Peer {

  private static final Logger logger = TorrentLoggerFactory.getLogger(Peer.class);

  private final InetSocketAddress address;
  private final String hostId;

  private ByteBuffer peerId;
  private volatile String hexPeerId;
  private volatile String hexInfoHash;

  /**
   * Instantiate a new peer.
   *
   * @param address The peer's address, with port.
   */
  public Peer(InetSocketAddress address) {
    this(address, null);
  }

  /**
   * Instantiate a new peer.
   *
   * @param ip   The peer's IP address.
   * @param port The peer's port.
   */
  public Peer(String ip, int port) {
    this(new InetSocketAddress(ip, port), null);
  }

  /**
   * Instantiate a new peer.
   *
   * @param ip     The peer's IP address.
   * @param port   The peer's port.
   * @param peerId The byte-encoded peer ID.
   */
  public Peer(String ip, int port, ByteBuffer peerId) {
    this(new InetSocketAddress(ip, port), peerId);
  }

  /**
   * Instantiate a new peer.
   *
   * @param address The peer's address, with port.
   * @param peerId  The byte-encoded peer ID.
   */
  public Peer(InetSocketAddress address, ByteBuffer peerId) {
    this.address = address;
    this.hostId = String.format("%s:%d",
            this.address.getAddress(),
            this.address.getPort());

    this.setPeerId(peerId);
  }

  /**
   * Tells whether this peer has a known peer ID yet or not.
   */
  public boolean hasPeerId() {
    return this.peerId != null;
  }

  /**
   * Returns the raw peer ID as a {@link ByteBuffer}.
   */
  public ByteBuffer getPeerId() {
    return this.peerId;
  }

  public byte[] getPeerIdArray() {
    return peerId == null ? null : peerId.array();
  }

  /**
   * Set a peer ID for this peer (usually during handshake).
   *
   * @param peerId The new peer ID for this peer.
   */
  public void setPeerId(ByteBuffer peerId) {
    if (peerId != null) {
      this.peerId = peerId;
      this.hexPeerId = TorrentUtils.byteArrayToHexString(peerId.array());
    } else {
      this.peerId = null;
      this.hexPeerId = null;
    }
  }

  public String getStringPeerId() {
    try {
      return new String(peerId.array(), Constants.BYTE_ENCODING);
    } catch (UnsupportedEncodingException e) {
      LoggerUtils.warnAndDebugDetails(logger, "can not get peer id as string", e);
    }
    return null;
  }

  /**
   * Get the hexadecimal-encoded string representation of this peer's ID.
   */
  public String getHexPeerId() {
    return this.hexPeerId;
  }

  /**
   * Get the shortened hexadecimal-encoded peer ID.
   */
  public String getShortHexPeerId() {
    return String.format("..%s",
            this.hexPeerId.substring(this.hexPeerId.length() - 6).toUpperCase());
  }

  /**
   * Returns this peer's IP address.
   */
  public String getIp() {
    return this.address.getAddress().getHostAddress();
  }

  /**
   * Returns this peer's InetAddress.
   */
  public InetSocketAddress getAddress() {
    return this.address;
  }

  /**
   * Returns this peer's port number.
   */
  public int getPort() {
    return this.address.getPort();
  }

  /**
   * Returns this peer's host identifier ("host:port").
   */
  public String getHostIdentifier() {
    return this.hostId;
  }

  /**
   * Returns a binary representation of the peer's IP.
   */
  public byte[] getRawIp() {
    final InetAddress address = this.address.getAddress();
    if (address == null) return null;
    return address.getAddress();
  }


  /**
   * Tells if two peers seem to look alike (i.e. they have the same IP, port
   * and peer ID if they have one).
   */
  public boolean looksLike(Peer other) {
    if (other == null) {
      return false;
    }

    return this.hostId.equals(other.hostId) && this.getPort() == other.getPort();
  }

  public void setTorrentHash(String hexInfoHash) {
    this.hexInfoHash = hexInfoHash;
  }

  public String getHexInfoHash() {
    return hexInfoHash;
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;

    Peer peer = (Peer) o;

    if (hexPeerId == null && peer.hexPeerId == null) return super.equals(o);

    if (hexPeerId != null ? !hexPeerId.equals(peer.hexPeerId) : peer.hexPeerId != null) return false;
    return hexInfoHash != null ? hexInfoHash.equals(peer.hexInfoHash) : peer.hexInfoHash == null;
  }

  @Override
  public int hashCode() {

    if (hexPeerId == null) return super.hashCode();

    int result = hexPeerId != null ? hexPeerId.hashCode() : 0;
    result = 31 * result + (hexInfoHash != null ? hexInfoHash.hashCode() : 0);
    return result;
  }

  /**
   * Returns a human-readable representation of this peer.
   */
  @Override
  public String toString() {
    return "Peer " + address + " for torrent " + hexInfoHash;
  }
} public final class TorrentLoggerFactory {

  @Nullable
  private static volatile String staticLoggersName = null;

  public static Logger getLogger(Class<?> clazz) {
    String name = staticLoggersName;
    if (name == null) {
      name = clazz.getName();
    }
    return LoggerFactory.getLogger(name);
  }

  public static void setStaticLoggersName(@Nullable String staticLoggersName) {
    TorrentLoggerFactory.staticLoggersName = staticLoggersName;
  }
} public final class TorrentUtils {

  private final static char[] HEX_SYMBOLS = "0123456789ABCDEF".toCharArray();

  /**
   * @param data for hashing
   * @return sha 1 hash of specified data
   */
  public static byte[] calculateSha1Hash(byte[] data) {
    return DigestUtils.sha1(data);
  }

  /**
   * Convert a byte string to a string containing an hexadecimal
   * representation of the original data.
   *
   * @param bytes The byte array to convert.
   */
  public static String byteArrayToHexString(byte[] bytes) {
    char[] hexChars = new char[bytes.length * 2];
    for (int j = 0; j < bytes.length; j++) {
      int v = bytes[j] & 0xFF;
      hexChars[j * 2] = HEX_SYMBOLS[v >>> 4];
      hexChars[j * 2 + 1] = HEX_SYMBOLS[v & 0x0F];
    }
    return new String(hexChars);
  }

  public static boolean isTrackerLessInfo(AnnounceableInformation information) {
    return information.getAnnounce() == null && information.getAnnounceList() == null;
  }

  public static List<String> getTorrentFileNames(TorrentMetadata metadata) {
    List<String> result = new ArrayList<String>();

    for (TorrentFile torrentFile : metadata.getFiles()) {
      result.add(torrentFile.getRelativePathAsString());
    }

    return result;
  }

} public class ConnectionClosedException extends IOException {

  public ConnectionClosedException() {
  }

  public ConnectionClosedException(Throwable cause) {
    super(cause);
  }
} public class ConnectionManager {

  private static final Logger logger = TorrentLoggerFactory.getLogger(ConnectionManager.class);

  private final Selector selector;
  private final TimeService myTimeService;
  private volatile ConnectionWorker myConnectionWorker;
  private int myBindPort;
  private final ConnectionManagerContext myContext;
  private volatile ServerSocketChannel myServerSocketChannel;
  private volatile Future<?> myWorkerFuture;
  private final NewConnectionAllower myIncomingConnectionAllower;
  private final NewConnectionAllower myOutgoingConnectionAllower;
  private final TimeoutStorage socketTimeoutStorage = new TimeoutStorageImpl();
  private final AtomicBoolean alreadyInit = new AtomicBoolean(false);
  private final AtomicInteger mySendBufferSize;
  private final AtomicInteger myReceiveBufferSize;

  public ConnectionManager(ConnectionManagerContext context,
                           TimeService timeService,
                           NewConnectionAllower newIncomingConnectionAllower,
                           NewConnectionAllower newOutgoingConnectionAllower,
                           SelectorFactory selectorFactory,
                           AtomicInteger mySendBufferSize,
                           AtomicInteger myReceiveBufferSize) throws IOException {
    this.mySendBufferSize = mySendBufferSize;
    this.myReceiveBufferSize = myReceiveBufferSize;
    this.selector = selectorFactory.newSelector();
    this.myTimeService = timeService;
    myContext = context;
    this.myIncomingConnectionAllower = newIncomingConnectionAllower;
    this.myOutgoingConnectionAllower = newOutgoingConnectionAllower;
  }

  public void initAndRunWorker(ServerChannelRegister serverChannelRegister) throws IOException {

    boolean wasInit = alreadyInit.getAndSet(true);

    if (wasInit) {
      throw new IllegalStateException("connection manager was already initialized");
    }

    myServerSocketChannel = serverChannelRegister.channelFor(selector);
    myServerSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new AcceptAttachmentImpl(myContext));
    myBindPort = myServerSocketChannel.socket().getLocalPort();
    String serverName = myServerSocketChannel.socket().toString();
    myConnectionWorker = new ConnectionWorker(selector, Arrays.asList(
            new InvalidKeyProcessor(),
            new AcceptableKeyProcessor(selector, serverName, myTimeService, myIncomingConnectionAllower, socketTimeoutStorage,
                    mySendBufferSize, myReceiveBufferSize),
            new ConnectableKeyProcessor(selector, myTimeService, socketTimeoutStorage,
                    mySendBufferSize, myReceiveBufferSize),
            new ReadableKeyProcessor(serverName),
            new WritableKeyProcessor()), DEFAULT_SELECTOR_SELECT_TIMEOUT_MILLIS, DEFAULT_CLEANUP_RUN_TIMEOUT_MILLIS,
            myTimeService,
            new CleanupKeyProcessor(myTimeService),
            myOutgoingConnectionAllower);
    myWorkerFuture = myContext.getExecutor().submit(myConnectionWorker);
  }

  public void setSelectorSelectTimeout(int timeout) {
    ConnectionWorker workerLocal = myConnectionWorker;
    checkThatWorkerIsInit(workerLocal);
    workerLocal.setSelectorSelectTimeout(timeout);
  }

  private void checkThatWorkerIsInit(ConnectionWorker worker) {
    if (worker == null) throw new IllegalStateException("Connection manager is not initialized!");
  }

  public boolean offerConnect(ConnectTask connectTask, int timeout, TimeUnit timeUnit) {
    if (myConnectionWorker == null) {
      return false;
    }
    return myConnectionWorker.offerConnect(connectTask, timeout, timeUnit);
  }

  public boolean offerWrite(WriteTask writeTask, int timeout, TimeUnit timeUnit) {
    if (myConnectionWorker == null) {
      return false;
    }
    return myConnectionWorker.offerWrite(writeTask, timeout, timeUnit);
  }


  public int getBindPort() {
    return myBindPort;
  }

  public void close(int timeout, TimeUnit timeUnit) {
    logger.debug("try close connection manager...");
    boolean successfullyClosed = true;
    if (myConnectionWorker != null) {
      myWorkerFuture.cancel(true);
      try {
        boolean shutdownCorrectly = myConnectionWorker.stop(timeout, timeUnit);
        if (!shutdownCorrectly) {
          successfullyClosed = false;
          logger.warn("unable to terminate worker in {} {}", timeout, timeUnit);
        }
      } catch (InterruptedException e) {
        successfullyClosed = false;
        LoggerUtils.warnAndDebugDetails(logger, "unable to await termination worker, thread was interrupted", e);
      }
    }
    try {
      this.myServerSocketChannel.close();
    } catch (Throwable e) {
      LoggerUtils.errorAndDebugDetails(logger, "unable to close server socket channel", e);
      successfullyClosed = false;
    }
    for (SelectionKey key : this.selector.keys()) {
      try {
        if (key.isValid()) {
          key.channel().close();
        }
      } catch (Throwable e) {
        logger.error("unable to close socket channel {}", key.channel());
        successfullyClosed = false;
        logger.debug("", e);
      }
    }
    try {
      this.selector.close();
    } catch (Throwable e) {
      LoggerUtils.errorAndDebugDetails(logger, "unable to close selector channel", e);
      successfullyClosed = false;
    }
    if (successfullyClosed) {
      logger.debug("connection manager is successfully closed");
    } else {
      logger.error("connection manager wasn't closed successfully");
    }
  }

  public void close() {
    close(1, TimeUnit.MINUTES);
  }

  public void setCleanupTimeout(long timeoutMillis) {
    ConnectionWorker workerLocal = myConnectionWorker;
    checkThatWorkerIsInit(workerLocal);
    workerLocal.setCleanupTimeout(timeoutMillis);
  }

  public void setSocketConnectionTimeout(long timeoutMillis) {
    socketTimeoutStorage.setTimeout(timeoutMillis);
  }

  public void closeChannel(Channel channel) throws IOException {
    channel.close();
  }
} public interface WriteListener {

  /**
   * invoked if write is failed by any reason
   *
   * @param message error description
   * @param e       exception if exist. Otherwise null
   */
  void onWriteFailed(String message, Throwable e);

  /**
   * invoked if write done correctly
   */
  void onWriteDone();

}
import com.turn.ttorrent.client.PeerInformation;
import com.turn.ttorrent.client.Piece;
import com.turn.ttorrent.client.SharedTorrent;
import com.turn.ttorrent.common.LoggerUtils;
import com.turn.ttorrent.common.Peer;
import com.turn.ttorrent.common.TorrentLoggerFactory;
import com.turn.ttorrent.common.TorrentUtils;
import com.turn.ttorrent.common.protocol.PeerMessage;
import com.turn.ttorrent.network.ConnectionClosedException;
import com.turn.ttorrent.network.ConnectionManager;
import com.turn.ttorrent.network.WriteListener;
import com.turn.ttorrent.network.WriteTask;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
      this.downloading = myRequestedPieces.size() > 0;
    }
  }

  /**
   * Cancel all pending requests.
   * <p/>
   * <p>
   * This queues CANCEL messages for all the requests in the queue, and
   * returns the list of requests that were in the queue.
   * </p>
   * <p/>
   * <p>
   * If no request queue existed, or if it was empty, an empty set of request
   * messages is returned.
   * </p>
   */
  public void cancelPendingRequests() {
    cancelPendingRequests(null);
  }

  public void cancelPendingRequests(@Nullable final Piece piece) {
    synchronized (this.requestsLock) {
      if (piece != null) {
        myRequestedPieces.remove(piece);
      } else {
        myRequestedPieces.clear();
      }
      this.downloading = myRequestedPieces.size() > 0;
    }
  }

  public int getRemainingRequestedPieces(final Piece piece) {
    synchronized (this.requestsLock) {
      Integer requestedBlocksCount = myRequestedPieces.get(piece);
      if (requestedBlocksCount == null) return 0;
      return requestedBlocksCount;
    }
  }

  /**
   * Handle an incoming message from this peer.
   *
   * @param msg The incoming, parsed message.
   */
  @Override
  public void handleMessage(PeerMessage msg) {
//    logger.trace("Received msg {} from {}", msg.getType(), this);
    if (isStopped.get())
      return;
    if (!torrent.isInitialized()) {
      torrent.initIfNecessary(this);
    }
    switch (msg.getType()) {
      case KEEP_ALIVE:
        // Nothing to do, we're keeping the connection open anyways.
        break;
      case CHOKE:
        this.choked = true;
        this.firePeerChoked();
        this.cancelPendingRequests();
        break;
      case UNCHOKE:
        this.choked = false;
        logger.trace("Peer {} is now accepting requests.", this);
        this.firePeerReady();
        break;
      case INTERESTED:
        this.interested = true;
        if (this.choking) {
          unchoke();
        }
        break;
      case NOT_INTERESTED:
        this.interested = false;
        if (!interesting) {
          unbind(true);
        }
        break;
      case HAVE:
        // Record this peer has the given piece
        PeerMessage.HaveMessage have = (PeerMessage.HaveMessage) msg;
        Piece havePiece = this.torrent.getPiece(have.getPieceIndex());

        synchronized (this.availablePiecesLock) {
          this.availablePieces.set(havePiece.getIndex());
          logger.trace("Peer {} now has {} [{}/{}].",
                  new Object[]{
                          this,
                          havePiece,
                          this.availablePieces.cardinality(),
                          this.torrent.getPieceCount()
                  });
        }

        this.firePieceAvailabity(havePiece);
        break;
      case BITFIELD:
        // Augment the hasPiece bit field from this BITFIELD message
        PeerMessage.BitfieldMessage bitfield =
                (PeerMessage.BitfieldMessage) msg;

        synchronized (this.availablePiecesLock) {
          this.availablePieces.or(bitfield.getBitfield());
          logger.trace("Recorded bitfield from {} with {} " +
                          "pieces(s) [{}/{}].",
                  new Object[]{
                          this,
                          bitfield.getBitfield().cardinality(),
                          this.availablePieces.cardinality(),
                          this.torrent.getPieceCount()
                  });
        }

        this.fireBitfieldAvailabity();
        break;
      case REQUEST:
        PeerMessage.RequestMessage request =
                (PeerMessage.RequestMessage) msg;
        logger.trace("Got request message for {} ({} {}@{}) from {}", new Object[]{
                Arrays.toString(TorrentUtils.getTorrentFileNames(torrent).toArray()),