@Override
    public void accept(T t) {
        updateTimeHighWaterMark(t.time());

        shortTermStorage.add(t);

        drainDueToLatestInput(t); //standard drain policy
        drainDueToTimeHighWaterMark();  //prevent blow-up when data goes backwards in time

        sizeHighWaterMark = Math.max(sizeHighWaterMark, shortTermStorage.size());
    }
@Test
    public void testVeryOldPointsAreAutoEvicted() {
        ConsumingArrayList<Point> downstreamConsumer = newConsumingArrayList();

        ApproximateTimeSorter<Point> sorter = new ApproximateTimeSorter<>(
            Duration.ofSeconds(10),
            downstreamConsumer
        );

        sorter.accept(Point.builder().latLong(0.0,0.0).time(EPOCH).build());
        sorter.accept(Point.builder().latLong(0.0,0.0).time(EPOCH.plusSeconds(4)).build());
        sorter.accept(Point.builder().latLong(0.0,0.0).time(EPOCH.plusSeconds(8)).build());
        assertThat("no evictions yet", downstreamConsumer, empty());

        //Adding this point evicts the 1st point because it is more than 10 seconds older than the most recent input
        sorter.accept(Point.builder().latLong(0.0,0.0).time(EPOCH.plusSeconds(12)).build()); //evict Epoch + 0
        assertThat(downstreamConsumer, hasSize(1));
        assertThat(downstreamConsumer.get(0).time(), equalTo(EPOCH));

        //This stale point should get instantly evicted because it is too old (with respect to time highwater mark)
        sorter.accept(Point.builder().latLong(0.0,0.0).time(EPOCH.minusSeconds(20)).build());
        assertThat(downstreamConsumer, hasSize(2));
        assertThat(downstreamConsumer.get(1).time(), equalTo(EPOCH.minusSeconds(20)));
    }