@Test void shouldPropagateFinalizedCheckpointMessages() throws IOException { when(req.getQueryString()).thenReturn("&topics=finalized_checkpoint"); manager.registerClient(client1); when(chainDataProvider.getStateRootFromBlockRoot(sampleCheckpointEvent.block)) .thenReturn(Optional.of(sampleCheckpointEvent.state)); triggerFinalizedCheckpointEvent(); verify(outputStream).print(stringArgs.capture()); final String eventString = stringArgs.getValue(); assertThat(eventString).contains("event: finalized_checkpoint\n"); final FinalizedCheckpointEvent event = jsonProvider.jsonToObject( eventString.substring(eventString.indexOf("{")), FinalizedCheckpointEvent.class); assertThat(event).isEqualTo(sampleCheckpointEvent); }
public void registerClient(final SseClient sseClient) { LOG.trace("connected " + sseClient.hashCode()); final List<String> allTopicsInContext = ListQueryParameterUtils.getParameterAsStringList(sseClient.ctx.queryParamMap(), TOPICS); sseClient.onClose( () -> { getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).remove(sseClient)); LOG.trace("disconnected " + sseClient.hashCode()); }); getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).add(sseClient)); }
EventSubscriptionManager implements ChainHeadChannel, FinalizedCheckpointChannel { public void registerClient(final SseClient sseClient) { LOG.trace("connected " + sseClient.hashCode()); final List<String> allTopicsInContext = ListQueryParameterUtils.getParameterAsStringList(sseClient.ctx.queryParamMap(), TOPICS); sseClient.onClose( () -> { getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).remove(sseClient)); LOG.trace("disconnected " + sseClient.hashCode()); }); getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).add(sseClient)); } }
EventSubscriptionManager implements ChainHeadChannel, FinalizedCheckpointChannel { public void registerClient(final SseClient sseClient) { LOG.trace("connected " + sseClient.hashCode()); final List<String> allTopicsInContext = ListQueryParameterUtils.getParameterAsStringList(sseClient.ctx.queryParamMap(), TOPICS); sseClient.onClose( () -> { getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).remove(sseClient)); LOG.trace("disconnected " + sseClient.hashCode()); }); getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).add(sseClient)); } EventSubscriptionManager(
      final ChainDataProvider provider,
      final JsonProvider jsonProvider,
      final EventChannels eventChannels); }
EventSubscriptionManager implements ChainHeadChannel, FinalizedCheckpointChannel { public void registerClient(final SseClient sseClient) { LOG.trace("connected " + sseClient.hashCode()); final List<String> allTopicsInContext = ListQueryParameterUtils.getParameterAsStringList(sseClient.ctx.queryParamMap(), TOPICS); sseClient.onClose( () -> { getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).remove(sseClient)); LOG.trace("disconnected " + sseClient.hashCode()); }); getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).add(sseClient)); } EventSubscriptionManager(
      final ChainDataProvider provider,
      final JsonProvider jsonProvider,
      final EventChannels eventChannels); void registerClient(final SseClient sseClient); @Override void chainHeadUpdated(
      final UInt64 slot,
      final Bytes32 stateRoot,
      final Bytes32 bestBlockRoot,
      final boolean epochTransition,
      final Optional<ReorgContext> optionalReorgContext); @Override void onNewFinalizedCheckpoint(final Checkpoint checkpoint); }
EventSubscriptionManager implements ChainHeadChannel, FinalizedCheckpointChannel { public void registerClient(final SseClient sseClient) { LOG.trace("connected " + sseClient.hashCode()); final List<String> allTopicsInContext = ListQueryParameterUtils.getParameterAsStringList(sseClient.ctx.queryParamMap(), TOPICS); sseClient.onClose( () -> { getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).remove(sseClient)); LOG.trace("disconnected " + sseClient.hashCode()); }); getTopics(allTopicsInContext).forEach(e -> eventClientMap.get(e).add(sseClient)); } EventSubscriptionManager(
      final ChainDataProvider provider,
      final JsonProvider jsonProvider,
      final EventChannels eventChannels); void registerClient(final SseClient sseClient); @Override void chainHeadUpdated(
      final UInt64 slot,
      final Bytes32 stateRoot,
      final Bytes32 bestBlockRoot,
      final boolean epochTransition,
      final Optional<ReorgContext> optionalReorgContext); @Override void onNewFinalizedCheckpoint(final Checkpoint checkpoint);  }