AzureBlobSystemProducer implements SystemProducer { @Override public synchronized void register(String source) { LOG.info("Registering source {}", source); if (isStarted) { throw new SystemProducerException("Cannot register once the producer is started."); } if (writerMap.containsKey(source)) { LOG.warn("Source: {} already registered", source); return; } writerMap.put(source, new ConcurrentHashMap<>()); sourceWriterCreationLockMap.put(source, new Object()); sourceSendFlushLockMap.put(source, new ReentrantReadWriteLock()); metrics.register(source); } AzureBlobSystemProducer(String systemName, AzureBlobConfig config, MetricsRegistry metricsRegistry); @Override synchronized void start(); @Override synchronized void stop(); @Override synchronized void register(String source); @Override void send(String source, OutgoingMessageEnvelope messageEnvelope); @Override void flush(String source);  }
@Test public void testRegisterMetrics() throws Exception { systemProducer.register(SOURCE); verify(mockMetricsRegistry).newCounter( String.format(AzureBlobSystemProducerMetrics.SYSTEM_METRIC_FORMAT, ACCOUNT_NAME, SYSTEM_NAME), AzureBlobBasicMetrics.EVENT_WRITE_RATE); verify(mockMetricsRegistry).newCounter(SOURCE, AzureBlobBasicMetrics.EVENT_WRITE_RATE); verify(mockMetricsRegistry).newCounter(SOURCE, AzureBlobBasicMetrics.EVENT_WRITE_BYTE_RATE); verify(mockMetricsRegistry).newCounter(SOURCE, AzureBlobBasicMetrics.EVENT_PRODUCE_ERROR); }
@Test public void testRegisterWithSystemName() throws Exception { systemProducer.register(SYSTEM_NAME); verify(mockMetricsRegistry).newCounter( String.format(AzureBlobSystemProducerMetrics.SYSTEM_METRIC_FORMAT, ACCOUNT_NAME, SYSTEM_NAME), AzureBlobBasicMetrics.EVENT_WRITE_RATE); }