@Test void test_onMessageIngested_MultipleMessageMonitors() { MessageMonitor<Message<?>> messageMonitorMock1 = mock(MessageMonitor.class); MessageMonitor.MonitorCallback callback1 = mock(MessageMonitor.MonitorCallback.class); MessageMonitor<Message<?>> messageMonitorMock2 = mock(MessageMonitor.class); MessageMonitor.MonitorCallback callback2 = mock(MessageMonitor.MonitorCallback.class); MultiMessageMonitor multiMessageMonitor = new MultiMessageMonitor(Arrays.asList(messageMonitorMock1, messageMonitorMock2)); Message messageMock = mock(Message.class); Mockito.when(messageMonitorMock1.onMessageIngested(messageMock)).thenReturn(callback1); Mockito.when(messageMonitorMock2.onMessageIngested(messageMock)).thenReturn(callback2); multiMessageMonitor.onMessageIngested(messageMock).reportSuccess(); Mockito.verify(messageMonitorMock1).onMessageIngested(ArgumentMatchers.same(messageMock)); Mockito.verify(callback1).reportSuccess(); Mockito.verify(messageMonitorMock2).onMessageIngested(ArgumentMatchers.same(messageMock)); Mockito.verify(callback2).reportSuccess(); }
@Override public MonitorCallback onMessageIngested(T message) { final List<MonitorCallback> monitorCallbacks = messageMonitors.stream() .map(messageMonitor -> messageMonitor.onMessageIngested(message)) .collect(Collectors.toList()); return new MonitorCallback() { @Override public void reportSuccess() { monitorCallbacks.forEach(MonitorCallback::reportSuccess); } @Override public void reportFailure(Throwable cause) { monitorCallbacks.forEach(resultCallback -> resultCallback.reportFailure(cause)); } @Override public void reportIgnored() { monitorCallbacks.forEach(MonitorCallback::reportIgnored); } }; }
MultiMessageMonitor implements MessageMonitor<T> { @Override public MonitorCallback onMessageIngested(T message) { final List<MonitorCallback> monitorCallbacks = messageMonitors.stream() .map(messageMonitor -> messageMonitor.onMessageIngested(message)) .collect(Collectors.toList()); return new MonitorCallback() { @Override public void reportSuccess() { monitorCallbacks.forEach(MonitorCallback::reportSuccess); } @Override public void reportFailure(Throwable cause) { monitorCallbacks.forEach(resultCallback -> resultCallback.reportFailure(cause)); } @Override public void reportIgnored() { monitorCallbacks.forEach(MonitorCallback::reportIgnored); } }; } }
MultiMessageMonitor implements MessageMonitor<T> { @Override public MonitorCallback onMessageIngested(T message) { final List<MonitorCallback> monitorCallbacks = messageMonitors.stream() .map(messageMonitor -> messageMonitor.onMessageIngested(message)) .collect(Collectors.toList()); return new MonitorCallback() { @Override public void reportSuccess() { monitorCallbacks.forEach(MonitorCallback::reportSuccess); } @Override public void reportFailure(Throwable cause) { monitorCallbacks.forEach(resultCallback -> resultCallback.reportFailure(cause)); } @Override public void reportIgnored() { monitorCallbacks.forEach(MonitorCallback::reportIgnored); } }; } @SafeVarargs  MultiMessageMonitor(MessageMonitor<? super T>... messageMonitors);  MultiMessageMonitor(List<MessageMonitor<? super T>> messageMonitors); }
MultiMessageMonitor implements MessageMonitor<T> { @Override public MonitorCallback onMessageIngested(T message) { final List<MonitorCallback> monitorCallbacks = messageMonitors.stream() .map(messageMonitor -> messageMonitor.onMessageIngested(message)) .collect(Collectors.toList()); return new MonitorCallback() { @Override public void reportSuccess() { monitorCallbacks.forEach(MonitorCallback::reportSuccess); } @Override public void reportFailure(Throwable cause) { monitorCallbacks.forEach(resultCallback -> resultCallback.reportFailure(cause)); } @Override public void reportIgnored() { monitorCallbacks.forEach(MonitorCallback::reportIgnored); } }; } @SafeVarargs  MultiMessageMonitor(MessageMonitor<? super T>... messageMonitors);  MultiMessageMonitor(List<MessageMonitor<? super T>> messageMonitors); @Override MonitorCallback onMessageIngested(T message); }
MultiMessageMonitor implements MessageMonitor<T> { @Override public MonitorCallback onMessageIngested(T message) { final List<MonitorCallback> monitorCallbacks = messageMonitors.stream() .map(messageMonitor -> messageMonitor.onMessageIngested(message)) .collect(Collectors.toList()); return new MonitorCallback() { @Override public void reportSuccess() { monitorCallbacks.forEach(MonitorCallback::reportSuccess); } @Override public void reportFailure(Throwable cause) { monitorCallbacks.forEach(resultCallback -> resultCallback.reportFailure(cause)); } @Override public void reportIgnored() { monitorCallbacks.forEach(MonitorCallback::reportIgnored); } }; } @SafeVarargs  MultiMessageMonitor(MessageMonitor<? super T>... messageMonitors);  MultiMessageMonitor(List<MessageMonitor<? super T>> messageMonitors); @Override MonitorCallback onMessageIngested(T message);  }