AnalyticsClient { public void enqueue(Message message) { try { messageQueue.put(message); } catch (InterruptedException e) { log.print(ERROR, e, "Interrupted while adding message %s.", message); } } AnalyticsClient(
      BlockingQueue<Message> messageQueue,
      SegmentService service,
      int maxQueueSize,
      long flushIntervalInMillis,
      Log log,
      ThreadFactory threadFactory,
      ExecutorService networkExecutor,
      List<Callback> callbacks); static AnalyticsClient create(
      SegmentService segmentService,
      int flushQueueSize,
      long flushIntervalInMillis,
      Log log,
      ThreadFactory threadFactory,
      ExecutorService networkExecutor,
      List<Callback> callbacks); void enqueue(Message message); void flush(); void shutdown();  }
@Test public void enqueueAddsToQueue(MessageBuilderTest builder) throws InterruptedException { AnalyticsClient client = newClient(); Message message = builder.get().userId("prateek").build(); client.enqueue(message); verify(messageQueue).put(message); }
@Test public void enqueueMaxTriggersFlush() { messageQueue = new LinkedBlockingQueue<>(); AnalyticsClient client = newClient(); for (int i = 0; i < 51; i++) { client.enqueue(TrackMessage.builder("Event " + i).userId("bar").build()); } wait(messageQueue); assertThat(captureBatch(networkExecutor).batch()).hasSize(50); }
@Test public void enqueueBeforeMaxDoesNotTriggerFlush() { messageQueue = new LinkedBlockingQueue<>(); AnalyticsClient client = newClient(); for (int i = 0; i < 5; i++) { client.enqueue(TrackMessage.builder("Event " + i).userId("bar").build()); } wait(messageQueue); verify(networkExecutor, never()).submit(any(Runnable.class)); }