Improve fixed for AsyncHttpClient#110. Also fixes AsyncHttpClient#156 by properly invoking the AsyncHandler/WebSocketListener

Author: jfarcand

api/src/test/java/org/asynchttpclient/websocket/CloseCodeReasonMessageTest.java

@@ -102,7 +102,6 @@ public void onError(Throwable t) {
  }
  }
 
-
  @Test(timeOut = 60000)
  public void wrongStatusCode() throws Throwable {
  AsyncHttpClient c = getAsyncHttpClient(null);
@@ -142,4 +141,44 @@ public void onError(Throwable t) {
  c.close();
  }
  }
+
+ @Test(timeOut = 60000)
+ public void wrongProtocolCode() throws Throwable {
+ AsyncHttpClient c = getAsyncHttpClient(null);
+ try {
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicReference<Throwable> throwable = new AtomicReference<Throwable>();
+
+ WebSocket websocket = c.prepareGet("ws://www.google.com").execute(new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WebSocketTextListener() {
+
+ @Override
+ public void onMessage(String message) {
+ }
+
+ @Override
+ public void onFragment(String fragment, boolean last) {
+ }
+
+ @Override
+ public void onOpen(WebSocket websocket) {
+ }
+
+ @Override
+ public void onClose(WebSocket websocket) {
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ throwable.set(t);
+ latch.countDown();
+ }
+ }).build()).get();
+
+ latch.await();
+ assertNotNull(throwable.get());
+ assertEquals(throwable.get().getClass(), IllegalStateException.class);
+ } finally {
+ c.close();
+ }
+ }
 }

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/EventHandler.java

@@ -42,7 +42,6 @@
 import org.glassfish.grizzly.http.util.Header;
 import org.glassfish.grizzly.http.util.HttpStatus;
 import org.glassfish.grizzly.utils.IdleTimeoutFilter;
-import org.glassfish.grizzly.websockets.HandshakeException;
 import org.glassfish.grizzly.websockets.SimpleWebSocket;
 import org.glassfish.grizzly.websockets.WebSocketHolder;
 
@@ -201,7 +200,12 @@ public void onInitialLineParsed(HttpHeader httpHeader,
  context.setCurrentState(handler.onStatusReceived(responseStatus));
  if (context.isWSRequest() && context.getCurrentState() == ABORT) {
  httpHeader.setSkipRemainder(true);
- context.abort(new HandshakeException("Upgrade failed"));
+ try {
+ context.result(handler.onCompleted());
+ context.done();
+ } catch (Exception e) {
+ context.abort(e);
+ }
  }
  }
  } catch (Exception e) {

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/WebSocketProtocol.java

@@ -15,7 +15,6 @@
  */
 package org.asynchttpclient.providers.netty.handler;
 
-import static io.netty.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpHeaders;
@@ -24,10 +23,6 @@
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
-
-import java.io.IOException;
-import java.util.Locale;
-
 import org.asynchttpclient.AsyncHandler.STATE;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.HttpResponseHeaders;
@@ -48,6 +43,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.util.Locale;
+
+import static io.netty.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS;
+
 final class WebSocketProtocol extends Protocol {
 
  private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketProtocol.class);
@@ -98,16 +98,25 @@ public void handle(ChannelHandlerContext ctx, NettyResponseFuture<?> future, Obj
  status = new ResponseStatus(future.getURI(), response, config);
  final boolean statusReceived = h.onStatusReceived(status) == STATE.UPGRADE;
 
+ if (!statusReceived) {
+ try {
+ h.onCompleted();
+ } finally {
+ future.done();
+ }
+ return;
+ }
+
  final boolean headerOK = h.onHeadersReceived(responseHeaders) == STATE.CONTINUE;
- if (!headerOK || !validStatus || !validUpgrade || !validConnection || !statusReceived) {
+ if (!headerOK || !validStatus || !validUpgrade || !validConnection) {
  channels.abort(future, new IOException("Invalid handshake response"));
  return;
  }
 
  String accept = response.headers().get(HttpHeaders.Names.SEC_WEBSOCKET_ACCEPT);
  String key = WebSocketUtil.getAcceptKey(future.getNettyRequest().getHttpRequest().headers().get(HttpHeaders.Names.SEC_WEBSOCKET_KEY));
  if (accept == null || !accept.equals(key)) {
- throw new IOException(String.format("Invalid challenge. Actual: %s. Expected: %s", accept, key));
+ channels.abort(future, new IOException(String.format("Invalid challenge. Actual: %s. Expected: %s", accept, key)));
  }
 
  Channels.upgradePipelineForWebSockets(ctx);