Lesson 5 - Java Server - Connection Manager

In the previous lesson, Java Server - Server Thread, we created a basic code template for our Java server thread. Today we're going to create a class to handle incoming connections from potential clients.

Connection Manager

Again, we'll get started by creating a new connection package in the core package and create classes related to the connection manager there:

• IConnectionManager - An interface providing the connection manager methods
• IConnectionManagerFactory - A factory interface for creating the connection manager
• ConnectionManager - An implementation of the IConnectionManager interface
• ConnectionManagerFactory - An implementation of the IConnectionManagerFactory interface
• IClient - An interface providing methods of the connected client
• Client - An implementation of the IClient interface

Now we'll start to fill the interfaces with methods, we'll start with IConnectionManager:

public interface IConnectionManager {
    void addClient(Socket socket) throws IOException;
    void onServerStart();
    void onServerStop();
}

The interface contains the main addClient() method, by which the client should be added to the list of connected clients. There are also two auxiliary methods, onServerStart() and onServerStop(), which are called at the start (resp. at the end) of the server's lifetime.

The IConnectionManagerFactory interface will contain a single method, getConnectionManager(), to create instances of the IConnectionManager interface:

public interface IConnectionManagerFactory {
    IConnectionManager getConnectionManager(int maxClients, int waitingQueueSize);
}

The method accepts two parameters, maxClients and waitingQueueSize.

The IClient interface will represent the connected client, so it will include methods for communicating with that client:

public interface IClient {
    void sendMessageAsync(Object message);
    void sendMessage(Object message) throws IOException;
    void close();
}

We have two methods for sending a message, because one method will be blocking and the other asynchronous. The close() method will close the connection with the client.

Implementing the Interfaces

Let's implement the interfaces.

Client

We'll start with the Client class because it's the only one that won't depend on the connection manager. The class will implement two interfaces: IClient and Runnable:

public class Client implements IClient, Runnable {
}

We'll define instance constants:

private final Socket socket;
private final ObjectOutputStream writer;

and one instance variable:

private ConnectionClosedListener connectionClosedListener;

whose data type we're going to create a few lines below.

The constructor will (for now) have only one parameter of the Socket type:

Client(Socket socket) throws IOException {
    this.socket = socket;
    writer = new ObjectOutputStream(socket.getOutputStream());
}

In the constructor, we store the reference to the socket to the instance constant and initialize the writer constant as a new ObjectOutputStream.

Next, we'll implement methods required by the interface:

@Override
public void close() {
    try {
        socket.close();
    } catch (IOException e) {
        e.printStackTrace
    }
}

The close() method only delegates to the same method of the Socket. We'll implement the asynchronous message in the future:

@Override
public void sendMessageAsync(Object message) {
    // TODO send the message as asynchronous
}

The blocking version of the message takes the object and sends it to the client:

@Override
public void sendMessage(Object message) throws IOException {
    writer.writeObject(message);
}

An exception may be thrown if the connection has been terminated. Now let's take a look at the run() method:

@Override
public void run() {
    try (ObjectInputStream reader = new ObjectInputStream(socket.getInputStream())) {
        Object received;
        while ((received = reader.readObject()) != null) {
            // TODO process the received message
        }
    } catch (EOFException | SocketException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    } catch (ClassNotFoundException e) {
        // Should never happen
        e.printStackTrace();
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        if (connectionClosedListener != null) {
            connectionClosedListener.onConnectionClosed();
        }
        close();
    }
}

The run() method looks complicated, but actually does nothing but accept messages from the client. Much of the code is taken up by exception handling. Let's explain those:

• EOFException | SocketException - The client has closed the connection properly
• IOException - An unexpected communication exception occurred
• ClassNotFoundException - This exception should never occur if we follow the communication protocol we'll design in the future
• Exception - Catching general exceptions

In the finally block, we inform the listener that the connection has been terminated and call the close() method to close the socket and release resources.

Now we'll create a ConnectionClosedListener functional interface in the Client class, which will be the listening for termination of the connection:

@FunctionalInterface
public interface ConnectionClosedListener {
    void onConnectionClosed();
}

The interface contains the only onConnectionClosed() method that is called when the client-server connection has been terminated.

Finally we'll add a setter for this listener:

void setConnectionClosedListener(ConnectionClosedListener connectionClosedListener) {
    this.connectionClosedListener = connectionClosedListener;
}

The method will be visible only in the package the class is in, and nowhere else. We don't need someone else to set up the listener.

ConnectionManager

The class will only implement the IConnectionManager interface and doesn't need to be public again:

class ConnectionManager implements IConnectionManager {
}

There will be a collection in the class that will contain the connected clients:

private final List<IClient> clients = new ArrayList<>();

a threadpool for individual clients:

private final ExecutorService pool;

and a constant representing the maximum number of actively communicating clients:

final int maxClients;

The constructor will initialize the variables mentioned above:

@Inject
public ConnectionManager(ExecutorService pool,int maxClients) {
    this.pool = pool;
    this.maxClients = maxClients;
}

Before we implement the methods required by the interface, we'll create a private insertClientToListOrQueue() method to decide whether to put a client in the active client collection or the waiting queue:

private synchronized void insertClientToListOrQueue(Client client) {
    if (clients.size() < maxClients) {
        clients.add(client);
        client.setConnectionClosedListener(() -> {
            clients.remove(client);
        });
        pool.submit(client);
    } else {
        // TODO add the client to the waiting queue
    }
}

We'll leave the inserting of the client to the waiting queue for the next lesson.

We'll now implement the methods of the interface:

@Override
public void addClient(Socket socket) throws IOException {
    insertClientToListOrQueue(new Client(socket));
}

The addClient() method only delegates to the insertClientToListOrQueue() method.

For now, we won't do anything in the onServerStart() method:

@Override
public void onServerStart() {}

When the server is stopped, we'll go through all the clients and close the connection with them. Finally, we'll terminate the threadpool itself:

@Override
public void onServerStop() {
    for (IClient client : clients) {
        client.close();
    }
    pool.shutdown();
}

Connection Manager Factory

Finally, we need to implement the IConnectionManagerFactory interface:

@Singleton
public class ConnectionManagerFactory implements IConnectionManagerFactory {
    @Override
    public IConnectionManager getConnectionManager(int maxClients, int waitingQueueSize) {
        final ExecutorService pool = Executors.newFixedThreadPool(maxClients);
        return new ConnectionManager(pool, maxClients);
    }
}

In the method, we'll create a fixed-size threadpool and return a new instance of the ConnectionManager class. We'll register the factory in the ServerModule class as we always do:

bind(IConnectionManagerFactory.class).to(ConnectionManagerFactory.class);

Modifying the Server Thread Factory

Because we changed the constructor signature of the ServerThread class, we must modify the factory for this class. In the ServerThreadFactory class, we'll create a new IConnectionManagerFactory instance constant that will be initialized through a constructor parameter:

private final IConnectionManagerFactory connectionManagerFactory;
@Inject
public ServerThreadFactory(IConnectionManagerFactory connectionManagerFactory) {
    this.connectionManagerFactory = connectionManagerFactory;
}

We now have everything ready to create a new ServerThread class instance properly:

return new ServerThread(connectionManagerFactory.getConnectionManager(maxClients, waitingQueueSize), port);

Using the Connection Manager

In the ServerThread class, we'll create a new instance constant of the IConnectionManager type. Next, we'll add a constructor parameter to initialize the constant defined above. Now let's move to the run() method. At the very beginning of the method, we'll call the connectionManager.onServerStart(); method to give the connection manager the possibility to initialize (which we'll implement in the future). Next, when we accept a new client using the accept() method, we'll call the connection manager again, this time using the addClient() method, and pass the received socket to it. At the end of the run() method, we'll call the connectionManager.onServerStop() method to inform the connection manager that the server is shutting down and it should take care of any connected clients.

In the next lesson, Java Server - Client Dispatcher, we'll take care of the clients that need to be moved to the waiting queue.