Lesson 7 - Java Server - Writing Thread

In the previous lesson, Java Server - Client Dispatcher, we discussed the queue of waiting clients. Today we're going to send messages to the client using a separate thread. This will make the server response to feel better.

Writing Thread

Until now, we were sending messages from the server in a thread that was processing incoming data. This is called synchronous processing. The problem with this approach is that we're doing an I/O operation, which is time-consuming. A much better solution is to handle all these activities in a separate thread. To do this, we'll create a thread which only goal will be: send messages from the server to the client.

In the core package, we'll create a new writer package, in which we'll create classes for the writer thread. We'll start by creating interfaces to define the methods as we always do. We'll create a IWriterThread interface and let it inherit from IThreadControl. The interface will have a single sendMessage() method. The method will accept two parameters: ObjectOutputStream to which the message will be written, and Object which is the message itself:

public interface IWriterThread extends IThreadControl {
    void sendMessage(ObjectOutputStream writer, Object message);
}

Next, we'll create a WriterThread class that will inherit from the Thread class and implement the interface declared above:

public class WriterThread extends Thread implements IWriterThread {
}

In the class we'll declare two instance constants and two variables:

private final Semaphore semaphore = new Semaphore(0);
private final Queue<QueueTuple> messageQueue = new ConcurrentLinkedQueue<>();
private boolean working = false;
private boolean interrupt = false;

The principle of the semaphore here is the same as in the previous lesson. The thread will sleep at the semaphore until it receives a signal to work. messageQueue contains the messages to be sent. It's of a QueueTuple type which we're going to create soon. The working variable indicates whether the thread is working or sleeping at the semaphore. The interrupt variable has the same purpose as in the previous lessons, when we created classes that inherit from the Thread class.

We'll leave the class constructor empty this time. To improve readability, we'll give the thread a better name. In the constructor, we'll call the Thread constructor with the overload talking the thread name as a parameter:

public WriterThread() {
    super("WriterThread");
}

Thread Methods

The interface tells us that we need to implement two methods: sendMessage() and shutdown(). Let's look at the first one. Sending a message will be nothing but adding a message to the message queue, and if the thread isn't working, we'll wake it up and set the working variable to true:

@Override
public void sendMessage(ObjectOutputStream writer, Object message) {
    messageQueue.add(new QueueTuple(outputStream, message));
    if (!working) {
        working = true;
        semaphore.release();
    }
}

The shutdown() method will be the same as in the previous lesson:

@Override
public void shutdown() {
    interrupt = true;
    semaphore.release();
    try {
        join();
    } catch (InterruptedException ignored) {}
}

We set the interrupt variable to true, release the semaphore, and wait for the writer thread to end.

Thread Logic Code

Now we're going to implement the code itself to take care of sending messages in our thread. The run() method will look like this:

@Override
public void run() {
    while(!interrupt) {
        while(messageQueue.isEmpty() && !interrupt) {
            try {
                semaphore.acquire();
            } catch (InterruptedException ignored) {}
        }
        working = true;
        while(!messageQueue.isEmpty()) {
            final QueueTuple entry = messageQueue.poll();
            try {
                entry.writer.writeLine(entry.message);
                entry.writer.flush();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        working = false;
    }
}

The code is very similar to what we implemented in the client dispatcher. We have an infinite loop depending on the interrupt variable. Then there's another infinite loop to make the thread wait at the semaphore. We discussed why we use a loop to make the thread waiting in the last lesson. In the next loop, we iterate through the message queue, and all messages that there are sent gradually. When there are no messages in the queue, the thread goes back to the semaphore.

All that remains is to declare the QueueTuple class. It's a simple container that will have two constants. It's no coincidence these constants are obtained in the sendMessage() method. The whole class code is below:

private static final class QueueTuple {
    final Object message;
    final ObjectOutputStream writer;

    private QueueTuple(ObjectOutputStream writer, Object message) {
        this.message = message;
        this.writer = writer;
    }
}

Nest the class in the WriterThread class because we won't use it anywhere else.

Wiring the Logic

In the ConnectionManager class we'll create an instance constant with the writer thread and initialize it in the constructor from a parameter:

private final IWriterThread writerThread;
public ConnectionManager(IClientDispatcher clientDispatcher, IWriterThread writerThread,
    ExecutorService pool, int maxClients) {
    this.clientDispatcher = clientDispatcher;
    this.writerThread = writerThread;
    this.pool = pool;
    this.maxClients = maxClients;
}

We'll run the thread in the onServerStart() method in the connection manager as soon as the client dispatcher starts:

writerThread.start();

We'll terminate the thread in the onServerStop() method again after stopping the client dispatcher:

writerThread.shutdown();

Next, we need to modify the Client class to have access to the writer thread. So we'll create an instance constant with the writer thread in it, but we'll initialize it in the constructor which will get it as a parameter:

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

I'll let the kind reader to create a Client class instance on an incoming connection.

Finally, we have to modify the connection manager factory because we have modified the constructor of the ConnectionManager class again. In the factory we'll add a new instance constant of the IWriterThread type which will be initialized in the constructor using a parameter:

private final IWriterThread writerThread;

@Inject
public ConnectionManagerFactory(IClientDispatcherFactory clientDispatcherFactory,
    IWriterThread writerThread) {
    this.clientDispatcherFactory = clientDispatcherFactory;
    this.writerThread = writerThread;
}

Creating an instance should no longer cause problems:

return new ConnectionManager(clientDispatcherFactory.getClientDispatcher(waitingQueueSize), writerThread, pool, maxClients);

That would be all of today's lesson. We didn't change the functionality, we just improved the server response. Next time, in the lesson Java Server - Communication Protocol, we'll design the communication protocol.

In the next lesson, Java Server - Communication Protocol, we'll design a communication protocol.