TestConnectListener connectListener = new TestConnectListener(log);
TestSendRequestListener requestListener = new TestSendRequestListener(log);
TestCloseListener closeListener = new TestCloseListener(log);
responseHandler.setConnectionListener(connectListener);
Channel channel = createClientBootstrap(responseHandler);
SocketAddress clientAddr = channel.getLocalAddress();
try
{
setListeners(responseHandler,respProcessor,requestListener,closeListener);
channel.write(new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/test"));
//It seems that there is a race condition between the writeFuture succeeding
//and the writeComplete message getting to the handler. Make sure that the
//writeComplete has got to the handler before we do anything else with
//the channel.
final GenericHttpResponseHandler handler = getResponseHandler(channel);
TestUtil.assertWithBackoff(new ConditionCheck()
{
@Override
public boolean check()
{
return handler._messageState.hasSentRequest();
}
}, "request sent", 1000, log);
HttpResponse resp = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
resp.setHeader(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);
sendServerResponse(clientAddr, resp, 1000);
HttpChunk chunk1 = new DefaultHttpChunk(ChannelBuffers.wrappedBuffer("chunk1".getBytes(Charset.defaultCharset())));
sendServerResponse(clientAddr, chunk1, 1000);
final List<String> callbacks = respProcessor.getCallbacks();
TestUtil.assertWithBackoff(new ConditionCheck()
{
@Override
public boolean check()
{
return 2 == callbacks.size();
}
}, "waiting for response processed", 1000, null);
Assert.assertEquals(callbacks.get(0), "startResponse");
Assert.assertEquals(callbacks.get(1), "addChunk");
Channels.fireExceptionCaught(channel, new ReadTimeoutException());
channel.close();
final List<String> closeCallbacks = closeListener.getCallbacks();
TestUtil.assertWithBackoff(new ConditionCheck()
{
@Override
public boolean check()
{
return 3 == callbacks.size() && 1==closeCallbacks.size();
}
}, "waiting for response processed", 1000, null);
final List<String> connectCallbacks = connectListener.getCallbacks();
final List<String> requestCallbacks = requestListener.getCallbacks();
Assert.assertEquals(callbacks.get(0), "startResponse");
Assert.assertEquals(callbacks.get(1), "addChunk");
Assert.assertTrue(callbacks.get(2).startsWith("channelException"));
//Assert.assertEquals(callbacks.get(3), "channelClosed"); // no more channelClosed after channel Exception
//make sure that no new callbacks have showed up
stateSanityCheck(connectCallbacks,requestCallbacks,callbacks,closeCallbacks);
Assert.assertEquals(callbacks.size(), 3);
}
finally
{
channel.close();
log.info("end");
}
}