@Test public void streamIdFor() { testSubject.register("clientStreamId", "clientId", ClientIdRegistry.ConnectionType.COMMAND); assertEquals("clientStreamId", testSubject.streamIdFor("clientId", ClientIdRegistry.ConnectionType.COMMAND)); try { testSubject.streamIdFor("clientId", ClientIdRegistry.ConnectionType.QUERY); fail("should not find query stream"); } catch (IllegalStateException illegalStateException) { } }
@Override public boolean register(String clientStreamId, String clientId, ConnectionType type) { Map<String, String> connectionTypeMap = clientIdMapPerType.computeIfAbsent(type, t -> new ConcurrentHashMap<>()); String prev = connectionTypeMap.put(clientStreamId, clientId); if (logger.isInfoEnabled()) { Set<String> streamSet = connectionTypeMap.entrySet() .stream() .filter(e -> e.getValue().equals(clientId)) .map(Map.Entry::getKey) .collect(Collectors.toSet()); if (streamSet.size() != 1) { logger.info("Multiple mappings for {} stream for clientId {}: {}", type, clientId, streamSet); } } return prev == null; }
DefaultClientIdRegistry implements ClientIdRegistry { @Override public boolean register(String clientStreamId, String clientId, ConnectionType type) { Map<String, String> connectionTypeMap = clientIdMapPerType.computeIfAbsent(type, t -> new ConcurrentHashMap<>()); String prev = connectionTypeMap.put(clientStreamId, clientId); if (logger.isInfoEnabled()) { Set<String> streamSet = connectionTypeMap.entrySet() .stream() .filter(e -> e.getValue().equals(clientId)) .map(Map.Entry::getKey) .collect(Collectors.toSet()); if (streamSet.size() != 1) { logger.info("Multiple mappings for {} stream for clientId {}: {}", type, clientId, streamSet); } } return prev == null; } }
DefaultClientIdRegistry implements ClientIdRegistry { @Override public boolean register(String clientStreamId, String clientId, ConnectionType type) { Map<String, String> connectionTypeMap = clientIdMapPerType.computeIfAbsent(type, t -> new ConcurrentHashMap<>()); String prev = connectionTypeMap.put(clientStreamId, clientId); if (logger.isInfoEnabled()) { Set<String> streamSet = connectionTypeMap.entrySet() .stream() .filter(e -> e.getValue().equals(clientId)) .map(Map.Entry::getKey) .collect(Collectors.toSet()); if (streamSet.size() != 1) { logger.info("Multiple mappings for {} stream for clientId {}: {}", type, clientId, streamSet); } } return prev == null; }  }
DefaultClientIdRegistry implements ClientIdRegistry { @Override public boolean register(String clientStreamId, String clientId, ConnectionType type) { Map<String, String> connectionTypeMap = clientIdMapPerType.computeIfAbsent(type, t -> new ConcurrentHashMap<>()); String prev = connectionTypeMap.put(clientStreamId, clientId); if (logger.isInfoEnabled()) { Set<String> streamSet = connectionTypeMap.entrySet() .stream() .filter(e -> e.getValue().equals(clientId)) .map(Map.Entry::getKey) .collect(Collectors.toSet()); if (streamSet.size() != 1) { logger.info("Multiple mappings for {} stream for clientId {}: {}", type, clientId, streamSet); } } return prev == null; }  @Override boolean register(String clientStreamId, String clientId, ConnectionType type); @Override boolean unregister(String clientStreamId, ConnectionType type); @Override String clientId(String clientStreamId); @Override Set<String> streamIdsFor(String clientId, ConnectionType type); @Override void printOn(Media media); @EventListener void on(ApplicationConnected event); @EventListener void on(ApplicationDisconnected event); @EventListener void on(SubscribeCommand event); @EventListener void on(CommandHandlerDisconnected event); @EventListener void on(SubscribeQuery event); @EventListener void on(QueryHandlerDisconnected event); }
DefaultClientIdRegistry implements ClientIdRegistry { @Override public boolean register(String clientStreamId, String clientId, ConnectionType type) { Map<String, String> connectionTypeMap = clientIdMapPerType.computeIfAbsent(type, t -> new ConcurrentHashMap<>()); String prev = connectionTypeMap.put(clientStreamId, clientId); if (logger.isInfoEnabled()) { Set<String> streamSet = connectionTypeMap.entrySet() .stream() .filter(e -> e.getValue().equals(clientId)) .map(Map.Entry::getKey) .collect(Collectors.toSet()); if (streamSet.size() != 1) { logger.info("Multiple mappings for {} stream for clientId {}: {}", type, clientId, streamSet); } } return prev == null; }  @Override boolean register(String clientStreamId, String clientId, ConnectionType type); @Override boolean unregister(String clientStreamId, ConnectionType type); @Override String clientId(String clientStreamId); @Override Set<String> streamIdsFor(String clientId, ConnectionType type); @Override void printOn(Media media); @EventListener void on(ApplicationConnected event); @EventListener void on(ApplicationDisconnected event); @EventListener void on(SubscribeCommand event); @EventListener void on(CommandHandlerDisconnected event); @EventListener void on(SubscribeQuery event); @EventListener void on(QueryHandlerDisconnected event);  }