<commit_before>package com.turn.ttorrent.client.network;

import com.turn.ttorrent.client.Handshake;
import com.turn.ttorrent.client.SharedTorrent;
import com.turn.ttorrent.client.peer.SharingPeer;
import com.turn.ttorrent.common.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
import java.text.ParseException;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;

public class HandshakeReceiver implements DataProcessor {

  private static final Logger logger = LoggerFactory.getLogger(HandshakeReceiver.class);

  private final PeersStorageProvider peersStorageProvider;
  private final TorrentsStorageProvider torrentsStorageProvider;
  private final ExecutorService executorService;
  private final SharingPeerFactory sharingPeerFactory;
  private final String myHostAddress;
  private final int myPort;
  private final boolean myIsOutgoingConnection;
  private ByteBuffer messageBytes;
  private int pstrLength;

  public HandshakeReceiver(PeersStorageProvider peersStorageProvider,
                           TorrentsStorageProvider torrentsStorageProvider,
                           ExecutorService executorService,
                           SharingPeerFactory sharingPeerFactory,
                           String hostAddress,
                           int port,
                           boolean isOutgoingListener) {
    this.peersStorageProvider = peersStorageProvider;
    this.torrentsStorageProvider = torrentsStorageProvider;
    this.executorService = executorService;
    this.sharingPeerFactory = sharingPeerFactory;
    myHostAddress = hostAddress;
    myPort = port;
    this.pstrLength = -1;
    this.myIsOutgoingConnection = isOutgoingListener;
  }

  @Override
  public DataProcessor processAndGetNext(ByteChannel socketChannel) throws IOException {
    if (pstrLength == -1) {
      ByteBuffer len = ByteBuffer.allocate(1);
      int readBytes = -1;
      try {
        readBytes = socketChannel.read(len);
      } catch (IOException ignored) {
      }
      if (readBytes == -1) {
        return new ShutdownProcessor();
      }
      if (readBytes == 0) {
        return this;
      }
      len.rewind();
      byte pstrLen = len.get();
      this.pstrLength = pstrLen;
      messageBytes = ByteBuffer.allocate(this.pstrLength + Handshake.BASE_HANDSHAKE_LENGTH);
      messageBytes.put(pstrLen);
    }
    int readBytes = -1;
    try {
      readBytes = socketChannel.read(messageBytes);
    } catch (IOException e) {
      LoggerUtils.warnAndDebugDetails(logger, "unable to read data from {}", socketChannel, e);
    }
    if (readBytes == -1) {
      return new ShutdownProcessor();
    }
    if (messageBytes.remaining() != 0) {
      return this;
    }
    Handshake hs = parseHandshake(socketChannel.toString());

    if (hs == null) {
      return new ShutdownProcessor();
    }

    SharedTorrent torrent = torrentsStorageProvider.getTorrentsStorage().getTorrent(hs.getHexInfoHash());

    if (torrent == null) {
      logger.debug("peer {} tries to download unknown torrent {}",
              Arrays.toString(hs.getPeerId()),
              hs.getHexInfoHash());
      return new ShutdownProcessor();
    }

    logger.debug("got handshake {} from {}", Arrays.toString(messageBytes.array()), socketChannel);

    final String peerId = new String(hs.getPeerId(), Torrent.BYTE_ENCODING);

    SharingPeer sharingPeer =
            sharingPeerFactory.createSharingPeer(myHostAddress, myPort, ByteBuffer.wrap(hs.getPeerId()), torrent, socketChannel);
    PeerUID peerUID = new PeerUID(peerId, hs.getHexInfoHash());

    SharingPeer old = peersStorageProvider.getPeersStorage().putIfAbsent(peerUID, sharingPeer, myIsOutgoingConnection);
    if (old != null) {
      logger.debug("Already connected to old peer {}, close current connection with {}", old, sharingPeer);
      return new ShutdownProcessor();
    }

    if (!myIsOutgoingConnection) {
      logger.debug("send handshake to {}", socketChannel);
      try {
        ConnectionUtils.sendHandshake(socketChannel, hs.getInfoHash(), peersStorageProvider.getPeersStorage().getSelf().getPeerIdArray());
      } catch (IOException e) {<commit_msg>added correct closing connection on incorrect handshake or unknown torrent<commit_after>package com.turn.ttorrent.client.network;

import com.turn.ttorrent.client.Handshake;
import com.turn.ttorrent.client.SharedTorrent;
import com.turn.ttorrent.client.peer.SharingPeer;
import com.turn.ttorrent.common.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
import java.text.ParseException;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;

public class HandshakeReceiver implements DataProcessor {

  private static final Logger logger = LoggerFactory.getLogger(HandshakeReceiver.class);

  private final PeersStorageProvider peersStorageProvider;
  private final TorrentsStorageProvider torrentsStorageProvider;
  private final ExecutorService executorService;
  private final SharingPeerFactory sharingPeerFactory;
  private final String myHostAddress;
  private final int myPort;
  private final boolean myIsOutgoingConnection;
  private ByteBuffer messageBytes;
  private int pstrLength;

  public HandshakeReceiver(PeersStorageProvider peersStorageProvider,
                           TorrentsStorageProvider torrentsStorageProvider,
                           ExecutorService executorService,
                           SharingPeerFactory sharingPeerFactory,
                           String hostAddress,
                           int port,
                           boolean isOutgoingListener) {
    this.peersStorageProvider = peersStorageProvider;
    this.torrentsStorageProvider = torrentsStorageProvider;
    this.executorService = executorService;
    this.sharingPeerFactory = sharingPeerFactory;
    myHostAddress = hostAddress;
    myPort = port;
    this.pstrLength = -1;
    this.myIsOutgoingConnection = isOutgoingListener;
  }

  @Override
  public DataProcessor processAndGetNext(ByteChannel socketChannel) throws IOException {
    if (pstrLength == -1) {
      ByteBuffer len = ByteBuffer.allocate(1);
      int readBytes = -1;
      try {
        readBytes = socketChannel.read(len);
      } catch (IOException ignored) {
      }
      if (readBytes == -1) {
        return new ShutdownProcessor().processAndGetNext(socketChannel);
      }
      if (readBytes == 0) {
        return this;
      }
      len.rewind();
      byte pstrLen = len.get();
      this.pstrLength = pstrLen;
      messageBytes = ByteBuffer.allocate(this.pstrLength + Handshake.BASE_HANDSHAKE_LENGTH);
      messageBytes.put(pstrLen);
    }
    int readBytes = -1;
    try {
      readBytes = socketChannel.read(messageBytes);
    } catch (IOException e) {
      LoggerUtils.warnAndDebugDetails(logger, "unable to read data from {}", socketChannel, e);
    }
    if (readBytes == -1) {
      return new ShutdownProcessor().processAndGetNext(socketChannel);
    }
    if (messageBytes.remaining() != 0) {
      return this;
    }
    Handshake hs = parseHandshake(socketChannel.toString());

    if (hs == null) {
      return new ShutdownProcessor().processAndGetNext(socketChannel);
    }

    SharedTorrent torrent = torrentsStorageProvider.getTorrentsStorage().getTorrent(hs.getHexInfoHash());

    if (torrent == null) {
      logger.debug("peer {} tries to download unknown torrent {}",
              Arrays.toString(hs.getPeerId()),
              hs.getHexInfoHash());
      return new ShutdownProcessor().processAndGetNext(socketChannel);
    }

    logger.debug("got handshake {} from {}", Arrays.toString(messageBytes.array()), socketChannel);

    final String peerId = new String(hs.getPeerId(), Torrent.BYTE_ENCODING);

    SharingPeer sharingPeer =
            sharingPeerFactory.createSharingPeer(myHostAddress, myPort, ByteBuffer.wrap(hs.getPeerId()), torrent, socketChannel);
    PeerUID peerUID = new PeerUID(peerId, hs.getHexInfoHash());

    SharingPeer old = peersStorageProvider.getPeersStorage().putIfAbsent(peerUID, sharingPeer, myIsOutgoingConnection);
    if (old != null) {
      logger.debug("Already connected to old peer {}, close current connection with {}", old, sharingPeer);
      return new ShutdownProcessor().processAndGetNext(socketChannel);
    }

    if (!myIsOutgoingConnection) {
      logger.debug("send handshake to {}", socketChannel);
      try {
        ConnectionUtils.sendHandshake(socketChannel, hs.getInfoHash(), peersStorageProvider.getPeersStorage().getSelf().getPeerIdArray());
      } catch (IOException e) {
{'fraction_non_alphanumeric': 0.09233409610983982, 'fraction_numerical': 0.002517162471395881, 'mean_word_length': 3.4258227848101264, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7661237', 'n_tokens_mistral': 2398, 'n_tokens_neox': 2338, 'n_words': 523}