Implement Remaining Peer Protocol Features

src/main/java/jtorrent/application/domain/Client.java

@@ -1,5 +1,6 @@
 package jtorrent.application.domain;
 
+import java.io.IOException;
 import java.lang.System.Logger;
 import java.lang.System.Logger.Level;
 import java.net.InetAddress;
@@ -8,11 +9,12 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import jtorrent.common.domain.util.BackgroundTask;
 import jtorrent.common.domain.util.Sha1Hash;
 import jtorrent.common.domain.util.rx.RxObservableList;
 import jtorrent.dht.domain.handler.DhtClient;
 import jtorrent.dht.domain.handler.DhtManager;
-import jtorrent.incoming.domain.IncomingConnectionManager;
+import jtorrent.incoming.domain.InboundConnectionListener;
 import jtorrent.lsd.domain.handler.LocalServiceDiscoveryManager;
 import jtorrent.lsd.domain.model.Announce;
 import jtorrent.peer.domain.communication.PeerSocket;
@@ -22,27 +24,28 @@
 import jtorrent.torrent.domain.repository.PieceRepository;
 import jtorrent.torrent.domain.repository.TorrentRepository;
 
-public class Client implements IncomingConnectionManager.Listener, LocalServiceDiscoveryManager.Listener,
-        TorrentHandler.Listener, DhtManager.PeerDiscoveryListener {
+public class Client implements LocalServiceDiscoveryManager.Listener, TorrentHandler.Listener,
+        DhtManager.PeerDiscoveryListener {
 
     private static final Logger LOGGER = System.getLogger(Client.class.getName());
 
-    private final IncomingConnectionManager incomingConnectionManager;
+    private final InboundConnectionListener inboundConnectionListener;
     private final LocalServiceDiscoveryManager localServiceDiscoveryManager;
     private final DhtClient dhtManager;
     private final Map<Sha1Hash, TorrentHandler> infoHashToTorrentHandler = new HashMap<>();
     private final TorrentRepository torrentRepository;
     private final PieceRepository pieceRepository;
+    private final HandleInboundConnectionsTask handleInboundConnectionsTask = new HandleInboundConnectionsTask();
 
     public Client(TorrentRepository torrentRepository, PieceRepository pieceRepository,
-            IncomingConnectionManager incomingConnectionManager,
+            InboundConnectionListener inboundConnectionListener,
             LocalServiceDiscoveryManager localServiceDiscoveryManager, DhtClient dhtClient) {
         this.torrentRepository = torrentRepository;
         this.pieceRepository = pieceRepository;
 
-        this.incomingConnectionManager = incomingConnectionManager;
-        this.incomingConnectionManager.addListener(this);
-        this.incomingConnectionManager.start();
+        this.inboundConnectionListener = inboundConnectionListener;
+        this.inboundConnectionListener.start();
+        handleInboundConnectionsTask.start();
 
         this.localServiceDiscoveryManager = localServiceDiscoveryManager;
         this.localServiceDiscoveryManager.addListener(this);
@@ -71,7 +74,8 @@ public Client(TorrentRepository torrentRepository, PieceRepository pieceReposito
     }
 
     public void shutdown() {
-        incomingConnectionManager.stop();
+        inboundConnectionListener.stop();
+        handleInboundConnectionsTask.stop();
         localServiceDiscoveryManager.stop();
         dhtManager.stop();
         infoHashToTorrentHandler.values().forEach(TorrentHandler::stop);
@@ -99,17 +103,6 @@ public void stopTorrent(Torrent torrent) {
         // TODO: remove from local service discovery
     }
 
-    @Override
-    public void onIncomingPeerConnection(PeerSocket peerSocket, Sha1Hash infoHash) {
-        if (!infoHashToTorrentHandler.containsKey(infoHash)) {
-            LOGGER.log(Level.ERROR, "No torrent found for infohash " + infoHash);
-            return;
-        }
-
-        TorrentHandler torrentHandler = infoHashToTorrentHandler.get(infoHash);
-        torrentHandler.handleIncomingPeerConnection(peerSocket);
-    }
-
     public RxObservableList<Torrent> getTorrents() {
         return torrentRepository.getTorrents();
     }
@@ -142,4 +135,28 @@ public void onPeersDiscovered(Sha1Hash infoHash, Collection<PeerContactInfo> pee
         TorrentHandler torrentHandler = infoHashToTorrentHandler.get(infoHash);
         peers.forEach(torrentHandler::handleDiscoveredPeerContact);
     }
+
+    private class HandleInboundConnectionsTask extends BackgroundTask {
+
+        @Override
+        protected void execute() throws InterruptedException {
+            InboundConnectionListener.InboundConnection inboundConnection =
+                    inboundConnectionListener.waitForIncomingConnection();
+
+            Sha1Hash infoHash = inboundConnection.getInfoHash();
+            if (!infoHashToTorrentHandler.containsKey(infoHash)) {
+                LOGGER.log(Level.ERROR, "No torrent found for infohash " + infoHash);
+                try {
+                    inboundConnection.reject();
+                } catch (IOException e) {
+                    LOGGER.log(Level.ERROR, "Error rejecting incoming connection", e);
+                }
+                return;
+            }
+
+            PeerSocket peerSocket = inboundConnection.accept();
+            TorrentHandler torrentHandler = infoHashToTorrentHandler.get(infoHash);
+            torrentHandler.handleInboundPeerConnection(peerSocket);
+        }
+    }
 }

src/main/java/jtorrent/application/presentation/JTorrent.java

@@ -9,7 +9,7 @@
 import jtorrent.application.presentation.viewmodel.ViewModel;
 import jtorrent.common.domain.Constants;
 import jtorrent.dht.domain.handler.DhtClient;
-import jtorrent.incoming.domain.IncomingConnectionManager;
+import jtorrent.incoming.domain.InboundConnectionListener;
 import jtorrent.lsd.domain.handler.LocalServiceDiscoveryManager;
 import jtorrent.torrent.data.repository.FilePieceRepository;
 import jtorrent.torrent.data.repository.FileTorrentRepository;
@@ -26,13 +26,13 @@ public class JTorrent extends Application {
     @Override
     public void init() throws Exception {
         ServerSocket serverSocket = new ServerSocket(Constants.PORT);
-        IncomingConnectionManager incomingConnectionManager = new IncomingConnectionManager(serverSocket);
+        InboundConnectionListener inboundConnectionListener = new InboundConnectionListener(serverSocket);
 
         DhtClient dhtClient = new DhtClient(Constants.PORT);
 
         TorrentRepository repository = new FileTorrentRepository();
         PieceRepository pieceRepository = new FilePieceRepository();
-        client = new Client(repository, pieceRepository, incomingConnectionManager, new LocalServiceDiscoveryManager(),
+        client = new Client(repository, pieceRepository, inboundConnectionListener, new LocalServiceDiscoveryManager(),
                 dhtClient);
     }
 

src/main/java/jtorrent/common/domain/model/Block.java

@@ -4,31 +4,25 @@
 
 public class Block {
 
-    private final int index;
-    private final int offset;
-    private final int length;
-
-    public Block(int index, int offset, int length) {
-        this.index = index;
-        this.offset = offset;
-        this.length = length;
-    }
+    private final int pieceIndex;
+    private final int blockIndex;
 
-    public int getIndex() {
-        return index;
+    public Block(int pieceIndex, int blockIndex) {
+        this.pieceIndex = pieceIndex;
+        this.blockIndex = blockIndex;
     }
 
-    public int getOffset() {
-        return offset;
+    public int getPieceIndex() {
+        return pieceIndex;
     }
 
-    public int getLength() {
-        return length;
+    public int getBlockIndex() {
+        return blockIndex;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(index, offset, length);
+        return Objects.hash(pieceIndex, blockIndex);
     }
 
     @Override
@@ -40,23 +34,12 @@ public boolean equals(Object o) {
             return false;
         }
         Block block = (Block) o;
-        return index == block.index
-                && offset == block.offset
-                && length == block.length;
+        return pieceIndex == block.pieceIndex
+                && blockIndex == block.blockIndex;
     }
 
     @Override
     public String toString() {
-        return "Block{"
-                + "index=" + index
-                + ", offset=" + offset
-                + ", length=" + length
-                + '}';
-    }
-
-    public enum Status {
-        MISSING,
-        REQUESTED,
-        RECEIVED
+        return String.format("[Piece %d, Block %d]", pieceIndex, blockIndex);
     }
 }

src/main/java/jtorrent/common/domain/util/BackgroundTask.java

@@ -2,7 +2,6 @@
 
 import java.lang.System.Logger;
 import java.lang.System.Logger.Level;
-import java.util.Optional;
 
 public abstract class BackgroundTask implements Runnable {
 
@@ -13,11 +12,11 @@ public abstract class BackgroundTask implements Runnable {
 
     protected BackgroundTask() {
         thread = new Thread(this);
-        getThreadName().ifPresent(thread::setName);
+        thread.setName(getThreadName());
     }
 
-    protected Optional<String> getThreadName() {
-        return Optional.empty();
+    protected String getThreadName() {
+        return getClass().getName();
     }
 
     @Override

src/main/java/jtorrent/common/domain/util/PeriodicTask.java

@@ -0,0 +1,53 @@
+package jtorrent.common.domain.util;
+
+import static jtorrent.common.domain.util.ValidationUtil.requireNonNull;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+public abstract class PeriodicTask implements Runnable {
+
+    private final ScheduledExecutorService scheduledExecutorService;
+    private ScheduledFuture<?> scheduledFuture;
+
+    protected PeriodicTask(ScheduledExecutorService scheduledExecutorService) {
+        this.scheduledExecutorService = requireNonNull(scheduledExecutorService);
+    }
+
+    public void scheduleWithFixedDelay(long delay, TimeUnit timeUnit) {
+        scheduleWithFixedDelay(0, delay, timeUnit);
+    }
+
+    public void scheduleWithFixedDelay(long initialDelay, long delay, TimeUnit timeUnit) {
+        checkNotRunning();
+        scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(this, initialDelay, delay, timeUnit);
+    }
+
+    public void scheduleAtFixedRate(long period, TimeUnit timeUnit) {
+        scheduleAtFixedRate(0, period, timeUnit);
+    }
+
+    public void scheduleAtFixedRate(long initialDelay, long period, TimeUnit timeUnit) {
+        checkNotRunning();
+        scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(this, initialDelay, period, timeUnit);
+    }
+
+    private void checkNotRunning() {
+        if (isRunning()) {
+            throw new IllegalStateException("Task already running");
+        }
+    }
+
+    public void stop() {
+        if (!isRunning()) {
+            return;
+        }
+
+        scheduledFuture.cancel(true);
+    }
+
+    private boolean isRunning() {
+        return scheduledFuture != null && !scheduledFuture.isCancelled();
+    }
+}