LargestPerTimeBucket implements EmissionStrategy { @Override public <T extends Batchable<?>, I> Queue<Batch<T, I>> compute(Stream<Batch<T, I>> candidates) { final Instant now = Instant.ofEpochMilli(scheduler.now()); final Instant cutLine = now.minus(minimumTimeInQueueMs, ChronoUnit.MILLIS); final Stream<Batch<T, I>> inQueueMinimumRequired = candidates.filter( batch -> !batch.getOldestItemTimestamp().isAfter(cutLine) ); PriorityQueue<Batch<T, I>> queue = new PriorityQueue<>(this::compare); inQueueMinimumRequired.forEach(queue::offer); return queue; } LargestPerTimeBucket(long minimumTimeInQueueMs, long timeWindowBucketSizeMs, Scheduler scheduler); @Override Queue<Batch<T, I>> compute(Stream<Batch<T, I>> candidates);  }
@Test public void filterOutBatchesNotQueueingForAMinimumPeriod() { final Instant now = Instant.ofEpochMilli(testScheduler.now()); final Instant past = now.minus(ofMinutes(10)); final Stream<Batch<BatchableOperationMock, String>> batches = Stream.of( Batch.of("first", new BatchableOperationMock(LOW, now, "first", "sub1", "some"), new BatchableOperationMock(HIGH, now, "first", "sub2", "some"), new BatchableOperationMock(LOW, now, "first", "sub3", "some") ), Batch.of("second", new BatchableOperationMock(LOW, now, "second", "sub1", "some"), new BatchableOperationMock(LOW, now, "second", "sub2", "some"), new BatchableOperationMock(HIGH, past, "second", "sub3", "some") ), Batch.of("third", new BatchableOperationMock(LOW, past, "third", "sub1", "someState") ) ); EmissionStrategy strategy = new LargestPerTimeBucket(300_000 , NO_BUCKETS, testScheduler); Queue<Batch<BatchableOperationMock, String>> toEmit = strategy.compute(batches); Assertions.assertThat(toEmit).hasSize(2); assertThat(toEmit.poll().getIndex()).isEqualTo("second"); assertThat(toEmit.poll().getIndex()).isEqualTo("third"); }
@Test public void batchesWithOlderItemsGoFirst() { final long groupInBucketsOfMs = 1; final Instant now = Instant.ofEpochMilli(testScheduler.now()); final Stream<Batch<BatchableOperationMock, String>> batches = Stream.of( Batch.of("first", new BatchableOperationMock(LOW, now, "first", "sub1", "foo"), new BatchableOperationMock(LOW, now.minus(ofSeconds(1)), "first", "sub2", "foo") ), Batch.of("second", new BatchableOperationMock(HIGH, now.minus(ofSeconds(2)), "second", "sub1", "foo"), new BatchableOperationMock(LOW, now, "second", "sub2", "foo") ), Batch.of("third", new BatchableOperationMock(LOW, now.minus(ofMillis(2_001)), "third", "sub1", "foo"), new BatchableOperationMock(LOW, now, "third", "sub2", "foo") ), Batch.of("fourth", new BatchableOperationMock(LOW, now.minus(ofMinutes(1)), "fourth", "sub1", "foo"), new BatchableOperationMock(HIGH, now, "third", "sub2", "foo") ), Batch.of("fifth", new BatchableOperationMock(LOW, now.minus(ofMillis(1)), "fifth", "sub1", "foo"), new BatchableOperationMock(LOW, now, "third", "sub2", "foo") ) ); EmissionStrategy strategy = new LargestPerTimeBucket(0, groupInBucketsOfMs, testScheduler); Queue<Batch<BatchableOperationMock, String>> toEmit = strategy.compute(batches); Assertions.assertThat(toEmit).hasSize(5); assertThat(toEmit.poll().getIndex()).isEqualTo("fourth"); assertThat(toEmit.poll().getIndex()).isEqualTo("third"); assertThat(toEmit.poll().getIndex()).isEqualTo("second"); assertThat(toEmit.poll().getIndex()).isEqualTo("first"); assertThat(toEmit.poll().getIndex()).isEqualTo("fifth"); }
@Test public void biggerBatchesInTheSameBucketGoFirst() { final long groupInBucketsOfMs = 60_000 ; final Instant now = Instant.ofEpochMilli(testScheduler.now()); final Instant past = now.minus(ofMinutes(10)); final Stream<Batch<BatchableOperationMock, String>> batches = Stream.of( Batch.of("smaller", new BatchableOperationMock(LOW, now, "smaller", "sub1", "foo"), new BatchableOperationMock(LOW, past.minus(ofSeconds(10)), "smaller", "sub2", "foo") ), Batch.of("bigger", new BatchableOperationMock(LOW, past, "bigger", "sub1", "foo"), new BatchableOperationMock(HIGH, now.minus(ofMinutes(5)), "bigger", "sub2", "foo"), new BatchableOperationMock(LOW, now, "bigger", "sub3", "foo") ), Batch.of("slightlyOlder", new BatchableOperationMock(LOW, now, "slightlyOlder", "sub1", "foo"), new BatchableOperationMock(LOW, past.minus(ofSeconds(11)), "slightlyOlder", "sub2", "foo") ), Batch.of("older", new BatchableOperationMock(LOW, past.minus(ofMinutes(5)), "older", "sub1", "foo") ) ); EmissionStrategy strategy = new LargestPerTimeBucket(0, groupInBucketsOfMs, testScheduler); Queue<Batch<BatchableOperationMock, String>> toEmit = strategy.compute(batches); Assertions.assertThat(toEmit).hasSize(4); assertThat(toEmit.poll().getIndex()).isEqualTo("older"); assertThat(toEmit.poll().getIndex()).isEqualTo("bigger"); assertThat(toEmit.poll().getIndex()).isEqualTo("slightlyOlder"); assertThat(toEmit.poll().getIndex()).isEqualTo("smaller"); }