ConnectivityCounterRegistry implements ConnectionMonitorRegistry<ConnectionMetricsCounter> { public SourceMetrics aggregateSourceMetrics(final ConnectionId connectionId) { return ConnectivityModelFactory.newSourceMetrics(aggregateMetrics(connectionId, MetricDirection.INBOUND)); } private  ConnectivityCounterRegistry(); static ConnectivityCounterRegistry fromConfig(final MonitoringCounterConfig config); @Override void initForConnection(final Connection connection); @Override void resetForConnection(final Connection connection); ConnectionMetricsCounter getCounter(
            final ConnectionId connectionId,
            final MetricType metricType,
            final MetricDirection metricDirection,
            final String address); @Override ConnectionMetricsCounter forOutboundDispatched(final ConnectionId connectionId, final String target); @Override ConnectionMetricsCounter forOutboundAcknowledged(final ConnectionId connectionId, final String target); @Override ConnectionMetricsCounter forOutboundFiltered(final ConnectionId connectionId, final String target); @Override ConnectionMetricsCounter forOutboundPublished(final ConnectionId connectionId, final String target); @Override ConnectionMetricsCounter forInboundConsumed(final ConnectionId connectionId, final String source); @Override ConnectionMetricsCounter forInboundAcknowledged(final ConnectionId connectionId, final String source); @Override ConnectionMetricsCounter forInboundMapped(final ConnectionId connectionId, final String source); @Override ConnectionMetricsCounter forInboundEnforced(final ConnectionId connectionId, final String source); @Override ConnectionMetricsCounter forInboundDropped(final ConnectionId connectionId, final String source); @Override ConnectionMetricsCounter forResponseDispatched(final ConnectionId connectionId); @Override ConnectionMetricsCounter forResponseDropped(final ConnectionId connectionId); @Override ConnectionMetricsCounter forResponseMapped(final ConnectionId connectionId); @Override ConnectionMetricsCounter forResponsePublished(final ConnectionId connectionId); SourceMetrics aggregateSourceMetrics(final ConnectionId connectionId); TargetMetrics aggregateTargetMetrics(final ConnectionId connectionId); static RetrieveConnectionMetricsResponse mergeRetrieveConnectionMetricsResponse(
            final RetrieveConnectionMetricsResponse first,
            final RetrieveConnectionMetricsResponse second); ConnectionMetrics aggregateConnectionMetrics(
            final SourceMetrics sourceMetrics, final TargetMetrics targetMetrics); static AddressMetric mergeAddressMetric(final AddressMetric first, final AddressMetric second); static Map<String, AddressMetric> mergeAddressMetricMap(final Map<String, AddressMetric> first,
            final Map<String, AddressMetric> second);  }
@Test public void testAggregateSourceMetrics() { final SourceMetrics sourceMetrics = COUNTER_REGISTRY.aggregateSourceMetrics(CONNECTION_ID); final Measurement[] expected = { getMeasurement(MAPPED, true), getMeasurement(MAPPED, false), getMeasurement(CONSUMED, true), getMeasurement(CONSUMED, false) }; assertThat(sourceMetrics.getAddressMetrics().get(SOURCE).getMeasurements()).containsExactlyInAnyOrder(expected); }