WebSocketNativeBalancingHandler extends WebSocketHandlerAdapter { public void setNextHandler(WebSocketHandler handler) { this.nextHandler = handler; handler.setListener(new WebSocketHandlerListener() { @Override public void connectionOpened(WebSocketChannel channel, String protocol) { if (!WebSocketHandshakeObject.KAAZING_EXTENDED_HANDSHAKE.equals(protocol)) { WebSocketNativeChannel wsChannel = (WebSocketNativeChannel)channel; wsChannel.balanced.set(2); listener.connectionOpened(channel, protocol); } } @Override public void redirected(WebSocketChannel channel, String location) { try { LOG.finest("Balancer redirect location = " + StringUtils.stripControlCharacters(location)); WSURI uri = new WSURI(location); reconnect(channel, uri, channel.getProtocol()); nextHandler.processClose(channel, 0, null); } catch (URISyntaxException e) { LOG.log(Level.WARNING, e.getMessage(), e); listener.connectionFailed(channel, e); } catch (Exception e) { LOG.log(Level.WARNING, e.getMessage(), e); listener.connectionFailed(channel, e); } } @Override public void authenticationRequested(WebSocketChannel channel, String location, String challenge) { listener.authenticationRequested(channel, location, challenge); } @Override public void binaryMessageReceived(WebSocketChannel channel, WrappedByteBuffer buf) { handleBinaryMessageReceived(channel, buf); } @Override public void textMessageReceived(WebSocketChannel channel, String message) { handleTextMessageReceived(channel, message); } @Override public void connectionClosed(WebSocketChannel channel, boolean wasClean, int code, String reason) { WebSocketNativeChannel wsChannel = (WebSocketNativeChannel)channel; if (wsChannel.reconnecting.compareAndSet(true, false)) { wsChannel.reconnected.set(true); String[] nextProtocols; String[] requestedProtocols = wsChannel.getRequestedProtocols(); if (requestedProtocols == null || requestedProtocols.length == 0) { nextProtocols = new String[] { WebSocketHandshakeObject.KAAZING_EXTENDED_HANDSHAKE }; } else { nextProtocols = new String[requestedProtocols.length+1]; nextProtocols[0] = WebSocketHandshakeObject.KAAZING_EXTENDED_HANDSHAKE; for (int i=0; i<requestedProtocols.length; i++) { nextProtocols[i+1] = requestedProtocols[i]; } } processConnect(channel, wsChannel.redirectUri, nextProtocols); } else { listener.connectionClosed(channel, wasClean, code, reason); } } @Override public void connectionClosed(WebSocketChannel channel, Exception ex) { listener.connectionClosed(channel, ex); } @Override public void connectionFailed(WebSocketChannel channel, Exception ex) { if (ex == null) { listener.connectionClosed(channel, false, 0, null); } else { listener.connectionClosed(channel, ex); } } @Override public void commandMessageReceived(WebSocketChannel channel, CommandMessage message) { listener.commandMessageReceived(channel, message); } }); }  @Override void processConnect(WebSocketChannel channel, WSURI uri, String[] protocols); void setNextHandler(WebSocketHandler handler);  }
@Test public void testProcessAuthorize() throws URISyntaxException { Mockery context = new Mockery(); final WebSocketHandler nextHandler = context.mock(WebSocketHandler.class); final WebSocketHandlerListener listener = context.mock(WebSocketHandlerListener.class); context.checking(new Expectations() { { oneOf(nextHandler).setListener(with(aNonNull(WebSocketHandlerListener.class))); will(saveParameter("listener", 0)); oneOf(nextHandler).processAuthorize(with(aNonNull(WebSocketChannel.class)), with("Application Basic relam")); will(new CustomAction("will fire 401 challenge") { @Override public Object invoke(Invocation invocation) { WebSocketHandlerListener listener = (WebSocketHandlerListener)lookup("listener"); WebSocketChannel channel = (WebSocketChannel)invocation.getParameter(0); listener.authenticationRequested(channel, "location", (String)invocation.getParameter(1)); return null; } }); oneOf(listener).authenticationRequested(with(aNonNull(WebSocketChannel.class)), with("location"), with("Application Basic relam")); } }); WSURI uri = new WSURI("ws: WebSocketChannel channel = new WebSocketChannel(uri); WebSocketNativeBalancingHandler handler = new WebSocketNativeBalancingHandler(); handler.setNextHandler(nextHandler); handler.setListener(listener); handler.processAuthorize(channel, "Application Basic relam"); context.assertIsSatisfied(); }
@Test public void testProcessBinaryMessage() throws URISyntaxException { Mockery context = new Mockery(); final WebSocketHandler nextHandler = context.mock(WebSocketHandler.class); final WebSocketHandlerListener listener = context.mock(WebSocketHandlerListener.class); final WrappedByteBuffer message = new WrappedByteBuffer("test message".getBytes()); context.checking(new Expectations() { { oneOf(nextHandler).setListener(with(aNonNull(WebSocketHandlerListener.class))); will(saveParameter("listener", 0)); oneOf(nextHandler).processBinaryMessage(with(aNonNull(WebSocketChannel.class)), with(message)); will(new CustomAction("will fire text message received") { @Override public Object invoke(Invocation invocation) { WebSocketHandlerListener listener = (WebSocketHandlerListener)lookup("listener"); WebSocketChannel channel = (WebSocketChannel)invocation.getParameter(0); listener.binaryMessageReceived(channel, message); return null; } }); oneOf(listener).binaryMessageReceived(with(aNonNull(WebSocketChannel.class)), with(message)); } }); WSURI uri = new WSURI("ws: WebSocketChannel channel = getNativeChannel(uri); WebSocketNativeBalancingHandler handler = new WebSocketNativeBalancingHandler(); handler.setNextHandler(nextHandler); handler.setListener(listener); handler.processBinaryMessage(channel, message); context.assertIsSatisfied(); }
@Test public void testProcessTextMessage() throws URISyntaxException { Mockery context = new Mockery(); final WebSocketHandler nextHandler = context.mock(WebSocketHandler.class); final WebSocketHandlerListener listener = context.mock(WebSocketHandlerListener.class); final WrappedByteBuffer message = new WrappedByteBuffer("test message".getBytes()); context.checking(new Expectations() { { oneOf(nextHandler).setListener(with(aNonNull(WebSocketHandlerListener.class))); will(saveParameter("listener", 0)); oneOf(nextHandler).processTextMessage(with(aNonNull(WebSocketChannel.class)), with("test message")); will(new CustomAction("will fire message received") { @Override public Object invoke(Invocation invocation) { WebSocketHandlerListener listener = (WebSocketHandlerListener)lookup("listener"); WebSocketChannel channel = (WebSocketChannel)invocation.getParameter(0); listener.binaryMessageReceived(channel, message); return null; } }); oneOf(listener).binaryMessageReceived(with(aNonNull(WebSocketChannel.class)), with(aNonNull(WrappedByteBuffer.class))); } }); WSURI uri = new WSURI("ws: WebSocketChannel channel = getNativeChannel(uri); WebSocketNativeBalancingHandler handler = new WebSocketNativeBalancingHandler(); handler.setNextHandler(nextHandler); handler.setListener(listener); handler.processTextMessage(channel, "test message"); context.assertIsSatisfied(); }
@Test public void testProcessBalancerMessage() throws URISyntaxException, UnsupportedEncodingException { Mockery context = new Mockery(); final WebSocketHandler nextHandler = context.mock(WebSocketHandler.class); final WebSocketHandlerListener listener = context.mock(WebSocketHandlerListener.class); final String balancerNoMessage = '\uf0ff' + "N"; context.checking(new Expectations() { { oneOf(nextHandler).setListener(with(aNonNull(WebSocketHandlerListener.class))); will(saveParameter("listener", 0)); oneOf(nextHandler).processTextMessage(with(aNonNull(WebSocketChannel.class)), with("test message")); will(new CustomAction("will balancer message") { @Override public Object invoke(Invocation invocation) throws UnsupportedEncodingException { WebSocketHandlerListener listener = (WebSocketHandlerListener)lookup("listener"); WebSocketChannel channel = (WebSocketChannel)invocation.getParameter(0); listener.textMessageReceived(channel, balancerNoMessage); return null; } }); oneOf(listener).connectionOpened(with(aNonNull(WebSocketChannel.class)), with("x-kaazing-handshake")); will(new CustomAction("will fire balancer message again") { @Override public Object invoke(Invocation invocation) throws UnsupportedEncodingException { WebSocketHandlerListener listener = (WebSocketHandlerListener)lookup("listener"); WebSocketChannel channel = (WebSocketChannel)invocation.getParameter(0); listener.textMessageReceived(channel, balancerNoMessage); return null; } }); oneOf(listener).connectionOpened(with(aNonNull(WebSocketChannel.class)), with("")); will(new CustomAction("will fire 3nd balancer message") { @Override public Object invoke(Invocation invocation) throws UnsupportedEncodingException { WebSocketHandlerListener listener = (WebSocketHandlerListener)lookup("listener"); WebSocketChannel channel = (WebSocketChannel)invocation.getParameter(0); listener.textMessageReceived(channel, balancerNoMessage); return null; } }); oneOf(listener).textMessageReceived(with(aNonNull(WebSocketChannel.class)), with(aNonNull(String.class))); } }); WSURI uri = new WSURI("ws: WebSocketChannel channel = getNativeChannel(uri); WebSocketNativeBalancingHandler handler = new WebSocketNativeBalancingHandler(); handler.setNextHandler(nextHandler); handler.setListener(listener); handler.processTextMessage(channel, "test message"); context.assertIsSatisfied(); }
@Test public void testProcessBalancerRedirectMessage() throws URISyntaxException, UnsupportedEncodingException { Mockery context = new Mockery(); final WebSocketHandler nextHandler = context.mock(WebSocketHandler.class); final WebSocketHandlerListener listener = context.mock(WebSocketHandlerListener.class); final String balancerRedirectMessage = '\uf0ff' + "Rws: context.checking(new Expectations() { { oneOf(nextHandler).setListener(with(aNonNull(WebSocketHandlerListener.class))); will(saveParameter("listener", 0)); oneOf(nextHandler).processTextMessage(with(aNonNull(WebSocketChannel.class)), with("test message")); will(new CustomAction("will fire redirect message") { @Override public Object invoke(Invocation invocation) { WebSocketHandlerListener listener = (WebSocketHandlerListener)lookup("listener"); WebSocketChannel channel = (WebSocketChannel)invocation.getParameter(0); listener.textMessageReceived(channel, balancerRedirectMessage); return null; } }); oneOf(nextHandler).processClose(with(aNonNull(WebSocketChannel.class)), with(0), with(aNull(String.class))); } }); WSURI uri = new WSURI("ws: WebSocketChannel channel = getNativeChannel(uri); WebSocketNativeBalancingHandler handler = new WebSocketNativeBalancingHandler(); handler.setNextHandler(nextHandler); handler.setListener(listener); handler.processTextMessage(channel, "test message"); context.assertIsSatisfied(); }