@Test void testMessagesWithMetadataAsCustomTag() { SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); MessageCountingMonitor testSubject = MessageCountingMonitor.buildMonitor(PROCESSOR_NAME, meterRegistry, message -> Tags .of("myMetaData", message.getMetaData() .get("myMetadataKey") .toString())); EventMessage<Object> foo = asEventMessage(1).withMetaData(Collections.singletonMap("myMetadataKey", "myMetadataValue1")); EventMessage<Object> bar = asEventMessage("bar").withMetaData(Collections.singletonMap("myMetadataKey", "myMetadataValue2")); ; EventMessage<Object> baz = asEventMessage("baz").withMetaData(Collections.singletonMap("myMetadataKey", "myMetadataValue2")); ; Map<? super Message<?>, MessageMonitor.MonitorCallback> callbacks = testSubject .onMessagesIngested(Arrays.asList(foo, bar, baz)); callbacks.get(foo).reportSuccess(); callbacks.get(bar).reportFailure(null); callbacks.get(baz).reportIgnored(); Collection<Counter> ingestedCounters = meterRegistry.find(PROCESSOR_NAME + ".ingestedCounter").counters(); Collection<Counter> processedCounters = meterRegistry.find(PROCESSOR_NAME + ".processedCounter").counters(); Collection<Counter> successCounters = meterRegistry.find(PROCESSOR_NAME + ".successCounter").counters(); Collection<Counter> failureCounters = meterRegistry.find(PROCESSOR_NAME + ".failureCounter").counters(); Collection<Counter> ignoredCounters = meterRegistry.find(PROCESSOR_NAME + ".ignoredCounter").counters(); assertEquals(2, ingestedCounters.size(), 0); assertEquals(2, processedCounters.size(), 0); assertEquals(2, successCounters.size(), 0); assertEquals(2, failureCounters.size(), 0); assertEquals(2, ignoredCounters.size(), 0); assertTrue(ingestedCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("myMetaData"), "myMetadataValue1")) .allMatch(counter -> counter.count() == 1)); assertTrue(ingestedCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("myMetaData"), "myMetadataValue2")) .allMatch(counter -> counter.count() == 2)); assertTrue(processedCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("myMetaData"), "myMetadataValue1")) .allMatch(counter -> counter.count() == 1)); assertTrue(processedCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("payloadType"), "myMetadataValue2")) .allMatch(counter -> counter.count() == 1)); assertTrue(successCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("payloadType"), "myMetadataValue1")) .allMatch(counter -> counter.count() == 1)); assertTrue(successCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("payloadType"), "myMetadataValue2")) .allMatch(counter -> counter.count() == 0)); assertTrue(failureCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("payloadType"), "myMetadataValue1")) .allMatch(counter -> counter.count() == 0)); assertTrue(failureCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("payloadType"), "myMetadataValue2")) .allMatch(counter -> counter.count() == 1)); assertTrue(ignoredCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("payloadType"), "myMetadataValue1")) .allMatch(counter -> counter.count() == 0)); assertTrue(ignoredCounters.stream() .filter(counter -> Objects .equals(counter.getId().getTag("payloadType"), "myMetadataValue2")) .allMatch(counter -> counter.count() == 1)); }
public static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry) { return new MessageCountingMonitor(meterNamePrefix, meterRegistry); }
MessageCountingMonitor implements MessageMonitor<Message<?>> { public static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry) { return new MessageCountingMonitor(meterNamePrefix, meterRegistry); } }
MessageCountingMonitor implements MessageMonitor<Message<?>> { public static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry) { return new MessageCountingMonitor(meterNamePrefix, meterRegistry); } private  MessageCountingMonitor(String meterNamePrefix, MeterRegistry meterRegistry); private  MessageCountingMonitor(String meterNamePrefix, MeterRegistry meterRegistry,
                                   Function<Message<?>, Iterable<Tag>> tagsBuilder); }
MessageCountingMonitor implements MessageMonitor<Message<?>> { public static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry) { return new MessageCountingMonitor(meterNamePrefix, meterRegistry); } private  MessageCountingMonitor(String meterNamePrefix, MeterRegistry meterRegistry); private  MessageCountingMonitor(String meterNamePrefix, MeterRegistry meterRegistry,
                                   Function<Message<?>, Iterable<Tag>> tagsBuilder); static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry); static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry,
                                                      Function<Message<?>, Iterable<Tag>> tagsBuilder); @Override MonitorCallback onMessageIngested(Message<?> message); }
MessageCountingMonitor implements MessageMonitor<Message<?>> { public static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry) { return new MessageCountingMonitor(meterNamePrefix, meterRegistry); } private  MessageCountingMonitor(String meterNamePrefix, MeterRegistry meterRegistry); private  MessageCountingMonitor(String meterNamePrefix, MeterRegistry meterRegistry,
                                   Function<Message<?>, Iterable<Tag>> tagsBuilder); static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry); static MessageCountingMonitor buildMonitor(String meterNamePrefix, MeterRegistry meterRegistry,
                                                      Function<Message<?>, Iterable<Tag>> tagsBuilder); @Override MonitorCallback onMessageIngested(Message<?> message);  }