Add handling concurrent connection limit

server/src/main/java/com/ftprx/server/ClientManager.java

@@ -0,0 +1,12 @@
+package com.ftprx.server;
+
+import com.ftprx.server.channel.Client;
+
+import java.util.List;
+
+/**
+ * Provides the APIs for managing connected client.
+ */
+public interface ClientManager {
+ List<Client> getClients();
+}

server/src/main/java/com/ftprx/server/Server.java

@@ -27,6 +27,8 @@
 import org.jetbrains.annotations.NotNull;
 import org.tinylog.Logger;
 
+import java.io.File;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.net.*;
 import java.time.Instant;
@@ -39,7 +41,7 @@
  * It receives commands from the client, sends replies,
  * and manages the server data transfer process.
  */
-public class Server {
+public class Server implements ClientManager {
  private static final int SO_TIMEOUT = 3000;
  private static Server instance = null;
  private final List<Client> clients;
@@ -55,6 +57,8 @@ private Server() {
  this.accountRepository = new FileAccountRepository("accounts.json", AccountFileFormat.JSON);
  this.status = ServerStatus.STOPPED;
  this.config = ConfigFactory.create(ServerConfig.class);
+
+ createConfigFileIfNotExists();
  }
 
  /**
@@ -67,7 +71,7 @@ public synchronized void start() {
  server = new ServerSocket();
  server.setSoTimeout(SO_TIMEOUT);
  server.bind(new InetSocketAddress(config.hostname(), config.port()));
- listenerThread = new ListenerThread(server);
+ listenerThread = new ListenerThread(server, this, config);
  listenerThread.registerClientConnectObserver(this::acceptClient);
  listenerThread.start();
  status = ServerStatus.RUNNING;
@@ -181,4 +185,18 @@ public synchronized static Server getInstance() {
  }
  return instance;
  }
+
+ private void createConfigFileIfNotExists() {
+ try {
+ File configFile = new File("server.properties");
+ if (!configFile.exists()) {
+ if (configFile.createNewFile()) {
+ Logger.info("Configuration file created: " + configFile.getAbsolutePath());
+ config.store(new FileOutputStream(configFile), "FtpRx server properties");
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
 }

server/src/main/java/com/ftprx/server/ServerConfig.java

@@ -5,12 +5,18 @@
 import org.aeonbits.owner.Mutable;
 import org.aeonbits.owner.Reloadable;
 
-@Config.Sources({"file:server.properties"})
+@Config.Sources("file:server.properties")
 public interface ServerConfig extends Mutable, Reloadable, Accessible {
 
+ @Key("port")
  @DefaultValue("21")
  int port();
 
+ @Key("hostname")
  @DefaultValue("127.0.0.1")
  String hostname();
+
+ @Key("concurrent-connection-limit")
+ @DefaultValue("0")
+ int concurrentConnectionLimit();
 }

server/src/main/java/com/ftprx/server/thread/ListenerThread.java

@@ -16,6 +16,8 @@
 
 package com.ftprx.server.thread;
 
+import com.ftprx.server.ClientManager;
+import com.ftprx.server.ServerConfig;
 import com.ftprx.server.util.SocketHelper;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -35,9 +37,15 @@ public class ListenerThread extends Thread {
  private static final String THREAD_NAME = "Protocol Interpreter Listening Thread";
  private final Set<ClientConnectObserver> observers;
  private final ServerSocket server;
+ private final ClientManager clientManager;
+ private final ServerConfig config;
 
- public ListenerThread(@NotNull ServerSocket server) {
+ public ListenerThread(@NotNull ServerSocket server,
+ @NotNull ClientManager clientManager,
+ @NotNull ServerConfig config) {
  this.server = Objects.requireNonNull(server, "Server must not be null");
+ this.clientManager = clientManager;
+ this.config = config;
  this.observers = Collections.newSetFromMap(new ConcurrentHashMap<>());
  setName(THREAD_NAME);
  }
@@ -47,6 +55,10 @@ public void run() {
  Logger.info("Listening for connections on port {}", server.getLocalPort());
  while (!Thread.currentThread().isInterrupted()) {
  if (SocketHelper.isServerSocketOpen(server)) {
+ if (!canAcceptNewClient()) {
+ Logger.warn("Client concurrent connection limit exceeded");
+ continue;
+ }
  try {
  Socket client = server.accept();
  notifyObservers(client);
@@ -58,6 +70,15 @@ public void run() {
  }
  }
 
+ private boolean canAcceptNewClient() {
+ if (config.concurrentConnectionLimit() != 0) {
+ var currentConnections = clientManager.getClients().size();
+ var isConnectionLimitExceeded = config.concurrentConnectionLimit() <= currentConnections;
+ return !isConnectionLimitExceeded;
+ }
+ return true;
+ }
+
  public void registerClientConnectObserver(@Nullable ClientConnectObserver observer) {
  Optional.ofNullable(observer).ifPresent(observers::add);
  }