Returning a Future<String> for the method is simple, we are going to implement the following method signature:
public Futute<String> sendMessage(String msg) {
The is relatively easy to do when you are known with the async programming structures. To solve the design problem, we are going to do the following steps:
When a message is written, add a Promise<String> to a ArrayBlockingQueue<Promise>
This will serve as a list of what messages have recently been send, and allows us to change our Future<String> objects return result.
When a message arrives back into the handler, resolve it against the head of the Queue
This allows us to get the correct future to change.
Update the state of the Promise<String>
We call promise.setSuccess() to finally set the state on the object, this will propagate back to the future object.
Example code
public class ClientHandler extends SimpleChannelInboundHandler<String> {
private ChannelHandlerContext ctx;
private BlockingQueue<Promise<String>> messageList = new ArrayBlockingQueue<>(16);
@Override
public void channelActive(ChannelHandlerContext ctx) {
super.channelActive(ctx);
this.ctx = ctx;
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
super.channelInactive(ctx);
synchronized(this){
Promise<String> prom;
while((prom = messageList.poll()) != null)
prom.setFailure(new IOException("Connection lost"));
messageList = null;
}
}
public Future<String> sendMessage(String message) {
if(ctx == null)
throw new IllegalStateException();
return sendMessage(message, ctx.executor().newPromise());
}
public Future<String> sendMessage(String message, Promise<String> prom) {
synchronized(this){
if(messageList == null) {
// Connection closed
prom.setFailure(new IllegalStateException());
} else if(messageList.offer(prom)) {
// Connection open and message accepted
ctx.writeAndFlush(message).addListener();
} else {
// Connection open and message rejected
prom.setFailure(new BufferOverflowException());
}
return prom;
}
}
@Override
protected void messageReceived(ChannelHandlerContext ctx, String msg) {
synchronized(this){
if(messageList != null) {
messageList.poll().setSuccess(msg);
}
}
}
}
Documentation breakdown
private ChannelHandlerContext ctx;
Used to store our reference to the ChannelHandlerContext, we use this so we can create promises
private BlockingQueue<Promise<String>> messageList = new ArrayBlockingQueue<>();
We keep the past messages in this list so we can change the result of the future
public void channelActive(ChannelHandlerContext ctx)
Called by netty when the connection becomes active. Init our variables here.
public void channelInactive(ChannelHandlerContext ctx)
Called by netty when the connection becomes inactive, either due to error or normal connection close.
protected void messageReceived(ChannelHandlerContext ctx, String msg)
Called by netty when a new message arrives, here pick out the head of the queue, and then we call setsuccess on it.
Warning advise
When using futures, there is 1 thing you need to lookout for, do not call get() from 1 of the netty threads if the future isn't done yet, failure to follow this simple rule will either result in a deadlock or a BlockingOperationException.
