序

本文主要研究一下AsyncHttpClient的IOExceptionFilter

IOExceptionFilter

org/asynchttpclient/filter/IOExceptionFilter.java

/*** This filter is invoked when an {@link java.io.IOException} occurs during an http transaction.*/
public interface IOExceptionFilter {/*** An {@link org.asynchttpclient.AsyncHttpClient} will invoke {@link IOExceptionFilter#filter} and will* use the returned {@link FilterContext} to replay the {@link org.asynchttpclient.Request} or abort the processing.** @param ctx a {@link FilterContext}* @param <T> the handler result type* @return {@link FilterContext}. The {@link FilterContext} instance may not the same as the original one.* @throws FilterException to interrupt the filter processing.*/<T> FilterContext<T> filter(FilterContext<T> ctx) throws FilterException;
}

IOExceptionFilter接口定义了filter方法，在捕获到IOException的时候执行

ResumableIOExceptionFilter

org/asynchttpclient/handler/resumable/ResumableIOExceptionFilter.java

/*** Simple {@link org.asynchttpclient.filter.IOExceptionFilter} that replay the current {@link org.asynchttpclient.Request} using a {@link ResumableAsyncHandler}*/
public class ResumableIOExceptionFilter implements IOExceptionFilter {public <T> FilterContext<T> filter(FilterContext<T> ctx) {if (ctx.getIOException() != null && ctx.getAsyncHandler() instanceof ResumableAsyncHandler) {Request request = ResumableAsyncHandler.class.cast(ctx.getAsyncHandler()).adjustRequestRange(ctx.getRequest());return new FilterContext.FilterContextBuilder<>(ctx).request(request).replayRequest(true).build();}return ctx;}
}

ResumableIOExceptionFilter实现了IOExceptionFilter接口，它在asyncHandler是ResumableAsyncHandler类型的时候执行adjustRequestRange方法获取新的request，然后再通过new FilterContext.FilterContextBuilder<>(ctx).request(request).replayRequest(true).build()返回

ResumableAsyncHandler

org/asynchttpclient/handler/resumable/ResumableAsyncHandler.java

  /*** Invoke this API if you want to set the Range header on your {@link Request} based on the last valid bytes* position.** @param request {@link Request}* @return a {@link Request} with the Range header properly set.*/public Request adjustRequestRange(Request request) {Long ri = resumableIndex.get(request.getUrl());if (ri != null) {byteTransferred.set(ri);}// The Resumableif (resumableListener != null && resumableListener.length() > 0 && byteTransferred.get() != resumableListener.length()) {byteTransferred.set(resumableListener.length());}RequestBuilder builder = new RequestBuilder(request);if (request.getHeaders().get(RANGE) == null && byteTransferred.get() != 0) {builder.setHeader(RANGE, "bytes=" + byteTransferred.get() + "-");}return builder.build();}

adjustRequestRange方法在byteTransferred不等于resumableListener.length()的时候会重置byteTransferred，对于有range的header则重置其bytes参数值

ResponseFiltersInterceptor

org/asynchttpclient/netty/handler/intercept/ResponseFiltersInterceptor.java

  public boolean exitAfterProcessingFilters(Channel channel,NettyResponseFuture<?> future,AsyncHandler<?> handler,HttpResponseStatus status,HttpHeaders responseHeaders) {FilterContext fc = new FilterContext.FilterContextBuilder().asyncHandler(handler).request(future.getCurrentRequest()).responseStatus(status).responseHeaders(responseHeaders).build();for (ResponseFilter asyncFilter : config.getResponseFilters()) {try {fc = asyncFilter.filter(fc);// FIXME Is it worth protecting against this?assertNotNull("fc", "filterContext");} catch (FilterException efe) {requestSender.abort(channel, future, efe);}}// The handler may have been wrapped.future.setAsyncHandler(fc.getAsyncHandler());// The request has changedif (fc.replayRequest()) {requestSender.replayRequest(future, fc, channel);return true;}return false;}

ResponseFiltersInterceptor的exitAfterProcessingFilters方法，对于fc.replayRequest()为true则执行requestSender.replayRequest(future, fc, channel)

replayRequest

org/asynchttpclient/netty/request/NettyRequestSender.java

  public void replayRequest(final NettyResponseFuture<?> future, FilterContext fc, Channel channel) {Request newRequest = fc.getRequest();future.setAsyncHandler(fc.getAsyncHandler());future.setChannelState(ChannelState.NEW);future.touch();LOGGER.debug("\n\nReplaying Request {}\n for Future {}\n", newRequest, future);try {future.getAsyncHandler().onRetry();} catch (Exception e) {LOGGER.error("onRetry crashed", e);abort(channel, future, e);return;}channelManager.drainChannelAndOffer(channel, future);sendNextRequest(newRequest, future);}

NettyRequestSender的replayRequest方法会触发future.getAsyncHandler().onRetry()回调，然后执行channelManager.drainChannelAndOffer(channel, future)及sendNextRequest

applyIoExceptionFiltersAndReplayRequest

org/asynchttpclient/netty/request/NettyRequestSender.java

  public boolean applyIoExceptionFiltersAndReplayRequest(NettyResponseFuture<?> future, IOException e,Channel channel) {boolean replayed = false;@SuppressWarnings({"unchecked", "rawtypes"})FilterContext<?> fc = new FilterContext.FilterContextBuilder().asyncHandler(future.getAsyncHandler()).request(future.getCurrentRequest()).ioException(e).build();for (IOExceptionFilter asyncFilter : config.getIoExceptionFilters()) {try {fc = asyncFilter.filter(fc);assertNotNull(fc, "filterContext");} catch (FilterException efe) {abort(channel, future, efe);}}if (fc.replayRequest() && future.incrementRetryAndCheck() && future.isReplayPossible()) {future.setKeepAlive(false);replayRequest(future, fc, channel);replayed = true;}return replayed;}

NettyRequestSender的applyIoExceptionFiltersAndReplayRequest方法会遍历config.getIoExceptionFilters()，然后挨个执行fc = asyncFilter.filter(fc)

小结

AsyncHttpClient的IOExceptionFilter接口定义了filter方法，在捕获到IOException的时候执行；ResumableIOExceptionFilter实现了IOExceptionFilter接口，它在asyncHandler是ResumableAsyncHandler类型的时候执行adjustRequestRange方法获取新的request，然后再通过new FilterContext.FilterContextBuilder<>(ctx).request(request).replayRequest(true).build()返回；NettyRequestSender的replayRequest方法会触发future.getAsyncHandler().onRetry()回调，然后执行channelManager.drainChannelAndOffer(channel, future)及sendNextRequest。