Reading and Writing Messages from a Socket

Question

This is code from my android client which communicates over wifi to a small server program (not coded in java). This is my first time playing around with sockets, so I'm sure there are lots of little gotchas to worry about. For this project I am limited to Java 7.

The protocol for sending a message is

• 4 Bytes for an integer message size M
• M Bytes for the message payload

The 4 Byte size is trimmed from the payload when reviving and added to the payload when sending.

I decided that the user might want to be able to add and drop connections, so I made a class to handle the socket, input stream and output stream. I didn't create every function of the three classes, I just picked a functions I needed.

public final class SocketIO {

    private Socket socket;

    private InputStream input;

    private OutputStream output;

    public SocketIO(){
        socket = null;
        input = null;
        output = null;
    }

    public void connect(final String ip, final int port) throws IOException {
        close();
        socket = new Socket(ip, port);
        input = socket.getInputStream();
        output = socket.getOutputStream();
    }

    public boolean isConnected() {
        if (socket == null){
            return false;
        }
        else if (socket.isConnected()){
            return true;
        }
        else {
            socket = null;
            return false;
        }
    }

    public void close() throws IOException {
        if (socket != null) {
            socket.close();
        }
        if (input != null) {
            input.close();
        }
        if (output != null) {
            output.close();
        }
    }

    public void read(byte[] bytes) throws IOException {
        if (!isConnected()) {
            throw new IOException("Socket not connected");
        }
        input.read(bytes);

    }

    public int read(byte[] bytes, int i, int remaining) throws IOException {
        if (!isConnected()) {
            throw new IOException("Socket not connected");
        }
        return input.read(bytes, i, remaining);
    }

    public void write(byte[] bytes) throws IOException {
        if (!isConnected()) {
            throw new IOException("Socket not connected");
        }
        output.write(bytes);
    }

    public void flush() throws IOException {
        if (!isConnected()) {
            throw new IOException("Socket not connected");
        }
        output.flush();
    }
}

I created two Runnable classes to read and write to the SocketIO class. They each require a Queue<byte[]> to accept incoming or send outgoing message payloads. For this I have been using a ConcurrentLinkedQueue.

public final class ByteWriter implements Runnable {

    private final Queue<byte[]> queue;

    final private SocketIO io;

    public ByteWriter(final SocketIO serverIo, final Queue<byte[]> source) {
        io = serverIo;
        queue = source;
    }

    @Override
    public final void run() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                if (io.isConnected()){
                    while (queue.size() > 0) {
                        final byte[] sendBytes = queue.remove();
                        io.write(ByteBuffer.allocate(4).putInt(sendBytes.length).array());
                        io.write(sendBytes);
                        io.flush();
                    }
                }
                Thread.sleep(100);
            }
        } catch (final IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

And here is the Reader:

public final class ByteReader implements Runnable {

    private final Queue<byte[]> queue;

    final private SocketIO io;

    public ByteReader(final SocketIO serverIo, final Queue<byte[]> destination) {
        io = serverIo;
        queue = destination;
    }

    @Override
    public final void run() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                if (!io.isConnected()) {
                    Thread.sleep(100);
                    continue;
                }
                final byte[] receiveSizeBytes = new byte[4];
                io.read(receiveSizeBytes);
                final int receiveSize = ByteBuffer.wrap(receiveSizeBytes).getInt();

                final byte[] receiveBytes = new byte[receiveSize];

                int bytesRemaining = receiveSize;
                int offset = 0;
                while (bytesRemaining > 0) {
                    final int readCount = io.read(receiveBytes, offset, bytesRemaining);
                    offset += readCount;
                    bytesRemaining -= readCount;
                    Thread.sleep(10);
                }
                queue.add(receiveBytes);
                Thread.sleep(100);
            }
        } catch (final IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

The class which manages these new objects:

public final class ServerConnection implements Runnable {

    private final String ip;

    private final int port;

    private final SocketIO io;

    private final Thread writer;

    private final Thread reader;

    public ServerConnection(
            final String ipAddress,
            final int portNumber,
            final Queue<byte[]> incoming,
            final Queue<byte[]> outgoing) {
        ip = ipAddress;
        port = portNumber;

        io = new SocketIO();
        writer = new Thread(new ByteWriter(io, outgoing));
        reader = new Thread(new ByteReader(io, incoming));
    }

    @Override
    public void run() {
        writer.start();
        reader.start();

        try {
            io.connect(ip, port);
            while (!Thread.currentThread().isInterrupted()) {
                if (!io.isConnected()) {
                    break;
                }
                Thread.sleep(100);
            }
        } catch (final IOException | InterruptedException e) {
            e.printStackTrace();
        }

        try {
            io.close();
            writer.interrupt();
            reader.interrupt();
            writer.join();
            reader.join();
        } catch (final IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Here is an example of establishing the connection in my program:

public class ControlPanel extends Activity {

    private final Queue<byte[]> incoming = new ConcurrentLinkedQueue<>();

    private final Queue<byte[]> outgoing = new ConcurrentLinkedQueue<>();

    private Thread connection = new Thread();

.......

    public void connectServer(final View view){
        connection.interrupt();
        try {
            connection.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        final EditText editText = (EditText) findViewById(R.id.editText);
        connection = new Thread(new ServerConnection(editText.getText().toString(), 1337, incoming, outgoing));
        connection.start();
    }

Answer 1

Resource leaks

You isConnected method has the potential to leak open resources. If the method is called while isConnected() on the underlying socket returns false, the socket will be set to null without calling close() on the socket.

public boolean isConnected() {
    if (socket == null){
        return false;
    }
    else if (socket.isConnected()){
        return true;
    }
    else {
        try {
            socket.close();
        } catch (IOException ignored) {
        }
        socket = null;
        return false;
    }
}

Remember that the close() explicitly states that the method must be idempotent, so closing a socket 2 times shouldn't give you a error.

Inconsistent order of modifers

private final Queue<byte[]> queue;
final private SocketIO io;

Choosing a consistent order of modifers makes your code looks better, I would go for private final, as this is used everywhere in your project

private final Queue<byte[]> queue;
private final SocketIO io;

Swallowing InterruptedException

   try {
       connection.join();
   } catch (InterruptedException e) {
       e.printStackTrace();
   }

Swallowing/suppressing this exception isn't good practise, as it blocks the propagation of the interrupted state. If you are forced to catch it, you should either make a looping structure that set the threads interrupted state at the end, or set the threads intterupted state directly by calling Thread.currentThread().interrupt();

    try {
        connection.join();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }

Unused thread

private Thread connection = new Thread();

By defining a Thread that is never started, you have the potential to create a memory leak.

You should use either use a null object, or a wrapper object that you crafted using the null-object design pattern.

private Thread connection = null;

A other way to this, is to start the thread directly. This is considered more as a hack than a proper solution.

private Thread connection = new Thread();
{
     connection.start();
}

Sleeping Threads

You use a large number of call to Thread.sleep() in your code, this is more like a anti-pattern for proper usage of locking. In the ideal world, you should let the threads wait for each other using Object.wait() in combination with Object.notifyAll(). Incase of your socket, don't sleep, but call the read() method again. Using this mechanism you can signal multiple conditions, like a new thing to read, but also if there is nothing left to write.

Large number of new Threads

Do you really need to create and destroy that number of Threads? a better implementation might call the methods of Executors to create its tasks, as it has automatic Thread management, the newCachedThreadPool has the performance characteris you require, automatic new threads when needed, but reusing old Threads when they are free.

Unbuffered IO

public void connect(final String ip, final int port) throws IOException {
    close();
    socket = new Socket(ip, port);
    input = socket.getInputStream();
    output = socket.getOutputStream();
}

You mainly use unbuffered io streams to the socket, this is really expansive as the calls propagate down the networking stack of the operating system. Wrapping the streams in BufferedInputStream and BufferedOutputStream makes the calls quicker as there are less half filled packets.

public void connect(final String ip, final int port) throws IOException {
    close();
    socket = new Socket(ip, port);
    input = new BufferedInputStream(socket.getInputStream());
    output = new BufferedOutputStream(socket.getOutputStream());
}

To many flushes

            if (io.isConnected()){
                while (queue.size() > 0) {
                    final byte[] sendBytes = queue.remove();
                    io.write(ByteBuffer.allocate(4).putInt(sendBytes.length).array());
                    io.write(sendBytes);
                    io.flush();
                }
            }

You flushing after every packet, by placing the final flush call after the loop, and wrapping everything into a if-statement that checks the size, you can send more packets in just one flush() call, this means only 1 tcp packet even if you send 5 small application specific packets

            if (io.isConnected()){
                if(!queue.isEmpty()) {
                    do {
                        final byte[] sendBytes = queue.remove();
                        io.write(ByteBuffer.allocate(4).putInt(sendBytes.length).array());
                        io.write(sendBytes);
                    } while (queue.size() > 0);
                    io.flush();
                }
            }