@Test public void testProcess() throws Exception { Exchange exchange = mock(Exchange.class); Exchange exchange1 = mock(Exchange.class); Exchange exchange2 = mock(Exchange.class); Exchange exchange3 = mock(Exchange.class); Message message1 = buildMessage(INTYGS_ID, HandelsekodEnum.ANDRAT, LocalDateTime.now()); Message message2 = buildMessage(INTYGS_ID, HandelsekodEnum.SIGNAT, LocalDateTime.now()); Message message3 = buildMessage("anotherintyg", HandelsekodEnum.ANDRAT, LocalDateTime.now()); when(exchange1.getIn()).thenReturn(message1); when(exchange2.getIn()).thenReturn(message2); when(exchange3.getIn()).thenReturn(message3); when(exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class)).thenReturn(Arrays.asList(exchange1, exchange2, exchange3)); List<Message> res = aggregator.process(exchange); assertNotNull(res); assertEquals(1, res.size()); }
public List<Message> process(Exchange exchange) { List<Exchange> grouped = exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class); if (grouped == null || grouped.isEmpty()) { LOG.info("No aggregated log messages, this is normal if camel aggregator has a batch timeout. Doing nothing."); return Collections.emptyList(); } List<Message> aggregatedList = grouped.stream() .map(Exchange::getIn) .collect(Collectors.toList()); return new NotificationMessageDiscardFilter().process(aggregatedList); }
NotificationAggregator { public List<Message> process(Exchange exchange) { List<Exchange> grouped = exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class); if (grouped == null || grouped.isEmpty()) { LOG.info("No aggregated log messages, this is normal if camel aggregator has a batch timeout. Doing nothing."); return Collections.emptyList(); } List<Message> aggregatedList = grouped.stream() .map(Exchange::getIn) .collect(Collectors.toList()); return new NotificationMessageDiscardFilter().process(aggregatedList); } }
NotificationAggregator { public List<Message> process(Exchange exchange) { List<Exchange> grouped = exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class); if (grouped == null || grouped.isEmpty()) { LOG.info("No aggregated log messages, this is normal if camel aggregator has a batch timeout. Doing nothing."); return Collections.emptyList(); } List<Message> aggregatedList = grouped.stream() .map(Exchange::getIn) .collect(Collectors.toList()); return new NotificationMessageDiscardFilter().process(aggregatedList); }  }
NotificationAggregator { public List<Message> process(Exchange exchange) { List<Exchange> grouped = exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class); if (grouped == null || grouped.isEmpty()) { LOG.info("No aggregated log messages, this is normal if camel aggregator has a batch timeout. Doing nothing."); return Collections.emptyList(); } List<Message> aggregatedList = grouped.stream() .map(Exchange::getIn) .collect(Collectors.toList()); return new NotificationMessageDiscardFilter().process(aggregatedList); }  List<Message> process(Exchange exchange); }
NotificationAggregator { public List<Message> process(Exchange exchange) { List<Exchange> grouped = exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class); if (grouped == null || grouped.isEmpty()) { LOG.info("No aggregated log messages, this is normal if camel aggregator has a batch timeout. Doing nothing."); return Collections.emptyList(); } List<Message> aggregatedList = grouped.stream() .map(Exchange::getIn) .collect(Collectors.toList()); return new NotificationMessageDiscardFilter().process(aggregatedList); }  List<Message> process(Exchange exchange);  }