ZookeeperCounterStorage implements CounterStorage { @Override public void setSubscriptionDeliveredCounter(TopicName topicName, String subscriptionName, long count) { try { subscriptionRepository.ensureSubscriptionExists(topicName, subscriptionName); incrementSharedCounter(subscriptionDeliveredCounter(topicName, subscriptionName), count); } catch (SubscriptionNotExistsException e) { LOGGER.debug("Trying to report metric on not existing subscription {} {}", topicName, subscriptionName); } } @Inject  ZookeeperCounterStorage(SharedCounter sharedCounter,
                                   SubscriptionRepository subscriptionRepository,
                                   PathsCompiler pathsCompiler,
                                   ConfigFactory configFactory); @Override void setTopicPublishedCounter(TopicName topicName, long count); @Override long getTopicPublishedCounter(TopicName topicName); @Override long getSubscriptionDeliveredCounter(TopicName topicName, String subscriptionName); @Override void setSubscriptionDeliveredCounter(TopicName topicName, String subscriptionName, long count); @Override void setSubscriptionDiscardedCounter(TopicName topicName, String subscriptionName, long count); @Override void incrementVolumeCounter(TopicName topicName, String subscriptionName, long value); @Override void incrementVolumeCounter(TopicName topicName, long value);  }
@Test public void shouldIncrementSubscriptionMetricUsingSharedCounter() { storage.setSubscriptionDeliveredCounter(TopicName.fromQualifiedName("test.topic"), "sub", 10); verify(sharedCounter).increment("/hermes/groups/test/topics/topic/subscriptions/sub/metrics/delivered", 10); }
@Test public void shouldNotIncrementSharedCounterForNonExistingSubscription() { TopicName topicName = TopicName.fromQualifiedName("test.topic"); String subscriptionName = "sub"; doThrow(new SubscriptionNotExistsException(topicName, subscriptionName)) .when(subscriptionRepository).ensureSubscriptionExists(topicName, subscriptionName); storage.setSubscriptionDeliveredCounter(topicName, subscriptionName, 1L); verifyZeroInteractions(sharedCounter); }