SocketNetwork implements CloseableNetwork { @Override public void close() { if (alive) { alive = false; if (conf.noOfParties() < 2) { logger.info("P{}: Network closed", conf.getMyId()); return; } ExceptionConverter.safe(() -> { closeCommunication(); logger.info("P{}: Network closed", conf.getMyId()); return null; }, "Unable to properly close the network."); } else { logger.info("P{}: Network already closed", conf.getMyId()); } } SocketNetwork(NetworkConfiguration conf, Map<Integer, Socket> socketMap);  SocketNetwork(NetworkConfiguration conf); @Override void send(int partyId, byte[] data); @Override byte[] receive(final int partyId); @Override void close(); @Override int getNoOfParties();  }
@Test(expected = IllegalArgumentException.class) @SuppressWarnings("resource") public void testMissingParty() throws InterruptedException, ExecutionException { final int numParties = 3; final int dropId = 2; List<NetworkConfiguration> confs = getNetConfs(numParties); ExecutorService es = Executors.newFixedThreadPool(numParties); List<Future<NetworkConnector>> fs = new ArrayList<>(numParties); try { for (int i = 0; i < numParties; i++) { final int id = i; fs.add(es.submit(() -> new Connector(confs.get(id), DEFAULT_CONNECTION_TIMEOUT))); } Map<Integer, Socket> missingPartyMap = fs.get(0).get().getSocketMap(); missingPartyMap.remove(dropId); new SocketNetwork(confs.get(0), missingPartyMap); } finally { for (Future<NetworkConnector> futureConn : fs) { for (Socket s : futureConn.get().getSocketMap().values()) { try { s.close(); } catch (IOException e) { } } } es.shutdownNow(); } }
@Test(expected = IllegalArgumentException.class) @SuppressWarnings("resource") public void testIdOutOfRange() throws InterruptedException, ExecutionException, IOException { final int numParties = 3; List<NetworkConfiguration> confs = getNetConfs(numParties); ExecutorService es = Executors.newFixedThreadPool(numParties); List<Future<NetworkConnector>> fs = new ArrayList<>(numParties); try { for (int i = 0; i < numParties; i++) { final int id = i; fs.add(es.submit(() -> new Connector(confs.get(id), DEFAULT_CONNECTION_TIMEOUT))); } Map<Integer, Socket> extraPartyMap = fs.get(0).get().getSocketMap(); extraPartyMap.put(numParties + 1, extraPartyMap.get(numParties)); new SocketNetwork(confs.get(0), extraPartyMap); } finally { for (Future<NetworkConnector> futureConn : fs) { for (Socket s : futureConn.get().getSocketMap().values()) { s.close(); } } es.shutdownNow(); } }
@Test(expected = IllegalArgumentException.class) @SuppressWarnings("resource") public void testClosedSockets() throws InterruptedException, ExecutionException, IOException { final int numParties = 3; List<NetworkConfiguration> confs = getNetConfs(numParties); ExecutorService es = Executors.newFixedThreadPool(numParties); List<Future<NetworkConnector>> fs = new ArrayList<>(numParties); try { for (int i = 0; i < numParties; i++) { final int id = i; fs.add(es.submit(() -> new Connector(confs.get(id), DEFAULT_CONNECTION_TIMEOUT))); } Map<Integer, Socket> socketMap = fs.get(0).get().getSocketMap(); socketMap.get(numParties).close(); new SocketNetwork(confs.get(0), socketMap); } finally { for (Future<NetworkConnector> futureConn : fs) { for (Socket s : futureConn.get().getSocketMap().values()) { s.close(); } } es.shutdownNow(); } }
@Test public void testClosedSocketReciever() throws InterruptedException, ExecutionException, IOException { final int numParties = 2; List<NetworkConfiguration> confs = getNetConfs(numParties); ExecutorService es = Executors.newFixedThreadPool(numParties); List<Future<NetworkConnector>> fs = new ArrayList<>(numParties); try { for (int i = 0; i < numParties; i++) { final int id = i; fs.add(es.submit(() -> new Connector(confs.get(id), DEFAULT_CONNECTION_TIMEOUT))); } Map<Integer, Socket> socketMap1 = fs.get(0).get().getSocketMap(); Map<Integer, Socket> socketMap2 = fs.get(1).get().getSocketMap(); Receiver r = new Receiver(socketMap1.get(2)); socketMap2.get(1).close(); } finally { for (Future<NetworkConnector> futureConn : fs) { for (Socket s : futureConn.get().getSocketMap().values()) { } } es.shutdownNow(); } }
@Test public void testSenderFailsWhileStopping() throws InterruptedException, ExecutionException, IOException { final int numParties = 2; List<NetworkConfiguration> confs = getNetConfs(numParties); ExecutorService es = Executors.newFixedThreadPool(numParties); List<Future<NetworkConnector>> fs = new ArrayList<>(numParties); try { for (int i = 0; i < numParties; i++) { final int id = i; fs.add(es.submit(() -> new Connector(confs.get(id), DEFAULT_CONNECTION_TIMEOUT))); } Map<Integer, Socket> socketMap = fs.get(0).get().getSocketMap(); Sender s = new Sender(socketMap.get(2)); socketMap.get(2).close(); s.stop(); } finally { for (Future<NetworkConnector> futureConn : fs) { for (Socket s : futureConn.get().getSocketMap().values()) { s.close(); } } es.shutdownNow(); } }
@Test public void testSenderBreakingSocket() throws Exception { final int numParties = 2; List<NetworkConfiguration> confs = getNetConfs(numParties); ExecutorService es = Executors.newFixedThreadPool(numParties); List<Future<NetworkConnector>> fs = new ArrayList<>(numParties); try { for (int i = 0; i < numParties; i++) { final int id = i; fs.add(es.submit(() -> new Connector(confs.get(id), DEFAULT_CONNECTION_TIMEOUT))); } Map<Integer, Socket> socketMap = fs.get(0).get().getSocketMap(); Socket socket = socketMap.get(2); Sender sender = new Sender(socket); socket.close(); sender.queueMessage(new byte[1]); } finally { for (Future<NetworkConnector> futureConn : fs) { for (Socket s : futureConn.get().getSocketMap().values()) { s.close(); } } es.shutdownNow(); } }