@Test void testInstantiatesOneMessageMonitorPerIngestedPayloadType() throws Exception { Field payloadTypeMonitorsField = testSubject.getClass().getDeclaredField("payloadTypeMonitors"); payloadTypeMonitorsField.setAccessible(true); String expectedStringMonitorName = STRING_MESSAGE.getPayloadType().getName(); String expectedIntegerMonitorName = INTEGER_MESSAGE.getPayloadType().getName(); testSubject.onMessageIngested(STRING_MESSAGE); testSubject.onMessageIngested(STRING_MESSAGE); testSubject.onMessageIngested(INTEGER_MESSAGE); Map<String, T> payloadTypeMonitors = ReflectionUtils.getFieldValue(payloadTypeMonitorsField, testSubject); assertEquals(2, payloadTypeMonitors.size()); MessageMonitor<Message<?>> messageMessageMonitor = payloadTypeMonitors.get(expectedStringMonitorName); assertNotNull(messageMessageMonitor); assertTrue(expectedMonitorClass.isInstance(messageMessageMonitor)); messageMessageMonitor = payloadTypeMonitors.get(expectedIntegerMonitorName); assertNotNull(messageMessageMonitor); assertTrue(expectedMonitorClass.isInstance(messageMessageMonitor)); assertEquals(1, meterRegistry.find(expectedStringMonitorName + ".capacity").meters().size()); assertEquals(1, meterRegistry.find(expectedIntegerMonitorName + ".capacity").meters().size()); }
@Override public MonitorCallback onMessageIngested(Message<?> message) { String monitorName = monitorNameBuilder.apply(message.getPayloadType()); MessageMonitor<Message<?>> messageMonitorForPayloadType = payloadTypeMonitors.computeIfAbsent(monitorName, monitorSupplier); return messageMonitorForPayloadType.onMessageIngested(message); }
PayloadTypeMessageMonitorWrapper implements MessageMonitor<Message<?>> { @Override public MonitorCallback onMessageIngested(Message<?> message) { String monitorName = monitorNameBuilder.apply(message.getPayloadType()); MessageMonitor<Message<?>> messageMonitorForPayloadType = payloadTypeMonitors.computeIfAbsent(monitorName, monitorSupplier); return messageMonitorForPayloadType.onMessageIngested(message); } }
PayloadTypeMessageMonitorWrapper implements MessageMonitor<Message<?>> { @Override public MonitorCallback onMessageIngested(Message<?> message) { String monitorName = monitorNameBuilder.apply(message.getPayloadType()); MessageMonitor<Message<?>> messageMonitorForPayloadType = payloadTypeMonitors.computeIfAbsent(monitorName, monitorSupplier); return messageMonitorForPayloadType.onMessageIngested(message); } PayloadTypeMessageMonitorWrapper(Function<String, T> monitorSupplier);  PayloadTypeMessageMonitorWrapper(Function<String, T> monitorSupplier,
                                            Function<Class<?>, String> monitorNameBuilder); }
PayloadTypeMessageMonitorWrapper implements MessageMonitor<Message<?>> { @Override public MonitorCallback onMessageIngested(Message<?> message) { String monitorName = monitorNameBuilder.apply(message.getPayloadType()); MessageMonitor<Message<?>> messageMonitorForPayloadType = payloadTypeMonitors.computeIfAbsent(monitorName, monitorSupplier); return messageMonitorForPayloadType.onMessageIngested(message); } PayloadTypeMessageMonitorWrapper(Function<String, T> monitorSupplier);  PayloadTypeMessageMonitorWrapper(Function<String, T> monitorSupplier,
                                            Function<Class<?>, String> monitorNameBuilder); @Override MonitorCallback onMessageIngested(Message<?> message); }
PayloadTypeMessageMonitorWrapper implements MessageMonitor<Message<?>> { @Override public MonitorCallback onMessageIngested(Message<?> message) { String monitorName = monitorNameBuilder.apply(message.getPayloadType()); MessageMonitor<Message<?>> messageMonitorForPayloadType = payloadTypeMonitors.computeIfAbsent(monitorName, monitorSupplier); return messageMonitorForPayloadType.onMessageIngested(message); } PayloadTypeMessageMonitorWrapper(Function<String, T> monitorSupplier);  PayloadTypeMessageMonitorWrapper(Function<String, T> monitorSupplier,
                                            Function<Class<?>, String> monitorNameBuilder); @Override MonitorCallback onMessageIngested(Message<?> message);  }