@Test public void messageIdShouldBeSet(){ CloudRuntimeEvent cloudRuntimeEvent = mock(CloudRuntimeEventImpl.class); when(cloudRuntimeEvent.getEventType()).thenReturn(ProcessRuntimeEvent.ProcessEvents.PROCESS_CREATED); EventToEntityConverter converter = mock(EventToEntityConverter.class); when(converters.getConverterByEventTypeName(ProcessRuntimeEvent.ProcessEvents.PROCESS_CREATED.name())).thenReturn(converter); AuditEventEntity entity = mock(AuditEventEntity.class); when(converter.convertToEntity(cloudRuntimeEvent)).thenReturn(entity); CloudRuntimeEvent[] events = {cloudRuntimeEvent}; HashMap <String,Object> headers = new HashMap<>(); headers.put("id", UUID.randomUUID()); handler.receiveCloudRuntimeEvent(headers, events); verify((CloudRuntimeEventImpl)cloudRuntimeEvent).setMessageId(headers.get("id").toString()); }
@SuppressWarnings("unchecked") @Override @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER) public void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events) { if (events != null) { AtomicInteger counter = new AtomicInteger(0); for (CloudRuntimeEvent event : events) { EventToEntityConverter converter = eventConverters.getConverterByEventTypeName(event.getEventType().name()); if (converter != null) { ((CloudRuntimeEventImpl)event).setMessageId((headers.get(MessageHeaders.ID).toString())); ((CloudRuntimeEventImpl)event).setSequenceNumber(counter.getAndIncrement()); eventsRepository.save((AuditEventEntity) converter.convertToEntity(event)); } else { LOGGER.warn(">>> Ignoring CloudRuntimeEvents type: " + event.getEventType().name()); } } } }
AuditConsumerChannelHandlerImpl implements AuditConsumerChannelHandler { @SuppressWarnings("unchecked") @Override @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER) public void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events) { if (events != null) { AtomicInteger counter = new AtomicInteger(0); for (CloudRuntimeEvent event : events) { EventToEntityConverter converter = eventConverters.getConverterByEventTypeName(event.getEventType().name()); if (converter != null) { ((CloudRuntimeEventImpl)event).setMessageId((headers.get(MessageHeaders.ID).toString())); ((CloudRuntimeEventImpl)event).setSequenceNumber(counter.getAndIncrement()); eventsRepository.save((AuditEventEntity) converter.convertToEntity(event)); } else { LOGGER.warn(">>> Ignoring CloudRuntimeEvents type: " + event.getEventType().name()); } } } } }
AuditConsumerChannelHandlerImpl implements AuditConsumerChannelHandler { @SuppressWarnings("unchecked") @Override @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER) public void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events) { if (events != null) { AtomicInteger counter = new AtomicInteger(0); for (CloudRuntimeEvent event : events) { EventToEntityConverter converter = eventConverters.getConverterByEventTypeName(event.getEventType().name()); if (converter != null) { ((CloudRuntimeEventImpl)event).setMessageId((headers.get(MessageHeaders.ID).toString())); ((CloudRuntimeEventImpl)event).setSequenceNumber(counter.getAndIncrement()); eventsRepository.save((AuditEventEntity) converter.convertToEntity(event)); } else { LOGGER.warn(">>> Ignoring CloudRuntimeEvents type: " + event.getEventType().name()); } } } } AuditConsumerChannelHandlerImpl(EventsRepository eventsRepository,
                                           APIEventToEntityConverters eventConverters); }
AuditConsumerChannelHandlerImpl implements AuditConsumerChannelHandler { @SuppressWarnings("unchecked") @Override @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER) public void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events) { if (events != null) { AtomicInteger counter = new AtomicInteger(0); for (CloudRuntimeEvent event : events) { EventToEntityConverter converter = eventConverters.getConverterByEventTypeName(event.getEventType().name()); if (converter != null) { ((CloudRuntimeEventImpl)event).setMessageId((headers.get(MessageHeaders.ID).toString())); ((CloudRuntimeEventImpl)event).setSequenceNumber(counter.getAndIncrement()); eventsRepository.save((AuditEventEntity) converter.convertToEntity(event)); } else { LOGGER.warn(">>> Ignoring CloudRuntimeEvents type: " + event.getEventType().name()); } } } } AuditConsumerChannelHandlerImpl(EventsRepository eventsRepository,
                                           APIEventToEntityConverters eventConverters); @SuppressWarnings("unchecked") @Override @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER) void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events); }
AuditConsumerChannelHandlerImpl implements AuditConsumerChannelHandler { @SuppressWarnings("unchecked") @Override @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER) public void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events) { if (events != null) { AtomicInteger counter = new AtomicInteger(0); for (CloudRuntimeEvent event : events) { EventToEntityConverter converter = eventConverters.getConverterByEventTypeName(event.getEventType().name()); if (converter != null) { ((CloudRuntimeEventImpl)event).setMessageId((headers.get(MessageHeaders.ID).toString())); ((CloudRuntimeEventImpl)event).setSequenceNumber(counter.getAndIncrement()); eventsRepository.save((AuditEventEntity) converter.convertToEntity(event)); } else { LOGGER.warn(">>> Ignoring CloudRuntimeEvents type: " + event.getEventType().name()); } } } } AuditConsumerChannelHandlerImpl(EventsRepository eventsRepository,
                                           APIEventToEntityConverters eventConverters); @SuppressWarnings("unchecked") @Override @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER) void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events);  }