Asynchronous non-blocking request handling

Mu Server optionally provides non-blocking reading of requests and writing of responses. Any handler can become an async handler by invoking the following method which returns a AsyncHandle:

AsyncHandle handle = request.handleAsync();

Calling the complete() method on that handle marks the response as complete.

Writing responses asynchronously

With an AsyncHandle, you can write byte buffers using one of two non-blocking methods, depending on whether you prefer a future or a callback to be notified when the write has completed:

• Future<Void> write(ByteBuffer data)
• void write(ByteBuffer data, DoneCallback callback)

Tip: to convert a string to a ByteBuffer, use Mutils.toByteBuffer(String)

In the following example, when a request for a JPG is made, the file is opened and segments of the file are read asynchronously. When some bytes are read from the file, they are passed on to the response. The write method with the DoneCallback is used so that after the bytes from the byte-buffer have been written to the response, another request to read more data is made.

public class AsyncWritingExample {
    public static void main(String[] args) {
        MuServer server = httpServer()
            .addHandler(Method.GET, "/guangzhou.jpg", (request, response, pathParams) -> {

                response.contentType(ContentTypes.IMAGE_JPEG);

                AsyncHandle asyncHandle = request.handleAsync();
                Path path = Paths.get("src/test/resources/guangzhou.jpg");

                AsynchronousFileChannel channel = AsynchronousFileChannel.open(path);
                FileChannelToResponsePump pump = new FileChannelToResponsePump(asyncHandle, channel);
                pump.pumpIt();

            })
            .start();

        System.out.println("Download the file at " + server.uri().resolve("/guangzhou.jpg"));

    }

    private static class FileChannelToResponsePump implements CompletionHandler<Integer, Object> {

        private final AsynchronousFileChannel channel;
        private final AsyncHandle asyncHandle;
        private final ByteBuffer buffer = ByteBuffer.allocate(8192);
        private long bytesSent = 0;

        FileChannelToResponsePump(AsyncHandle asyncHandle, AsynchronousFileChannel channel) {
            this.asyncHandle = asyncHandle;
            this.channel = channel;
        }

        public void pumpIt() {
            // Read from the file into the buffer, starting at position 0, and call 'this' instance's
            // completed method when read.
            channel.read(buffer, 0, null, this);
        }

        @Override
        public void completed(Integer bytesRead, Object attachment) {
            buffer.flip();
            if (bytesRead != -1) {

                // Write the jpg bytes from the buffer to the response channel
                asyncHandle.write(buffer, new DoneCallback() {

                    @Override
                    public void onComplete(Throwable errorIfFailed) throws Exception {
                        if (errorIfFailed == null) {
                            // Clear the buffer, and then start reading the next segment of data
                            buffer.clear();
                            bytesSent += bytesRead;
                            channel.read(buffer, bytesSent, null, FileChannelToResponsePump.this);
                        } else {
                            // client probably disconnected so stop reading and close the request
                            asyncHandle.complete();
                        }
                    }

                });
            } else {
                // File reading is complete. End the response and close the file channel.
                asyncHandle.complete();
                try {
                    channel.close();
                } catch (IOException ignored) { }
            }
        }

        @Override
        public void failed(Throwable exc, Object attachment) {
            // File reading failed. Log an error and send a 500 to the client.
            asyncHandle.complete(exc);
        }
    }
}

Note: the file handler described on the static resources page is uses the above approach so it's recommended you use that if you need to serve files.

Reading request bodies asynchronously

If you expect a request body, it can be read asynchronously by adding a RequestBodyListener with the AsyncHandle's setReadListener method:

public class AsyncRequestBodyExample {
    public static void main(String[] args) {
        MuServer server = httpServer()
            .addHandler(Method.POST, "/upload", new AsyncRequestBodyHandler())
            .addHandler(ResourceHandlerBuilder.classpathHandler("/samples"))
            .start();

        System.out.println("Upload a file at " + server.uri().resolve("/async-upload.html"));

    }

    private static class AsyncRequestBodyHandler implements RouteHandler {
        @Override
        public void handle(MuRequest request, MuResponse response, Map<String, String> pathParams) {

            AsyncHandle asyncHandle = request.handleAsync();

            long expectedSize = Long.parseLong(request.headers().get("content-length", "-1"));

            response.contentType("text/plain;charset=utf-8");
            AtomicLong total = new AtomicLong();

            asyncHandle.setReadListener(new RequestBodyListener() {

                public void onDataReceived(ByteBuffer buffer, DoneCallback doneCallback) {
                    int dataSize = buffer.remaining();
                    String message = "Received " + dataSize + " bytes - total so far is "
                        + total.addAndGet(dataSize) + " of " + expectedSize + " bytes\n";
                    // Note: do not call any blocking response writing methods from a data-received
                    // callback. Instead use the async writer on the handle.
                    asyncHandle.write(Mutils.toByteBuffer(message), doneCallback);
                    System.out.print(message);
                }

                public void onComplete() {
                    String message = "\n\nUpload complete. Received " + total.get() + " bytes.";
                    System.out.println(message);
                    asyncHandle.write(ByteBuffer.wrap(message.getBytes(UTF_8)));
                    asyncHandle.complete();
                }

                public void onError(Throwable t) {
                    System.out.println("\n\nError while reading: " + t);
                    asyncHandle.complete();
                }
            });

        }
    }
}

Note that if you do not need non-blocking request reading, the blocking request body readers can still be used by async handlers.