@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"); }
@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 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 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); }
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); }
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);  }