@Test public void onEventProcessorStatusChange() { EventProcessorInfo processorInfo = EventProcessorInfo.newBuilder() .setActiveThreads(10) .setAvailableThreads(20) .setProcessorName("Name") .build(); ClientEventProcessorInfo clientEventProcessorInfo = new ClientEventProcessorInfo("client", "client", "context", processorInfo); EventProcessorStatusUpdate event = new EventProcessorStatusUpdate(clientEventProcessorInfo); EventProcessorEvents.EventProcessorStatusUpdated updatedEvent = testSubject .onEventProcessorStatusChange(event); assertEquals("client", updatedEvent.eventProcessorStatus().getClientId()); assertEquals("context", updatedEvent.eventProcessorStatus().getContext()); ClientProcessor clientProcessor = StreamSupport.stream(Spliterators.spliterator(testSubject.iterator(), 100, 0), false). filter(cp -> cp.clientId().equals("client")) .findFirst().orElse(null); assertNotNull(clientProcessor); }
@EventListener public EventProcessorStatusUpdated onEventProcessorStatusChange(EventProcessorStatusUpdate event) { ClientEventProcessorInfo processorStatus = event.eventProcessorStatus(); String clientId = processorStatus.getClientId(); String clientStreamId = processorStatus.getClientStreamId(); Map<String, ClientProcessor> clientData = cache.computeIfAbsent(clientStreamId, c -> new HashMap<>()); EventProcessorInfo eventProcessorInfo = processorStatus.getEventProcessorInfo(); ClientProcessor clientProcessor = mapping.map(clientId, clients.get(clientStreamId), processorStatus.getContext(), eventProcessorInfo); clientData.put(eventProcessorInfo.getProcessorName(), clientProcessor); return new EventProcessorStatusUpdated(processorStatus, false); }
ProcessorsInfoTarget implements ClientProcessors { @EventListener public EventProcessorStatusUpdated onEventProcessorStatusChange(EventProcessorStatusUpdate event) { ClientEventProcessorInfo processorStatus = event.eventProcessorStatus(); String clientId = processorStatus.getClientId(); String clientStreamId = processorStatus.getClientStreamId(); Map<String, ClientProcessor> clientData = cache.computeIfAbsent(clientStreamId, c -> new HashMap<>()); EventProcessorInfo eventProcessorInfo = processorStatus.getEventProcessorInfo(); ClientProcessor clientProcessor = mapping.map(clientId, clients.get(clientStreamId), processorStatus.getContext(), eventProcessorInfo); clientData.put(eventProcessorInfo.getProcessorName(), clientProcessor); return new EventProcessorStatusUpdated(processorStatus, false); } }
ProcessorsInfoTarget implements ClientProcessors { @EventListener public EventProcessorStatusUpdated onEventProcessorStatusChange(EventProcessorStatusUpdate event) { ClientEventProcessorInfo processorStatus = event.eventProcessorStatus(); String clientId = processorStatus.getClientId(); String clientStreamId = processorStatus.getClientStreamId(); Map<String, ClientProcessor> clientData = cache.computeIfAbsent(clientStreamId, c -> new HashMap<>()); EventProcessorInfo eventProcessorInfo = processorStatus.getEventProcessorInfo(); ClientProcessor clientProcessor = mapping.map(clientId, clients.get(clientStreamId), processorStatus.getContext(), eventProcessorInfo); clientData.put(eventProcessorInfo.getProcessorName(), clientProcessor); return new EventProcessorStatusUpdated(processorStatus, false); } ProcessorsInfoTarget(); }
ProcessorsInfoTarget implements ClientProcessors { @EventListener public EventProcessorStatusUpdated onEventProcessorStatusChange(EventProcessorStatusUpdate event) { ClientEventProcessorInfo processorStatus = event.eventProcessorStatus(); String clientId = processorStatus.getClientId(); String clientStreamId = processorStatus.getClientStreamId(); Map<String, ClientProcessor> clientData = cache.computeIfAbsent(clientStreamId, c -> new HashMap<>()); EventProcessorInfo eventProcessorInfo = processorStatus.getEventProcessorInfo(); ClientProcessor clientProcessor = mapping.map(clientId, clients.get(clientStreamId), processorStatus.getContext(), eventProcessorInfo); clientData.put(eventProcessorInfo.getProcessorName(), clientProcessor); return new EventProcessorStatusUpdated(processorStatus, false); } ProcessorsInfoTarget(); @EventListener EventProcessorStatusUpdated onEventProcessorStatusChange(EventProcessorStatusUpdate event); @EventListener void onClientConnected(TopologyEvents.ApplicationConnected event); @EventListener void onClientDisconnected(TopologyEvents.ApplicationDisconnected event); @Nonnull @Override Iterator<ClientProcessor> iterator(); }
ProcessorsInfoTarget implements ClientProcessors { @EventListener public EventProcessorStatusUpdated onEventProcessorStatusChange(EventProcessorStatusUpdate event) { ClientEventProcessorInfo processorStatus = event.eventProcessorStatus(); String clientId = processorStatus.getClientId(); String clientStreamId = processorStatus.getClientStreamId(); Map<String, ClientProcessor> clientData = cache.computeIfAbsent(clientStreamId, c -> new HashMap<>()); EventProcessorInfo eventProcessorInfo = processorStatus.getEventProcessorInfo(); ClientProcessor clientProcessor = mapping.map(clientId, clients.get(clientStreamId), processorStatus.getContext(), eventProcessorInfo); clientData.put(eventProcessorInfo.getProcessorName(), clientProcessor); return new EventProcessorStatusUpdated(processorStatus, false); } ProcessorsInfoTarget(); @EventListener EventProcessorStatusUpdated onEventProcessorStatusChange(EventProcessorStatusUpdate event); @EventListener void onClientConnected(TopologyEvents.ApplicationConnected event); @EventListener void onClientDisconnected(TopologyEvents.ApplicationDisconnected event); @Nonnull @Override Iterator<ClientProcessor> iterator();  }