BasicWebSocketMessageTransmitter implements WebSocketMessageTransmitter { @Override public synchronized void transmit(String endpointId, String message) { Optional<Session> sessionOptional = registry.get(endpointId); if (!sessionOptional.isPresent()) { sessionOptional = registry.getByPartialMatch(endpointId).stream().findFirst(); } if (!sessionOptional.isPresent() || !sessionOptional.get().isOpen()) { LOG.trace("Session is not registered or closed, adding message to pending"); reSender.add(endpointId, message); } else { LOG.trace("Session registered and open, sending message"); try { sessionOptional.get().getBasicRemote().sendText(message); } catch (IOException e) { LOG.error("Error while trying to send a message to a basic websocket remote endpoint", e); } } } @Inject  BasicWebSocketMessageTransmitter(
      WebSocketSessionRegistry registry, MessagesReSender reSender); @Override synchronized void transmit(String endpointId, String message);  }
@Test public void shouldSendDirectMessageIfSessionIsOpenAndEndpointIsSet() throws IOException { transmitter.transmit(ENDPOINT_ID, MESSAGE); verify(session).getBasicRemote(); verify(remote).sendText(MESSAGE); verify(reSender, never()).add(eq(ENDPOINT_ID), anyString()); }
@Test public void shouldAddMessageToPendingIfSessionIsNotOpenedAndEndpointIsSet() throws IOException { when(session.isOpen()).thenReturn(false); transmitter.transmit(ENDPOINT_ID, MESSAGE); verify(session, never()).getBasicRemote(); verify(remote, never()).sendText(MESSAGE); verify(reSender).add(ENDPOINT_ID, MESSAGE); }