RealtimeResource { @OnMessage public void publishEvent(ChatAlyticsEvent event) { Set<Session> closedSessions = Sets.newHashSet(); for (Session session : sessions) { if (!session.isOpen()) { closedSessions.add(session); continue; } session.getAsyncRemote().sendObject(event); } sessions.removeAll(closedSessions); } RealtimeResource(); @OnOpen void openSocket(@PathParam(RT_COMPUTE_ENDPOINT_PARAM) ConnectionType type,
                           Session session); @OnMessage void publishEvent(ChatAlyticsEvent event); @OnClose void close(Session session, CloseReason reason); @OnError void onError(Throwable t); int numSessions(); static final String RT_COMPUTE_ENDPOINT; static final String RT_FULL_ENDPOINT; }
@Test public void testPublishEvent() { MessageSummary actualEvent = mock(MessageSummary.class); String eventType = actualEvent.getClass().getSimpleName(); ChatAlyticsEvent event = new ChatAlyticsEvent(DateTime.now(), eventType, actualEvent); Async asyncRemote = mock(Async.class); when(session.getAsyncRemote()).thenReturn(asyncRemote); ConnectionType type = ConnectionType.SUBSCRIBER; underTest.openSocket(type, session); verify(session).getId(); verify(session).setMaxIdleTimeout(0); verifyNoMoreInteractions(session); Session closedSession = mock(Session.class); when(closedSession.getId()).thenReturn("id2"); when(closedSession.isOpen()).thenReturn(false); underTest.openSocket(type, closedSession); verify(closedSession).getId(); verify(closedSession).setMaxIdleTimeout(0); verifyNoMoreInteractions(closedSession); verify(session).isOpen(); verifyNoMoreInteractions(session); assertEquals(2, underTest.numSessions()); underTest.publishEvent(event); verify(session, times(2)).isOpen(); verify(session).getAsyncRemote(); verifyNoMoreInteractions(session); verify(asyncRemote).sendObject(event); verifyNoMoreInteractions(asyncRemote); verify(closedSession).isOpen(); verifyNoMoreInteractions(closedSession); assertEquals(1, underTest.numSessions()); }