OfframpWebSocketHandler extends AbstractWebSocketHandler { @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { sessionId = session.getId(); Map<String, Object> attributes = session.getAttributes(); version = (String) attributes.get(VERSION); roadName = (String) attributes.get(ROAD_NAME); streamName = (String) attributes.get(STREAM_NAME); DefaultOffset defaultOffset = (DefaultOffset) attributes.get(DEFAULT_OFFSET); @SuppressWarnings("unchecked") Set<Sensitivity> grants = (Set<Sensitivity>) attributes.get(GRANTS); Authentication authentication = (Authentication) session.getPrincipal(); RoadConsumer consumer; metrics = metricsFactory.create(roadName, streamName); try { authorisation.checkAuthorisation(authentication, roadName, grants); consumer = consumerFactory.create(roadName, streamName, defaultOffset); MessageFunction messageFunction = messageFunctionFactory.create(roadName, grants); EventSender sender = bytes -> sendEvent(session, bytes); service = serviceFactory.create(version, consumer, messageFunction, sender, metrics); } catch (UnknownRoadException e) { metrics.markRoadNotFound(); session.close(); throw e; } Scheduler scheduler = Schedulers.newSingle( String.format("offramp[v%s,%s.%s:%s]", version, roadName, streamName, sessionId)); metrics.incrementActiveConnections(); Mono<Void> mono = Mono .fromRunnable(service) .subscribeOn(scheduler) .doOnError(t -> true, t -> { log.error("Road: {}, stream: {}, sessionId: {} - Error in OfframpService", roadName, streamName, sessionId, t); }) .doOnSuccess(s -> { log.info("Road: {}, stream: {}, sessionId: {} - OfframpService Completed", roadName, streamName, sessionId); }) .doOnTerminate(() -> { disposables.dispose(); metrics.decrementActiveConnections(); }).then(); disposables.add(() -> close(service)); disposables.add(() -> close(consumer)); disposables.add(() -> close(() -> session.close(SERVER_ERROR))); disposables.add(scheduler); disposables.add(mono.subscribe()); log.info("Road: {}, stream: {}, sessionId: {} - Connection established with defaultOffset: {}", roadName, streamName, sessionId, defaultOffset); metrics.markConnectionEstablished(); }  @Override void afterConnectionEstablished(WebSocketSession session); @Override void afterConnectionClosed(WebSocketSession session, CloseStatus status); @Override void handleTransportError(WebSocketSession session, Throwable exception);  }
@Test public void afterConnectionEstablished() throws Exception { underTest.afterConnectionEstablished(session); verify(metrics).markConnectionEstablished(); assertThat(underTest.getService(), is(service)); assertThat(underTest.getMetrics(), is(metrics)); }