Netty Synthetic Client with Asynchronous Subscribers

Question

I am creating a server that uses commands from numerous sources such as JMS, SNMP, HTTP, etc. They are all asynchronous and working fine. The server supports one connection to one element of outdated equipment, which has a request / response architecture with the user TCP protocol. Ideally, I would like to use a single command, such as a block type method

public Response issueCommandToLegacyHardware(Command command) 

or this method of asynchronous type

 public Future<Response> issueCommandToLegacyHardware(Command command) 

I am relatively new to Netty and asynchronous programming, mostly learning it when I go. My current thought is that my LegacyHardwareClient class will have a public synchronized issueCommandToLegacyHardware(Command command) , make the client channel write obsolete hardware, and then take() from the SynchronousQueue<Response> , which will be blocked. The ChannelInboundHandler in the pipeline will offer() a Response on SynchronousQueue>Response> , which will allow take() unlock and receive data.

Is this too confusing? Are there any examples around synchronous Netty client implementations that I can look at? Are there any better methods for Netty? Obviously, I could only use standard Java sockets, but Netty's ability to analyze user protocols, along with the simplicity of maintanability, is too great to refuse.

UPDATE: As for the implementation, I used ArrayBlockingQueue <> (), and I used put () and remove (), not offer () and remove (). Since I wanted to make sure that subsequent requests to outdated equipment were sent only when answers were sent to any active requests, because outdated hardware behavior is unknown otherwise.

The reason for the suggestion () and remove () did not help me because the offer () command would not skip anything if take () was not actively blocking the other side. The converse is true that remove () will not return anything if there is no blocking of put () input that inserts data. I could not use put () / remove (), since the remove () statement was never reached, since no message was requested on the channel to trigger an event from which remove () would be called. I could not use the offer () / take (), because the offer () operator would return false, since the call to take () had not yet been completed. Using ArrayBlockingQueue <> () with a capacity of 1, he ensured that only one command was executed at a time. Any other commands are blocked until there is enough space to insert, with a capacity of 1, this means that it must be empty. The queue was emptied after receiving a response from outdated equipment. This provided good synchronous behavior with respect to legacy equipment, but provided an asynchronous API to users of legacy equipment, for which there were many.

Answer 1

Instead of developing a blocking application using SynchronousQueue<Response> , create it in a non-blocking way using SynchronousQueue<Promise<Response>> .

Then your public Future<Response> issueCommandToLegacyHardware(Command command) should use offer() to add DefaultPromise<>() to the queue, and then the netty pipeline can use remove() to get an answer for this request, I noticed that I used remove() instead of take() , since only under exceptional circumstances is no element present.

Quickly doing this can be:

 public class MyLastHandler extends SimpleInboundHandler<Response> { private final SynchronousQueue<Promise<Response>> queue; public MyLastHandler (SynchronousQueue<Promise<Response>> queue) { super(); this.queue = queue; } // The following is called messageReceived(ChannelHandlerContext, Response) in 5.0. @Override public void channelRead0(ChannelHandlerContext ctx, Response msg) { this.queue.remove().setSuccss(msg); // Or setFailure(Throwable) } } 

The above handler should be placed last in the chain.

An implementation of public Future<Response> issueCommandToLegacyHardware(Command command) might look like this:

 Channel channel = ....; SynchronousQueue<Promise<Response>> queue = ....; public Future<Response> issueCommandToLegacyHardware(Command command) { return issueCommandToLegacyHardware(command, channel.eventLoop().newPromise()); } public Future<Response> issueCommandToLegacyHardware(Command command, Promise<Response> promise) { queue.offer(promise); channel.write(command); return promise; } 

Using the overload approach on issueCommandToLegacyHardware also the design pattern used for Channel.write , which makes it really flexible.

This design pattern can be used in client code as follows:

 issueCommandToLegacyHardware( Command.TAKE_OVER_THE_WORLD_WITH_FIRE, channel.eventLoop().newPromise() ).addListener( (Future<Response> f) -> { System.out.println("We have taken over the world: " + f.get()); } ); 

The advantage of this design pattern is that no unnecessary locking is used anywhere, just asynchronous logic.

Appendix I: Javadoc:

Promise Future DefaultPromise