AzureBlobSystemProducer implements SystemProducer { @Override public void send(String source, OutgoingMessageEnvelope messageEnvelope) { if (!isStarted) { throw new SystemProducerException("Trying to send before producer has started."); } if (isStopped) { throw new SystemProducerException("Sending after producer has been stopped."); } ReadWriteLock lock = sourceSendFlushLockMap.get(source); if (lock == null) { throw new SystemProducerException("Attempting to send to source: " + source + " but it was not registered"); } lock.readLock().lock(); try { AzureBlobWriter writer = getOrCreateWriter(source, messageEnvelope); writer.write(messageEnvelope); metrics.updateWriteMetrics(source); } catch (Exception e) { metrics.updateErrorMetrics(source); Object partitionKey = getPartitionKey(messageEnvelope); String msg = "Send failed for source: " + source + ", system: " + systemName + ", stream: " + messageEnvelope.getSystemStream().getStream() + ", partitionKey: " + ((partitionKey != null) ? partitionKey : "null"); throw new SystemProducerException(msg, e); } finally { lock.readLock().unlock(); } } 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 testSend() throws IOException { int numberOfMessages = 10; Counter mockWriteCounter = mock(Counter.class); when(mockMetricsRegistry.newCounter(SOURCE, AzureBlobBasicMetrics.EVENT_WRITE_RATE)).thenReturn(mockWriteCounter); systemProducer.register(SOURCE); systemProducer.start(); for (int i = 0; i < numberOfMessages; i++) { systemProducer.send(SOURCE, ome); } verify(mockAzureWriter, times(numberOfMessages)).write(ome); verify(mockWriteCounter, times(numberOfMessages)).inc(); }