WebSocketAdapter extends CommAdapter<String> implements HttpServerInstanceListener { public WebSocket connectTo (String destination) { if (httpClient==null) { throw new IllegalStateException("Adapter not initialized properly for client mode"); } return clientEndpointFactory.createFor(httpClient, destination, metricsCollector); } private  WebSocketAdapter(Builder builder); WebSocket connectTo(String destination); Observable<WebSocket> acceptConnections(String destination); @Override void httpServerInstanceCreated(HttpServer httpServer); static Builder builder(); Future<HttpServer> shutdown(); Future<HttpServer> shutdown(Duration timeout);  }
@Test void metricsProperlyMaintained() throws Exception { String destinationUri = "ws: CountDownLatch closeLatch = new CountDownLatch(1); Counter totalSubscriptions = metrics.getCounter("websocket", "subscriptions", "total"); Counter specificSubscriptions = metrics.getCounter("websocket", "subscriptions", destinationUri); Meter totalNumberOfReceivedMessages = metrics.getMeter("websocket", "received", "total"); Meter totalNumberOfSentMessages = metrics.getMeter("websocket", "sent", "total"); assertThat(totalSubscriptions.getCount(), is(0L)); assertThat(specificSubscriptions.getCount(), is(0L)); assertThat(totalNumberOfSentMessages.getCount(), is(0L)); WebSocket endpoint = sut.connectTo(destinationUri); endpoint.onClose(socketReasonPair -> closeLatch.countDown() ); assertThat(totalSubscriptions.getCount(), is(1L)); assertThat(specificSubscriptions.getCount(), is(1L)); assertThat(totalNumberOfSentMessages.getCount(), is(0L)); assertThat(totalNumberOfReceivedMessages.getCount(), is(0L)); endpoint.send("Hallo"); assertThat(totalSubscriptions.getCount(), is(1L)); assertThat(specificSubscriptions.getCount(), is(1L)); assertThat(totalNumberOfSentMessages.getCount(), is(1L)); endpoint.close(); closeLatch.await(10, TimeUnit.SECONDS); assertThat(closeLatch.getCount(), is(0L)); assertThat(totalSubscriptions.getCount(), is(0L)); assertThat(specificSubscriptions.getCount(), is(0L)); assertThat(totalNumberOfSentMessages.getCount(), is(1L)); }
@Test void sendAndReceiveAfterInitiatingConnection() throws Exception { String destinationUri = "ws: CountDownLatch closeLatch = new CountDownLatch(1); WebSocket endpoint = sut.connectTo(destinationUri); endpoint.onClose(socketReasonPair -> closeLatch.countDown()); testEchoFromClientPerspective(destinationUri, endpoint); endpoint.close(); closeLatch.await(10, TimeUnit.SECONDS); assertThat(closeLatch.getCount(), is(0L)); }