Re: [jetty-dev] JavaOne Async IO presentation

[Jarym <emrul.i@xxxxxxxxx>]

Thank you Greg,

I will give this a go. In the meantime I had to get something working quick
so changed my application design to support HTTP PUT for file-upload.  In
many ways this is better since I can stream data to my target destination
rather than have to mess around with parsing.  I still want to support HTTP
POST / multipart but I will give that a go when I get some personal time...

If anyone is interested in my implementation I share it below.  I borrowed
the Async proxy code from Jetty and replaced the proxy destination with an
async file channel.  It helped me better understand iterating callbacks too.
Performance seems to be decent but hard to tell since JMeter doesn't appear
great at uploading large files with high concurrency.  The idea of the code
below is not to be fast but also to reliably support many concurrent
uploads.


First the upload servlet: 
@WebServlet(name = "AsyncUploadServlet", urlPatterns =
"/AsyncUploadServlet/*", asyncSupported = true)
public class AsyncUploadServlet extends HttpServlet {
    @Override
    protected void doPut(HttpServletRequest req, HttpServletResponse resp)
throws ServletException, IOException {

        // Write everything to a folder named 'files' (it must already
exist)
        File file = File.createTempFile("post", ".html", new File("files"));
        AsynchronousFileChannel channel =
AsynchronousFileChannel.open(file.toPath(), StandardOpenOption.WRITE);// new
FileOutputStream(file, false).getChannel();

        // Begin async upload

        AsyncContext context = req.startAsync();
        context.setTimeout(60000 * 60 * 6);
        //request.getPart("");
        // set up async listener
        context.addListener(new AsyncListener() {
            public void onComplete(AsyncEvent event) throws IOException {
               
//event.getSuppliedResponse().getOutputStream().print("Complete");
            }

            public void onError(AsyncEvent event) {
                System.out.println(event.getThrowable());
            }

            public void onStartAsync(AsyncEvent event) {
            }

            public void onTimeout(AsyncEvent event) {
                System.out.println("my asyncListener.onTimeout");
            }
        });
        ServletInputStream input = req.getInputStream();
        ReadListener readListener = new UploadReadListener(req, resp,
context, channel);
        input.setReadListener(readListener);

    }


Then the UploadReadListener:


public class UploadReadListener extends IteratingCallback implements
ReadListener, CompletionHandler<Integer, Object> {
    private static final Logger logger =
LoggerFactory.getLogger(UploadReadListener.class);
    private final byte[] buffer = new byte[8192];
    private int writePosition = 0;
    private final HttpServletRequest request;
    private ServletInputStream input = null;
    private HttpServletResponse res = null;
    private AsyncContext ac = null;
    private int requestId = 0;
    private final AsynchronousFileChannel channel;

    UploadReadListener(HttpServletRequest req, HttpServletResponse r,
AsyncContext c, AsynchronousFileChannel chan) {
        //input = in;
        request = req;
        requestId = logger.isDebugEnabled() ? getRequestId(request) :
requestId;
        try {
            input = req.getInputStream();
        } catch (IOException e) {
            logger.error("Error calling getInputStream()", e);
        }
        res = r;
        ac = c;
        channel = chan;
    }

    public void onDataAvailable() throws IOException {
        iterate();
    }

    public void onAllDataRead() throws IOException {
        System.out.println("Data is all read");
        if (channel.isOpen())
            channel.close();
        request.getAsyncContext().complete();
    }

    public void onError(final Throwable t) {
        request.getAsyncContext().complete();
        t.printStackTrace();
        try {
            if (channel.isOpen()) channel.close();
        }
        catch (Exception ex) {

        }
    }

    @Override
    protected Action process() throws Exception {
        // First check for isReady() because it has
        // side effects, and then for isFinished().
        while (input.isReady() && !input.isFinished())
        {
            int read = input.read(buffer);
            if (logger.isDebugEnabled())
                logger.debug("{} asynchronous read {} bytes on {}",
requestId, read, input);
            if (read > 0)
            {
                if (logger.isDebugEnabled())
                    logger.debug("{} Sending to file destination: {} bytes",
requestId, read);
                onDataUploaded(request, buffer, 0, read);
                return Action.SCHEDULED;
            }
        }

        if (input.isFinished())
        {
            if (logger.isDebugEnabled())
                logger.debug("{} asynchronous read complete on {}",
requestId, input);
            return Action.SUCCEEDED;
        }
        else
        {
            if (logger.isDebugEnabled())
                logger.debug("{} asynchronous read pending on {}",
requestId, input);
            return Action.IDLE;
        }

    }


    protected void onDataUploaded(HttpServletRequest request, byte[] buffer,
int offset, int length)
    {
        channel.write(ByteBuffer.wrap(buffer, offset, length),
writePosition, null, this);
        writePosition += length-offset;
    }

    protected int getRequestId(HttpServletRequest request)
    {
        return System.identityHashCode(request);
    }

    @Override
    public void completed(Integer result, Object attachment) {
        succeeded();
    }

    @Override
    public void failed(Throwable exc, Object attachment) {
        if (logger.isDebugEnabled())
            logger.debug("async write Failed", exc);
    }
}





--
View this message in context: http://jetty.4.x6.nabble.com/jetty-dev-JavaOne-Async-IO-presentation-tp4963158p4963218.html
Sent from the Jetty Dev mailing list archive at Nabble.com.