Lesson 17 - Java Chat - Client - Server Connection Part 1

In the previous lesson, Java Chat - Client - View Local Servers, we focused on displaying local servers.

In today's Java tutorial, we're going to design the functions and interfaces of the class that will be responsible for establishing and maintaining a connection to the server.

Communication Requirements

All communication with the Java server must be asynchronous to prevent the GUI from freezing while sending messages. Asynchronous communication will be handled by two threads:

• ReaderThread - the thread will receive messages from the server
• WriterThread - The thread will send messages to the server

Next, it'll be necessary to process received messages from the server in some way. If the connection is interrupted during communication, it's a good idea to react somehow. We'll also need an enumeration that will represent the connection status.

For today's work we'll create a new net package, which will again be located next to the other controller and service packages.

Preparing Interfaces

Before we start implementing server communication threads, we'll prepare two auxiliary interfaces: OnDataReceivedListener and LostConnectionHandler:

@FunctionalInterface
public interface OnDataReceivedListener {
    void onDataReceived(IMessage message);
}

The OnDataReceivedListener interface, as the name suggests, contains an onDataReceived() method, which is called each time a message arrives from the server.

@FunctionalInterface
public interface LostConnectionHandler {
    void onLostConnection();
}

Using the LostConnectionHandler interface and its onLostConnection() method, we'll be informed that the connection to the server has been lost for unknown reasons (the server has shut down, the internet cable has been disconnected, ...).

Connection Status

We'll represent the state of the connection by a ConnectionState enumeration, which will have these values:

• DISCONNECTED - the client is disconnected from the server
• CONNECTING - The client is trying to connect to the server
• CONNECTED - the client has successfully established a connection with the server

public enum ConnectionState {
    DISCONNECTED, CONNECTING, CONNECTED;
}

Reader Thread

After declaring the basic interfaces, we can create the reading thread:

public class ReaderThread extends Thread {
    private final InputStream inputStream;
    private final OnDataReceivedListener dataReceivedListener;
    private final LostConnectionHandler lostConnectionHandler;
    private boolean interrupt = false;

    public ReaderThread(final InputStream inputStream, OnDataReceivedListener dataReceivedListener,
        LostConnectionHandler lostConnectionHandler) {
        super("ReaderThread");
        this.lostConnectionHandler = lostConnectionHandler;
        assert dataReceivedListener != null;
        this.dataReceivedListener = dataReceivedListener;
        this.inputStream = inputStream;
    }

    public void shutdown() {
        interrupt = true;
    }

    @Override
    public void run() {
        try (final ObjectInputStream reader = new ObjectInputStream(inputStream)) {
            IMessage received;
            while ((received = (IMessage) reader.readObject()) != null && !interrupt) {
                dataReceivedListener.onDataReceived(received);
            }
        } catch (EOFException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (lostConnectionHandler != null) {
                lostConnectionHandler.onLostConnection();
            }
        }
    }
}

The class contains three instance constants:

• inputStream - the stream from which we'll read messages
• dataReceivedListener - a listener for incoming messages
• lostConnectionHandler - a lost connection handler

The interrupt variable serves as an indicator whether the thread should be safely terminated. Reading data from the server itself works on the same principle as reading data from the client on the server. The readObject() of the ObjectInputStream instance is called in an infinite loop. When a message arrives, the onDataReceived() method is called to pass the message for processing.

Writer Thread

The client-side writer thread will be very similar to the server-side writer thread. The only difference is that on the client-side we don't have to store information where the message will be sent, as it's always for the server.

public class WriterThread extends Thread {

    private final Semaphore semaphore = new Semaphore(0);
    private final Queue<IMessage> messageQueue = new ConcurrentLinkedQueue<>();
    private final AtomicBoolean working = new AtomicBoolean(false);
    private final ObjectOutputStream writer;
    private final LostConnectionHandler lostConnectionHandler;
    private boolean interrupt = false;

    public WriterThread(final OutputStream outputStream, LostConnectionHandler lostConnectionHandler) throws IOException {
        super("WriterThread");
        this.lostConnectionHandler = lostConnectionHandler;
        this.writer = new ObjectOutputStream(outputStream);
    }

    public void shutdown() {
        interrupt = true;
        messageQueue.clear();
        semaphore.release();
    }

    public void addMessageToQueue(IMessage message) {
        messageQueue.add(message);
        if (!working.get()) {
            semaphore.release();
        }
    }

    @Override
    public void run() {
        do {
            while(messageQueue.isEmpty() && !interrupt) {
                try {
                    semaphore.acquire();
                } catch (InterruptedException ignored) {}
            }

            working.set(true);
            while (!messageQueue.isEmpty()) {
                final IMessage msg = messageQueue.poll();
                assert msg != null;
                try {
                    writer.writeObject(msg);
                    writer.flush();
                } catch (IOException e) {
                    e.printStackTrace();
                    interrupt = true;
                    if (lostConnectionHandler != null) {
                        lostConnectionHandler.onLostConnection();
                    }
                }
            }
            working.set(false);
        } while(!interrupt);
    }
}

Client Communicator

Before implementation, we'll first design the features that the communicator will have. Let's write these features into an interface that we'll later implement.

Feature Design

The communicator will have the following features:

• connect() - Connects to the server
• disconnect() - Disconnects from the current server
• sendMessage() - Sends a message to the server
• getConnectionState() - Gets the status of the current connection with the server
• getConnectedServerName() - Gets the name of the connected server
• (un)registerMessageObserver() - Adds/removes recipients of incoming messages

Creating the Interface

Once we've made it clear what features the communicator will have, we'll create an interface that describes these features as methods:

public interface IClientCommunicationService {
    CompletableFuture<Boolean> connect(String host, int port);
    CompletableFuture<Boolean> disconnect();
    void sendMessage(IMessage message);
    CompletableFuture<IMessage> sendMessageFuture(IMessage message);
    void registerMessageObserver(String messageType, OnDataReceivedListener listener);
    void unregisterMessageObserver(String messageType, OnDataReceivedListener listener);
    ReadOnlyObjectProperty<ConnectionState> connectionStateProperty();
    ConnectionState getConnectionState();
    String getConnectedServerName();
}

The methods correspond to the features we listed above. It's worth mentioning there are 2 sendMessage() methods. Both versions send a message to the server. The only difference is that the method with the Future suffix will expect a response from the server. I'd stop here for a moment and describe using the CompletableFuture class because understanding this is very important.

CompletableFuture

Let's start from the beginning. To define an asynchronous operation we use the Runnable interface, which contains the run() method. Instances of this interface run in separate threads so the "demanding" calculation is run in parallel. The problem with this approach is that it doesn't define a response to the calculation result or that it failed. Over time, Java developers added the Future<> interface, which represents a result of an asynchronous calculation. The main benefit is in the definition of the get() method, which returns the result of the asynchronous calculation. At the same time, however, a new problem has emerged: the call to the get() method is blocking, so if the intensive calculation didn't complete before calling the get() method, the calling thread has been blocked. If the calling thread was the GUI thread, the form would freeze.

The solution finally came with the CompletableFuture class which eliminated the use of the get() method. This class implements the CompletionStage interface, which defines a fluent API to describe exactly how the results of complex calculations will be processed. The main interface methods are:

• thenApply() - A transformation method that takes an input, modifies it somehow, and returns a new output
• thenAccept() - An end method that takes an input and processes it
• exceptionally() - Allows us to handle exceptions

The thenApply() and thenAccept() methods also exist with the Async suffix. Those methods can be run in the context of a different thread than the one in which they were called. The interface, of course, contains many other methods, but for us these are the most important. We'll explain everything while implementing the communicator.

That would be all for today's lesson.

Next time, in the lesson Java Chat - Client - Server Connection Part 2, we'll focus only on implementing the communicator.