Compare commits

...

8 Commits

Author SHA1 Message Date
bvn13 f8727f1032 test github 2022-02-21 02:11:50 +03:00
bvn13 b6ce8899e4 test github 2022-02-21 02:10:17 +03:00
bvn13 0b745dcb74 many fixes 2022-02-21 02:08:23 +03:00
bvn13 98fc7703f4 many fixes 2022-02-21 02:07:35 +03:00
bvn13 63df33457f added possibility to use multiple bytes as separator 2022-02-19 14:02:27 +03:00
bvn13 1dcf9dec7a fixed to close socket while client and server stopping 2022-02-17 23:23:08 +03:00
bvn13 63689f61e8 settings for token 2022-01-29 11:18:55 +03:00
bvn13 22c0f7effe remove unnecessary dependency 2022-01-29 04:13:34 +03:00
12 changed files with 304 additions and 119 deletions

View File

@ -5,7 +5,7 @@ plugins {
}
group 'me.bvn13'
version '1.2.1'
version '1.2.8'
repositories {
mavenCentral()
@ -16,9 +16,6 @@ dependencies {
// https://mvnrepository.com/artifact/org.slf4j/slf4j-api
implementation group: 'org.slf4j', name: 'slf4j-api', version: '1.7.35'
// https://mvnrepository.com/artifact/commons-io/commons-io
implementation group: 'commons-io', name: 'commons-io', version: '2.11.0'
// https://mvnrepository.com/artifact/org.apache.commons/commons-lang3
implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.12.0'
@ -37,14 +34,18 @@ tasks.withType(Test).configureEach {
maxParallelForks = 1
}
task sourceJar(type: Jar) {
from sourceSets.main.allJava
}
publishing {
repositories {
maven {
name = "GitHubPackages"
url = uri("https://maven.pkg.github.com/bvn13/sewy")
credentials {
username = project.findProperty("gpr.user") ?: System.getenv("USERNAME")
password = project.findProperty("gpr.key") ?: System.getenv("TOKEN")
username = project.findProperty("gpr.user") ?: System.getenv("GITHUB_USER")
password = project.findProperty("gpr.key") ?: System.getenv("GITHUB_TOKEN")
}
}
}
@ -52,5 +53,12 @@ publishing {
gpr(MavenPublication) {
from(components.java)
}
// mavenJava(MavenPublication) {
// from components.java
//
// artifact sourceJar {
// classifier "sources"
// }
// }
}
}

View File

@ -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);
@ -63,9 +63,10 @@ public abstract class AbstractClientListener implements Runnable {
/**
* Reads line (separated with '\n') from socket
*
* @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) {
@ -78,56 +79,69 @@ public abstract class AbstractClientListener implements Runnable {
/**
* Reads data from socket until {@code separator} is encountered
*
* @param separator byte to separate data portions
* @return array of bytes read from socket
*/
public byte[] readBytes(byte separator) {
final List<Byte> data = new ArrayList<>(256);
try {
while (socket.isConnected()) {
byte[] portion = in.readNBytes(1);
if (portion == null || portion.length == 0 || portion[0] == 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;
while (socket.isConnected() && !socket.isClosed()) {
byte[] portion = in.readNBytes(1);
if (portion == null || portion.length == 0) {
break;
}
if (portion[0] == separator[separatorPosition]) {
if (separatorPosition == separator.length - 1) {
break;
}
data.add(portion[0]);
separatorPosition++;
buffer.add(portion[0]);
continue;
} else {
separatorPosition = 0;
data.addAll(buffer);
buffer.clear();
}
final byte[] bytes = new byte[data.size()];
int i = 0;
for (Byte aByte : data) {
bytes[i++] = aByte;
}
if (log.isTraceEnabled()) log.trace("Received: " + new String(bytes));
return bytes;
} catch (IOException e) {
throw new RuntimeException(e);
data.add(portion[0]);
}
final byte[] bytes = new byte[data.size()];
int i = 0;
for (Byte aByte : data) {
bytes[i++] = aByte;
}
if (log.isTraceEnabled()) log.trace("Received {} bytes: {}", bytes.length, bytes);
return bytes;
}
/**
* Writes line into socket ending with default separator '\n'.
* Flushes after writing.
* @param bytes bytes to be sent into socket
*
* @param bytes bytes to be sent into socket
* @param separator byte to separate data portions
*/
public void writeBytes(byte[] bytes, byte separator) {
if (log.isTraceEnabled()) log.trace("Sending: " + new String(bytes));
try {
out.write(bytes);
out.write(separator);
out.flush();
} catch (IOException e) {
throw new RuntimeException(e);
}
public void writeBytes(byte[] bytes, byte[] separator) throws IOException {
if (log.isTraceEnabled()) log.trace("Sending {} bytes: {}", bytes.length, bytes);
out.write(bytes);
out.write(separator);
out.flush();
}
/**
* Writes line into socket ending with default separator '\n'.
* Flushes after writing.
*
* @param data data to be sent into socket
*/
public void writeLine(String data) {
if (log.isTraceEnabled()) log.trace("Sending: " + data);
writeBytes(data.getBytes(), getSeparator());
try {
writeBytes(data.getBytes(), getSeparator());
} catch (Exception e) {
log.error("", e);
}
}
/**
@ -139,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");
}
}
}

View File

@ -89,7 +89,16 @@ public class Client<T extends AbstractClientListener> {
*/
public void stop() {
log.debug("Stopping client");
client.stop();
if (client != null) {
client.stop();
}
try {
if (socket != null) {
socket.close();
}
} catch (IOException e) {
log.error("Failed to close socket");
}
executor.shutdown();
}
@ -106,7 +115,11 @@ public class Client<T extends AbstractClientListener> {
* @return the line read from socket
*/
public String readLine() {
return client.readLine();
try {
return client.readLine();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
@ -114,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);
}

View File

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

View File

@ -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,8 +51,9 @@ 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 host host to connect to
* @param port port to be used while connecting
* @param clientListenerClass client listener class describing protocol of communications
*/
public CommandClient(String host, int port, Class clientListenerClass) {
@ -60,8 +62,9 @@ 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 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)})
*/
public CommandClient(String host, int port, Function<Socket, CommandClientListener> clientListenerConstructor) {
@ -71,10 +74,12 @@ public class CommandClient extends Client<CommandClientListener> {
/**
* Sends command to server
*
* @param command command to be sent
* @param <T> generic type
* @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);
}

View File

@ -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,34 +40,38 @@ public class CommandClientListener extends AbstractClientListener implements Abs
public void run() {
for (Thread.yield(); !socket.isConnected() && !socket.isClosed(); Thread.yield()) {
}
while (socket.isConnected()) {
Thread.yield();
byte[] line = readBytes(getSeparator());
if (line == null || line.length == 0) {
continue;
}
final Object command;
while (socket.isConnected() && !socket.isClosed()) {
try {
command = SerializationUtils.deserialize(line);
} catch (Throwable e) {
log.warn("Deserialization exception occurred!", e);
continue;
Thread.yield();
byte[] line = readBytes(getSeparator());
if (line == null || line.length == 0) {
continue;
}
final Object command;
try {
command = SerializationUtils.deserialize(line);
} catch (Throwable e) {
log.warn("Deserialization exception occurred!", e);
continue;
}
if (command == null) {
continue;
}
if (!Sewy.getRegisteredDataTypes().contains(command.getClass())) {
log.error("Unexpected command received");
continue;
}
log.debug("Command received: " + command.getClass());
if (!(command instanceof AbstractCommand)) {
log.warn("Incorrect command received: " + command);
continue;
}
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);
}
if (command == null) {
continue;
}
if (!Sewy.getRegisteredDataTypes().contains(command.getClass())) {
log.error("Unexpected command received");
continue;
}
log.debug("Command received: " + command);
if (!(command instanceof AbstractCommand)) {
log.warn("Incorrect command received: " + command);
continue;
}
final Serializable response = onCommand((AbstractCommand) command);
log.debug(format("Response for %s is: %s", command, response));
writeBytes(SerializationUtils.serialize(response), getSeparator());
}
}
@ -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
* @param <T> generic type
*/
public <T extends AbstractCommand> void send(T command) {
log.debug("Start to send command: " + command);
public <T extends AbstractCommand> void send(T command) throws IOException {
log.debug("Start to send command: {}", command);
writeBytes(SerializationUtils.serialize(command), getSeparator());
}

View File

@ -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.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import java.util.function.Function;
import static java.lang.String.format;
@ -50,8 +43,8 @@ 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 host host to bind in order to start listen to clients
* @param port port to start listen to
* @param clientListenerClass client listen class to be used for communication
*/
public CommandServer(String host, int port, Class clientListenerClass) {
@ -59,9 +52,8 @@ 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 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)})
*/
@SuppressWarnings("unchecked")
@ -73,10 +65,12 @@ public class CommandServer extends Server<CommandClientListener> {
socket = server;
while (!server.isClosed()) {
final Socket client = server.accept();
final CommandClientListener clientListener = clientListenerConstructor.apply(client);
executor.execute(clientListener);
clients.add(clientListener);
if (!isMaximumClientsAchieved()) {
final Socket client = server.accept();
final CommandClientListener clientListener = clientListenerConstructor.apply(client);
executor.execute(clientListener);
clients.add(clientListener);
}
}
} catch (IOException 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
*
* @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);
for (CommandClientListener client : clients) {
client.send(command);
}
send(command, client -> {});
}
/**
* 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);
}
}
}
}

View File

@ -46,12 +46,14 @@ public class Server<T extends AbstractClientListener> {
protected ServerSocket socket;
private int maxClientsCount;
protected Server() {
}
/**
* @param host host to bind in order to start listen to clients
* @param port port to start listen to
* @param host host to bind in order to start listen to clients
* @param port port to start listen to
* @param clientListenerClass client listen class to be used for communication
*/
public Server(String host, int port, Class clientListenerClass) {
@ -59,9 +61,8 @@ 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 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)})
*/
@SuppressWarnings("unchecked")
@ -73,17 +74,19 @@ public class Server<T extends AbstractClientListener> {
socket = server;
while (!server.isClosed()) {
final Socket client = server.accept();
final T clientListener = clientListenerConstructor.apply(client);
executor.execute(clientListener);
clients.add(clientListener);
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);
}
});
}
/**
@ -98,15 +101,45 @@ public class Server<T extends AbstractClientListener> {
client.stop();
iterator.remove();
}
try {
if (socket != null) {
socket.close();
}
} catch (IOException e) {
log.error("Failed to close socket");
}
executor.shutdown();
}
/**
* 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;
}
}

View File

@ -33,7 +33,7 @@ public final class Sewy {
private static final ReentrantLock LOCK = new ReentrantLock();
private final List<Class<?>> registeredDataTypes = new CopyOnWriteArrayList<>();
private byte separator = SEPARATOR;
private byte[] separator = new byte[] { SEPARATOR };
/**
* Registers command in white list for further communications
@ -55,11 +55,11 @@ public final class Sewy {
}
}
public static byte getSeparator() {
public static byte[] getSeparator() {
return getInstance().separator;
}
public static void setSeparator(byte separator) {
public static void setSeparator(byte[] separator) {
try {
LOCK.lock();
getInstance().separator = separator;

View File

@ -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();
final String data = readLine();
writeLine(data);
try {
final String data = readLine();
writeLine(data);
} catch (Exception e) {
log.error("", e);
}
}
}
}

View File

@ -16,6 +16,7 @@
package me.bvn13.sewy;
import me.bvn13.sewy.command.AbstractCommand;
import me.bvn13.sewy.command.ComplexCommand;
import me.bvn13.sewy.command.PingCommand;
import me.bvn13.sewy.command.PongCommand;
import org.junit.jupiter.api.Assertions;
@ -23,6 +24,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
public class ServerTest {
@ -81,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);
@ -112,4 +114,43 @@ public class ServerTest {
Assertions.assertTrue(latency.get() > 0);
}
@ParameterizedTest
@ValueSource(ints = START_PORT + 7)
void wideSeparatorTest(int port) throws Exception {
Sewy.register(ComplexCommand.class);
Sewy.setSeparator(new byte[] { '\n', 'e', 'n', 'd', '\n' });
AtomicReference<ComplexCommand> check = new AtomicReference<>();
CommandServer server = new CommandServer("localhost", port, (socket) -> new CommandClientListener(socket) {
@Override
public AbstractCommand onCommand(AbstractCommand command) {
if (command instanceof ComplexCommand) {
check.set((ComplexCommand) command);
return null;
}
throw new IllegalArgumentException(command.toString());
}
});
CommandClient client = new CommandClient("localhost", port, (socket) -> new CommandClientListener(socket) {
@Override
public AbstractCommand onCommand(AbstractCommand command) {
throw new IllegalArgumentException(command.toString());
}
});
ComplexCommand command = new ComplexCommand();
command.add(new ComplexCommand.SimpleData("a1"));
command.add(new ComplexCommand.SimpleData("b2"));
command.add(new ComplexCommand.SimpleData("finish"));
client.send(command);
Thread.sleep(1000);
Assertions.assertNotNull(check.get());
Assertions.assertEquals(3, check.get().getDatum().size());
Assertions.assertEquals("a1", check.get().getDatum().get(0).getString());
Assertions.assertEquals("b2", check.get().getDatum().get(1).getString());
Assertions.assertEquals("finish", check.get().getDatum().get(2).getString());
}
}

View File

@ -0,0 +1,44 @@
package me.bvn13.sewy.command;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
public class ComplexCommand extends AbstractCommand {
private List<SimpleData> datum;
public ComplexCommand() {
datum = new ArrayList<>();
}
public ComplexCommand(List<SimpleData> datum) {
this.datum = datum;
}
public void add(SimpleData data) {
datum.add(data);
}
public ComplexCommand setDatum(List<SimpleData> datum) {
this.datum = datum;
return this;
}
public List<SimpleData> getDatum() {
return datum;
}
public static class SimpleData implements Serializable {
private final String string;
public SimpleData(String string) {
this.string = string;
}
public String getString() {
return string;
}
}
}