Sending multiple messages through only one SocketChannel

java nio socketchannel
Leo picture Leo · Mar 28, 2012 · Viewed 8.2k times · Source

After reading this tutorial: http://rox-xmlrpc.sourceforge.net/niotut/ (it's about writing non-blocking server and client, and I read NIO part, skiped SSL part), now I'm trying to rewrite my own client, but I'm stuck in a problem when trying to edit the client code.

Firstly, I want to let you see the client code of the tutorial, it's include 2 files:

But I edited NIOClient.java a bit in main function for explanation my problem as below:

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;

public class NIOClient implements Runnable {
// The host:port combination to connect to
private InetAddress hostAddress;
private int port;

// The selector we'll be monitoring
private Selector selector;

// The buffer into which we'll read data when it's available
private ByteBuffer readBuffer = ByteBuffer.allocate(8192);

// A list of PendingChange instances
private List pendingChanges = new LinkedList();

// Maps a SocketChannel to a list of ByteBuffer instances
private Map pendingData = new HashMap();

// Maps a SocketChannel to a RspHandler
private Map rspHandlers = Collections.synchronizedMap(new HashMap());

public NIOClient(InetAddress hostAddress, int port) throws IOException {
    this.hostAddress = hostAddress;
    this.port = port;
    this.selector = this.initSelector();
}

public void send(byte[] data, RspHandler handler) throws IOException {
    // Start a new connection
    SocketChannel socket = this.initiateConnection();

    // Register the response handler
    this.rspHandlers.put(socket, handler);

    // And queue the data we want written
    synchronized (this.pendingData) {
        List queue = (List) this.pendingData.get(socket);
        if (queue == null) {
            queue = new ArrayList();
            this.pendingData.put(socket, queue);
        }
        queue.add(ByteBuffer.wrap(data));
    }

    // Finally, wake up our selecting thread so it can make the required changes
    this.selector.wakeup();
}

public void run() {
    while (true) {
        try {
            // Process any pending changes
            synchronized (this.pendingChanges) {
                Iterator changes = this.pendingChanges.iterator();
                while (changes.hasNext()) {
                    ChangeRequest change = (ChangeRequest) changes.next();
                    switch (change.type) {
                    case ChangeRequest.CHANGEOPS:
                        SelectionKey key = change.socket.keyFor(this.selector);
                        key.interestOps(change.ops);
                        break;
                    case ChangeRequest.REGISTER:
                        change.socket.register(this.selector, change.ops);
                        break;
                    }
                }
                this.pendingChanges.clear();
            }

            // Wait for an event one of the registered channels
            this.selector.select();

            // Iterate over the set of keys for which events are available
            Iterator selectedKeys = this.selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = (SelectionKey) selectedKeys.next();
                selectedKeys.remove();

                if (!key.isValid()) {
                    continue;
                }

                // Check what event is available and deal with it
                if (key.isConnectable()) {
                    this.finishConnection(key);
                } else if (key.isReadable()) {
                    this.read(key);
                } else if (key.isWritable()) {
                    this.write(key);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

private void read(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    // Clear out our read buffer so it's ready for new data
    this.readBuffer.clear();

    // Attempt to read off the channel
    int numRead;
    try {
        numRead = socketChannel.read(this.readBuffer);
    } catch (IOException e) {
        // The remote forcibly closed the connection, cancel
        // the selection key and close the channel.
        key.cancel();
        socketChannel.close();
        return;
    }

    if (numRead == -1) {
        // Remote entity shut the socket down cleanly. Do the
        // same from our end and cancel the channel.
        key.channel().close();
        key.cancel();
        return;
    }

    // Handle the response
    this.handleResponse(socketChannel, this.readBuffer.array(), numRead);
}

private void handleResponse(SocketChannel socketChannel, byte[] data, int numRead) throws IOException {
    // Make a correctly sized copy of the data before handing it
    // to the client
    byte[] rspData = new byte[numRead];
    System.arraycopy(data, 0, rspData, 0, numRead);

    // Look up the handler for this channel
    RspHandler handler = (RspHandler) this.rspHandlers.get(socketChannel);

    // And pass the response to it
    if (handler.handleResponse(rspData)) {
        // The handler has seen enough, close the connection
        socketChannel.close();
        socketChannel.keyFor(this.selector).cancel();
    }
}

private void write(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    synchronized (this.pendingData) {
        List queue = (List) this.pendingData.get(socketChannel);

        // Write until there's not more data ...
        while (!queue.isEmpty()) {
            ByteBuffer buf = (ByteBuffer) queue.get(0);
            socketChannel.write(buf);
            if (buf.remaining() > 0) {
                // ... or the socket's buffer fills up
                break;
            }
            queue.remove(0);
        }

        if (queue.isEmpty()) {
            // We wrote away all data, so we're no longer interested
            // in writing on this socket. Switch back to waiting for
            // data.
            key.interestOps(SelectionKey.OP_READ);
        }
    }
}

private void finishConnection(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    // Finish the connection. If the connection operation failed
    // this will raise an IOException.
    try {
        socketChannel.finishConnect();
    } catch (IOException e) {
        // Cancel the channel's registration with our selector
        System.out.println(e);
        key.cancel();
        return;
    }

    // Register an interest in writing on this channel
    key.interestOps(SelectionKey.OP_WRITE);
}

private SocketChannel initiateConnection() throws IOException {
    // Create a non-blocking socket channel
    SocketChannel socketChannel = SocketChannel.open();
    socketChannel.configureBlocking(false);

    // Kick off connection establishment
    socketChannel.connect(new InetSocketAddress(this.hostAddress, this.port));

    // Queue a channel registration since the caller is not the 
    // selecting thread. As part of the registration we'll register
    // an interest in connection events. These are raised when a channel
    // is ready to complete connection establishment.
    synchronized(this.pendingChanges) {
        this.pendingChanges.add(new ChangeRequest(socketChannel, ChangeRequest.REGISTER, SelectionKey.OP_CONNECT));
    }

    return socketChannel;
}

private Selector initSelector() throws IOException {
    // Create a new selector
    return SelectorProvider.provider().openSelector();
}

public static void main(String[] args) {
    try {
        NIOClient client = new NIOClient(
                InetAddress.getByName("127.0.0.1"), 9090);
        Thread t = new Thread(client);
        t.setDaemon(true);
        t.start();

        // 1st
        client.send("hehe|||".getBytes());
        System.out.println("SEND: " + "hehe|||");
        handler.waitForResponse();

        System.out.println("------------");

        // 2nd
        client.send(("hehe|||" + " 2").getBytes());
        System.out.println("SEND: " + "hehe|||" + " 2");
        handler.waitForResponse();

    } catch (Exception e) {
        e.printStackTrace();
    }
}
}

My edited client just does a simple thing is send a message to server then receive a echoed message from server. Of course, the above code works well. It send 2 messages then received them back correctly.

But the things I don't want in the above client is: the send function calls this code:

    // Start a new connection
    SocketChannel socket = this.initiateConnection();

It means every distinguishing message will correspond with every distinguishing new SocketChannel, but now I want to use only one SocketChannel for sending many messages, so I change client like following code:

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;

public class MyClient implements Runnable {
// The host:port combination to connect to
private InetAddress hostAddress;
private int port;

// The selector we'll be monitoring
private Selector selector;

// The buffer into which we'll read data when it's available
private ByteBuffer readBuffer = ByteBuffer.allocate(8);

// A list of PendingChange instances
private List pendingChanges = new LinkedList();

// Maps a SocketChannel to a list of ByteBuffer instances
private Map pendingData = new HashMap();

// Maps a SocketChannel to a RspHandler
private Map rspHandlers = Collections.synchronizedMap(new HashMap());


private SocketChannel socket;
private static MyResponseHandler handler;

public MyClient(InetAddress hostAddress, int port) throws IOException {
    this.hostAddress = hostAddress;
    this.port = port;
    this.selector = this.initSelector();

    // Start a new connection
    socket = this.initiateConnection();

    handler = new MyResponseHandler();
    // Register the response handler
    this.rspHandlers.put(socket, handler);      
}

public void send(byte[] data) throws IOException {

    // And queue the data we want written
    synchronized (this.pendingData) {
        List queue = (List) this.pendingData.get(socket);
        if (queue == null) {
            queue = new ArrayList();
            this.pendingData.put(socket, queue);
        }
        queue.add(ByteBuffer.wrap(data));
    }

    // Finally, wake up our selecting thread so it can make the required changes
    this.selector.wakeup();
}

public void run() {
    while (true) {
        try {
            // Process any pending changes
            synchronized (this.pendingChanges) {
                Iterator changes = this.pendingChanges.iterator();
                while (changes.hasNext()) {
                    ChangeRequest change = (ChangeRequest) changes.next();
                    switch (change.type) {
                    case ChangeRequest.CHANGEOPS:
                        SelectionKey key = change.socket.keyFor(this.selector);
                        key.interestOps(change.ops);
                        break;
                    case ChangeRequest.REGISTER:
                        change.socket.register(this.selector, change.ops);
                        break;
                    }
                }
                this.pendingChanges.clear();
            }

            // Wait for an event one of the registered channels
            this.selector.select();

            // Iterate over the set of keys for which events are available
            Iterator selectedKeys = this.selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = (SelectionKey) selectedKeys.next();
                selectedKeys.remove();

                if (!key.isValid()) {
                    continue;
                }

                // Check what event is available and deal with it
                if (key.isConnectable()) {
                    this.finishConnection(key);
                } else if (key.isReadable()) {
                    this.read(key);
                } else if (key.isWritable()) {
                    this.write(key);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

private void read(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    // Clear out our read buffer so it's ready for new data
    this.readBuffer.clear();

    // Attempt to read off the channel
    int numRead;
    try {
        numRead = socketChannel.read(this.readBuffer);
    } catch (IOException e) {
        // The remote forcibly closed the connection, cancel
        // the selection key and close the channel.
        key.cancel();
        socketChannel.close();
        return;
    }

    if (numRead == -1) {
        // Remote entity shut the socket down cleanly. Do the
        // same from our end and cancel the channel.
        key.channel().close();
        key.cancel();
        return;
    }

    // Handle the response
    this.handleResponse(socketChannel, this.readBuffer.array(), numRead);
}

private void handleResponse(SocketChannel socketChannel, byte[] data, int numRead) throws IOException {
    // Make a correctly sized copy of the data before handing it
    // to the client
    byte[] rspData = new byte[numRead];
    System.arraycopy(data, 0, rspData, 0, numRead);

    // Look up the handler for this channel
    MyResponseHandler handler = (MyResponseHandler) this.rspHandlers.get(socketChannel);

    // And pass the response to it
    if (handler.handleResponse(rspData)) {
        // The handler has seen enough, close the connection
        socketChannel.close();
        socketChannel.keyFor(this.selector).cancel();
    }
}

private void write(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    synchronized (this.pendingData) {
        List queue = (List) this.pendingData.get(socketChannel);

        // Write until there's not more data ...
        while (!queue.isEmpty()) {
            ByteBuffer buf = (ByteBuffer) queue.remove(0);
            socketChannel.write(buf);

            //-- DEBUG --
            System.out.println("===>>> socketChannel.write: " + new String(buf.array()));

            if (buf.remaining() > 0) {
                // ... or the socket's buffer fills up
                break;
            }
        }

        if (queue.isEmpty()) {
            // We wrote away all data, so we're no longer interested
            // in writing on this socket. Switch back to waiting for
            // data.
            key.interestOps(SelectionKey.OP_READ);
        }
    }
}

private void finishConnection(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    // Finish the connection. If the connection operation failed
    // this will raise an IOException.
    try {
        socketChannel.finishConnect();
    } catch (IOException e) {
        // Cancel the channel's registration with our selector
        System.out.println(e);
        key.cancel();
        return;
    }

    // Register an interest in writing on this channel
    key.interestOps(SelectionKey.OP_WRITE);
}

private SocketChannel initiateConnection() throws IOException {
    // Create a non-blocking socket channel
    SocketChannel socketChannel = SocketChannel.open();
    socketChannel.configureBlocking(false);

    // Kick off connection establishment
    socketChannel.connect(new InetSocketAddress(this.hostAddress, this.port));

    // Queue a channel registration since the caller is not the 
    // selecting thread. As part of the registration we'll register
    // an interest in connection events. These are raised when a channel
    // is ready to complete connection establishment.
    synchronized(this.pendingChanges) {
        this.pendingChanges.add(new ChangeRequest(socketChannel, ChangeRequest.REGISTER, SelectionKey.OP_CONNECT));
    }

    return socketChannel;
}

private Selector initSelector() throws IOException {
    // Create a new selector
    return SelectorProvider.provider().openSelector();
}

public static void main(String[] args) {
    try {
        MyClient client = new MyClient(
                InetAddress.getByName("127.0.0.1"), 9090);
        Thread t = new Thread(client);
        t.setDaemon(true);
        t.start();

        // 1st
        client.send("hehe|||".getBytes());
        System.out.println("SEND: " + "hehe|||");
        handler.waitForResponse();

        System.out.println("------------");

        // 2nd
        client.send(("hehe|||" + " 2").getBytes());
        System.out.println("SEND: " + "hehe|||" + " 2");
        handler.waitForResponse();

    } catch (Exception e) {
        e.printStackTrace();
    }
}
}

But after running the above client, I just see only the 1st message is sent and received back, and after debugging, I know that the 2nd message is not sent, but I don't know why and how to resolve that problem.

Anyone knows the anwser?

Thank you for reading my very long question.

Answer

user207421 picture user207421 · Mar 29, 2012

You've started from the wrong place. There are numerous problems with that article. All that pending-change-queue stuff is an immense, unnecessary complication. Just wakeup() the selector if you need to register/deregister from another thread (although why you would need to do that at all is a complete mystery to me), and you can change interestOps any time the same way with perfect reliability, even if the FUD he is spreading about different implementations ever comes true.

The article has several other issues which show that the author doesn't really know what he's talking about. An IOException doesn't necessarily mean 'the remote forcibly closed the connection'. His finishConnection() method ignores the return value which, if false, means the connection is still pending, so it registers the channel beyond the OP_CONNECT stage prematurely. Closing a channel cancels the key, so all those cancel() calls that immediately precede or follow close() calls are redundant and can be removed (although there are places where he cancels without closing, which is also wrong where it occurs).

Further:

nowhere in the two methods just introduced do we request that the OP_CONNECT flag be set on the socket channel's selection key. If we did that we'd overwrite the OP_CONNECT flag and never complete the connection. And if we combined them then we'd run the risk of trying to write on an unconnected channel (or at least having to deal with that case)"

This is just complete Grade A nonsense. Setting OP_CONNECT twice, or 'combining them then', whatever that means, cannot possibly cause you to 'never complete the connection' or 'try to write on an unconnected channel'. He seems to think that setting a bit twice clears it.

Data has already been queued (or we wouldn't be establishing a connection in the first place).

A strange and unexplained assumption.

Instead of that pretty dubious mess, I would have a good look at the 'Taming the NIO Circus' thread he cites, if you can still find it after the Oracle forum migration. Disclaimer: I wrote some of it.

Related questions

Java.nio Channels and TLS

How do I secure a Java SocketChannel, ServerSocketChannel or, perhaps even, a DatagramChannel with TLS? I know that there are some frameworks (#1 #2) that advertise to be able, but I want to know if it is possible to achieve this with …

Read More
Read timeout for an NIO SocketChannel?

What is the best way to set a timeout to close a NIO SocketChannel if there is no data is received for a certain period after the connection is established?

Read More
Java NIO: How to know when SocketChannel read() is complete with non-blocking I/O

I am currently using a non-blocking SocketChannel (Java 1.6) to act as a client to a Redis server. Redis accepts plain-text commands directly over a socket, terminated by CRLF and responds in-like, a quick example: SEND: 'PING\r\n' RECV: '+…

Read More