@Test void test_onMessageIngested_SingleMessageMonitor_failure() { MessageMonitor<Message<?>> messageMonitorMock = mock(MessageMonitor.class); MessageMonitor.MonitorCallback callback = mock(MessageMonitor.MonitorCallback.class); MultiMessageMonitor multiMessageMonitor = new MultiMessageMonitor(Arrays.asList(messageMonitorMock)); Message messageMock = mock(Message.class); Mockito.when(messageMonitorMock.onMessageIngested(messageMock)).thenReturn(callback); MessageMonitor.MonitorCallback monitorCallback = multiMessageMonitor.onMessageIngested(messageMock); Throwable throwable = new Throwable(); monitorCallback.reportFailure(throwable); Mockito.verify(messageMonitorMock).onMessageIngested(ArgumentMatchers.same(messageMock)); Mockito.verify(callback).reportFailure(ArgumentMatchers.same(throwable)); }
@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);  }