many fixes

build.gradle

@@ -5,7 +5,7 @@ plugins {
 }
 
 group 'me.bvn13'
-version '1.2.4'
+version '1.2.5'
 
 repositories {
     mavenCentral()
@@ -34,6 +34,10 @@ tasks.withType(Test).configureEach {
     maxParallelForks = 1
 }
 
+task sourceJar(type: Jar) {
+    from sourceSets.main.allJava
+}
+
 publishing {
     repositories {
         maven {
@@ -49,5 +53,12 @@ publishing {
         gpr(MavenPublication) {
             from(components.java)
         }
+        mavenJava(MavenPublication) {
+            from components.java
+
+            artifact sourceJar {
+                classifier "sources"
+            }
+        }
     }
 }

src/main/java/me/bvn13/sewy/AbstractClientListener.java

@@ -47,7 +47,7 @@ public abstract class AbstractClientListener implements Runnable {
             this.in = socket.getInputStream();
             log.debug("BufferedReader successfully created");
             log.debug("PrintWriter successfully created");
-            out = socket.getOutputStream();
+            this.out = socket.getOutputStream();
             log.debug("OutputStream successfully created");
         } catch (IOException e) {
             throw new RuntimeException(e);
@@ -66,7 +66,7 @@ public abstract class AbstractClientListener implements Runnable {
      *
      * @return the line read from socket
      */
-    public String readLine() {
+    public String readLine() throws IOException {
         final byte[] bytes = readBytes(getSeparator());
         final StringBuilder sb = new StringBuilder();
         for (byte aByte : bytes) {
@@ -83,12 +83,11 @@ public abstract class AbstractClientListener implements Runnable {
      * @param separator byte to separate data portions
      * @return array of bytes read from socket
      */
-    public byte[] readBytes(byte[] separator) {
+    public byte[] readBytes(byte[] separator) throws IOException {
         final List<Byte> data = new ArrayList<>(2048 * 2048);
         List<Byte> buffer = new ArrayList<>(separator.length);
         int separatorPosition = 0;
-        try {
+        while (socket.isConnected() && !socket.isClosed()) {
-            while (socket.isConnected()) {
             byte[] portion = in.readNBytes(1);
             if (portion == null || portion.length == 0) {
                 break;
@@ -114,9 +113,6 @@ public abstract class AbstractClientListener implements Runnable {
         }
         if (log.isTraceEnabled()) log.trace("Received {} bytes: {}", bytes.length, bytes);
         return bytes;
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
     }
 
     /**
@@ -126,15 +122,11 @@ public abstract class AbstractClientListener implements Runnable {
      * @param bytes     bytes to be sent into socket
      * @param separator byte to separate data portions
      */
-    public void writeBytes(byte[] bytes, byte[] separator) {
+    public void writeBytes(byte[] bytes, byte[] separator) throws IOException {
         if (log.isTraceEnabled()) log.trace("Sending {} bytes: {}", bytes.length, bytes);
-        try {
         out.write(bytes);
         out.write(separator);
         out.flush();
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
     }
 
     /**
@@ -145,7 +137,11 @@ public abstract class AbstractClientListener implements Runnable {
      */
     public void writeLine(String data) {
         if (log.isTraceEnabled()) log.trace("Sending: " + data);
+        try {
             writeBytes(data.getBytes(), getSeparator());
+        } catch (Exception e) {
+            log.error("", e);
+        }
     }
 
     /**
@@ -157,12 +153,12 @@ public abstract class AbstractClientListener implements Runnable {
             out.close();
             in.close();
         } catch (IOException e) {
-            log.warn("Unable to close IN client buffer");
+            log.warn("Unable to close IN/OUT client buffer");
         }
         try {
             socket.close();
         } catch (IOException e) {
-            throw new RuntimeException(e);
+            log.warn("Unable to close socket");
         }
     }
 }

src/main/java/me/bvn13/sewy/Client.java

@@ -115,7 +115,11 @@ public class Client<T extends AbstractClientListener> {
      * @return the line read from socket
      */
     public String readLine() {
+        try {
             return client.readLine();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
     }
 
     /**
@@ -123,7 +127,7 @@ public class Client<T extends AbstractClientListener> {
      * @param separator
      * @return
      */
-    public byte[] readBytes(byte[] separator) {
+    public byte[] readBytes(byte[] separator) throws IOException {
         return client.readBytes(separator);
     }
 

src/main/java/me/bvn13/sewy/ClientListenerFactory.java

@@ -26,6 +26,7 @@ class ClientListenerFactory {
 
     /**
      * Creates client listener constructor
+     *
      * @param clientListenerClass class to be used as client listener
      * @param <T>                 generic type
      * @return lambda method to create client listener
@@ -33,9 +34,8 @@ class ClientListenerFactory {
     @SuppressWarnings("unchecked")
     static <T extends AbstractClientListener> Function<Socket, T> createClientListenerConstructor(Class clientListenerClass) {
 
-        if (clientListenerClass.getGenericSuperclass() == null
+        if (clientListenerClass.getGenericSuperclass() == null) {
-                /*|| !clientListenerClass.getGenericSuperclass().equals(T.class)*/) {
+            throw new IllegalArgumentException("Wrong client listener of type: " + clientListenerClass.getName());
-            throw new IllegalArgumentException("Wrong client listener of type: "+clientListenerClass.getName());
         }
 
         return (client) -> {

src/main/java/me/bvn13/sewy/CommandClient.java

@@ -19,10 +19,10 @@ import me.bvn13.sewy.command.AbstractCommand;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.net.Socket;
 import java.util.function.Function;
 
-import static java.lang.String.format;
 import static me.bvn13.sewy.ClientListenerFactory.createClientListenerConstructor;
 
 /**
@@ -41,6 +41,7 @@ public class CommandClient extends Client<CommandClientListener> {
 
     /**
      * Starts to connect to server immediately
+     *
      * @param host host to connect to
      * @param port port to be used while connecting
      */
@@ -50,6 +51,7 @@ public class CommandClient extends Client<CommandClientListener> {
 
     /**
      * Starts to connect to server immediately
+     *
      * @param host                host to connect to
      * @param port                port to be used while connecting
      * @param clientListenerClass client listener class describing protocol of communications
@@ -60,6 +62,7 @@ public class CommandClient extends Client<CommandClientListener> {
 
     /**
      * Connects to server immediately
+     *
      * @param host                      host to connect to
      * @param port                      port to be used while connecting
      * @param clientListenerConstructor to provide constructor for client listener (see {@link me.bvn13.sewy.Client#Client(java.lang.String, int, java.lang.Class)})
@@ -71,10 +74,12 @@ public class CommandClient extends Client<CommandClientListener> {
 
     /**
      * Sends command to server
+     *
      * @param command command to be sent
      * @param <T>     generic type
+     * @throws IOException if any error occurred while sending
      */
-    public <T extends AbstractCommand> void send(T command) {
+    public <T extends AbstractCommand> void send(T command) throws IOException {
         log.debug("Start to send command: " + command);
         client.send(command);
     }

src/main/java/me/bvn13/sewy/CommandClientListener.java

@@ -18,6 +18,7 @@ package me.bvn13.sewy;
 import me.bvn13.sewy.command.AbstractCommand;
 import org.apache.commons.lang3.SerializationUtils;
 
+import java.io.IOException;
 import java.io.Serializable;
 import java.net.Socket;
 
@@ -39,7 +40,8 @@ public class CommandClientListener extends AbstractClientListener implements Abs
     public void run() {
         for (Thread.yield(); !socket.isConnected() && !socket.isClosed(); Thread.yield()) {
         }
-        while (socket.isConnected()) {
+        while (socket.isConnected() && !socket.isClosed()) {
+            try {
                 Thread.yield();
                 byte[] line = readBytes(getSeparator());
                 if (line == null || line.length == 0) {
@@ -67,6 +69,9 @@ public class CommandClientListener extends AbstractClientListener implements Abs
                 final Serializable response = onCommand((AbstractCommand) command);
                 log.debug(format("Response for %s is: %s", command, response));
                 writeBytes(SerializationUtils.serialize(response), getSeparator());
+            } catch (Exception e) {
+                log.error("Failed to communicate!", e);
+            }
         }
     }
 
@@ -85,11 +90,12 @@ public class CommandClientListener extends AbstractClientListener implements Abs
 
     /**
      * Sends command to opposite side
+     *
      * @param command command to be sent
      * @param <T>     generic type
      */
-    public <T extends AbstractCommand> void send(T command) {
+    public <T extends AbstractCommand> void send(T command) throws IOException {
-        log.debug("Start to send command: " + command);
+        log.debug("Start to send command: {}", command);
         writeBytes(SerializationUtils.serialize(command), getSeparator());
     }
 

src/main/java/me/bvn13/sewy/CommandServer.java

@@ -16,19 +16,12 @@
 package me.bvn13.sewy;
 
 import me.bvn13.sewy.command.AbstractCommand;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
-import java.util.ArrayList;
+import java.util.function.Consumer;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.function.Function;
 
 import static java.lang.String.format;
@@ -59,7 +52,6 @@ public class CommandServer extends Server<CommandClientListener> {
     }
 
     /**
-     *
      * @param host                      host to bind in order to start listen to clients
      * @param port                      port to start listen to
      * @param clientListenerConstructor to provide constructor for client listener (see {@link CommandServer#CommandServer(String, int, Class)})
@@ -73,11 +65,13 @@ public class CommandServer extends Server<CommandClientListener> {
                 socket = server;
 
                 while (!server.isClosed()) {
+                    if (!isMaximumClientsAchieved()) {
                         final Socket client = server.accept();
                         final CommandClientListener clientListener = clientListenerConstructor.apply(client);
                         executor.execute(clientListener);
                         clients.add(clientListener);
                     }
+                }
 
             } catch (IOException e) {
                 log.error(format("Error while conversation with %s:%d", host, port), e);
@@ -87,14 +81,31 @@ public class CommandServer extends Server<CommandClientListener> {
 
     /**
      * Sends command to every client
+     *
      * @param command     command to be sent
      * @param <T>         generic type
      */
     public <T extends AbstractCommand> void send(T command) {
-        log.debug("Start to send command: " + command);
+        send(command, client -> {});
-        for (CommandClientListener client : clients) {
-            client.send(command);
-        }
     }
 
+    /**
+     * Sends command to every client
+     *
+     * @param command     command to be sent
+     * @param <T>         generic type
+     * @param onException for catching errors while sending. Do not throw any Exception inside onException callback -
+     *                    it leads to stopping sending the command to remaining clients
+     */
+    public <T extends AbstractCommand> void send(T command, Consumer<CommandClientListener> onException) {
+        log.debug("Start to send command: " + command);
+        for (CommandClientListener client : clients) {
+            try {
+                client.send(command);
+            } catch (IOException e) {
+                log.error("Failed to send command " + command, e);
+                onException.accept(client);
+            }
+        }
+    }
 }

src/main/java/me/bvn13/sewy/Server.java

@@ -46,6 +46,8 @@ public class Server<T extends AbstractClientListener> {
 
     protected ServerSocket socket;
 
+    private int maxClientsCount;
+
     protected Server() {
     }
 
@@ -59,7 +61,6 @@ public class Server<T extends AbstractClientListener> {
     }
 
     /**
-     *
      * @param host                      host to bind in order to start listen to clients
      * @param port                      port to start listen to
      * @param clientListenerConstructor to provide constructor for client listener (see {@link me.bvn13.sewy.Server#Server(java.lang.String, int, java.lang.Class)})
@@ -73,17 +74,19 @@ public class Server<T extends AbstractClientListener> {
                 socket = server;
 
                 while (!server.isClosed()) {
+                    if (!isMaximumClientsAchieved()) {
                         final Socket client = server.accept();
                         final T clientListener = clientListenerConstructor.apply(client);
                         executor.execute(clientListener);
                         clients.add(clientListener);
                     }
+                    Thread.yield();
+                }
 
             } catch (IOException e) {
                 log.error(format("Error while conversation with %s:%d", host, port), e);
             }
         });
-
     }
 
     /**
@@ -110,10 +113,33 @@ public class Server<T extends AbstractClientListener> {
 
     /**
      * To check whether the server is ready for new connections
+     *
      * @return
      */
     public boolean isListening() {
         return socket != null && socket.isBound();
     }
 
+    /**
+     * Returns count of connected clients
+     *
+     * @return count of connected clients
+     */
+    public int getClientsCount() {
+        return clients.size();
+    }
+
+    /**
+     * Sets maximum clients to be connected to server
+     *
+     * @param count maximum clients count
+     */
+    public void setMaxClientsCount(int count) {
+        maxClientsCount = count;
+    }
+
+    protected boolean isMaximumClientsAchieved() {
+        return maxClientsCount == 0
+                || clients.size() >= maxClientsCount;
+    }
 }

src/test/java/me/bvn13/sewy/EchoClientListener.java

@@ -15,6 +15,9 @@
  */
 package me.bvn13.sewy;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.net.Socket;
 
 /**
@@ -22,6 +25,9 @@ import java.net.Socket;
  * Writes into socket all the data received before
  */
 public class EchoClientListener extends AbstractClientListener {
+
+    private final Logger log = LoggerFactory.getLogger(EchoClientListener.class);
+
     public EchoClientListener(Socket socket) {
         super(socket);
     }
@@ -33,8 +39,12 @@ public class EchoClientListener extends AbstractClientListener {
     public void run() {
         while (socket.isConnected()) {
             Thread.yield();
+            try {
                 final String data = readLine();
                 writeLine(data);
+            } catch (Exception e) {
+                log.error("", e);
+            }
         }
     }
 }

src/test/java/me/bvn13/sewy/ServerTest.java

@@ -83,7 +83,7 @@ public class ServerTest {
 
     @ParameterizedTest
     @ValueSource(ints = START_PORT + 6)
-    void serverIsAbleToPingPong(int port) throws InterruptedException {
+    void serverIsAbleToPingPong(int port) throws Exception {
         Sewy.register(PingCommand.class);
         Sewy.register(PongCommand.class);
 
@@ -116,7 +116,7 @@ public class ServerTest {
 
     @ParameterizedTest
     @ValueSource(ints = START_PORT + 7)
-    void wideSeparatorTest(int port) throws InterruptedException {
+    void wideSeparatorTest(int port) throws Exception {
         Sewy.register(ComplexCommand.class);
         Sewy.setSeparator(new byte[] { '\n', 'e', 'n', 'd', '\n' });