BatchSizePolicy implements FlushPolicy<MESSAGE_T> { @Override public boolean shouldFlush(String sensorType, WriterConfiguration configurations, List<BulkMessage<MESSAGE_T>> messages) { boolean shouldFlush = false; int batchSize = messages.size(); int configuredBatchSize = configurations.getBatchSize(sensorType); if (batchSize >= configuredBatchSize) { LOG.debug("Batch size of {} reached. Flushing {} messages for sensor {}.", configuredBatchSize, batchSize, sensorType); shouldFlush = true; } return shouldFlush; }  @Override boolean shouldFlush(String sensorType, WriterConfiguration configurations, List<BulkMessage<MESSAGE_T>> messages); @Override void onFlush(String sensorType, BulkWriterResponse response);  }
@Test public void shouldFlushWhenBatchSizeReached() { BatchSizePolicy<JSONObject> batchSizePolicy = new BatchSizePolicy<>(); messages.add(new BulkMessage<>("message1", new JSONObject())); messages.add(new BulkMessage<>("message2", new JSONObject())); assertTrue(batchSizePolicy.shouldFlush(sensorType, configurations, messages)); }
@Test public void shouldNotFlushWhenBatchSizeNotReached() { BatchSizePolicy<JSONObject> batchSizePolicy = new BatchSizePolicy<>(); messages.add(new BulkMessage<>("message1", new JSONObject())); assertFalse(batchSizePolicy.shouldFlush(sensorType, configurations, messages)); }
@Test public void shouldFlushWhenBatchSizeExceeded() { BatchSizePolicy<JSONObject> batchSizePolicy = new BatchSizePolicy<>(); messages.add(new BulkMessage<>("message1", new JSONObject())); messages.add(new BulkMessage<>("message2", new JSONObject())); messages.add(new BulkMessage<>("message3", new JSONObject())); assertTrue(batchSizePolicy.shouldFlush(sensorType, configurations, messages)); }