Sessions extends WindowFn<Object, IntervalWindow> { public static Sessions withGapDuration(Duration gapDuration) { return new Sessions(gapDuration); } private  Sessions(Duration gapDuration); static Sessions withGapDuration(Duration gapDuration); @Override Collection<IntervalWindow> assignWindows(AssignContext c); @Override void mergeWindows(MergeContext c); @Override Coder<IntervalWindow> windowCoder(); @Override boolean isCompatible(WindowFn<?, ?> other); @Override void verifyCompatibility(WindowFn<?, ?> other); @Override TypeDescriptor<IntervalWindow> getWindowTypeDescriptor(); @Override WindowMappingFn<IntervalWindow> getDefaultWindowMappingFn(); Duration getGapDuration(); @Override void populateDisplayData(DisplayData.Builder builder); @Override boolean equals(@Nullable Object object); @Override int hashCode();  }
@Test public void testSimple() throws Exception { Map<IntervalWindow, Set<String>> expected = new HashMap<>(); expected.put(new IntervalWindow(new Instant(0), new Instant(10)), set(0)); expected.put(new IntervalWindow(new Instant(10), new Instant(20)), set(10)); expected.put(new IntervalWindow(new Instant(101), new Instant(111)), set(101)); assertEquals( expected, runWindowFn(Sessions.withGapDuration(new Duration(10)), Arrays.asList(0L, 10L, 101L))); }
@Test public void testConsecutive() throws Exception { Map<IntervalWindow, Set<String>> expected = new HashMap<>(); expected.put(new IntervalWindow(new Instant(1), new Instant(19)), set(1, 2, 5, 9)); expected.put(new IntervalWindow(new Instant(100), new Instant(111)), set(100, 101)); assertEquals( expected, runWindowFn( Sessions.withGapDuration(new Duration(10)), Arrays.asList(1L, 2L, 5L, 9L, 100L, 101L))); }
@Test public void testMerging() throws Exception { Map<IntervalWindow, Set<String>> expected = new HashMap<>(); expected.put(new IntervalWindow(new Instant(1), new Instant(40)), set(1, 10, 15, 22, 30)); expected.put(new IntervalWindow(new Instant(95), new Instant(111)), set(95, 100, 101)); assertEquals( expected, runWindowFn( Sessions.withGapDuration(new Duration(10)), Arrays.asList(1L, 15L, 30L, 100L, 101L, 95L, 22L, 10L))); }
@Test public void testTimeUnit() throws Exception { Map<IntervalWindow, Set<String>> expected = new HashMap<>(); expected.put(new IntervalWindow(new Instant(1), new Instant(2000)), set(1, 2, 1000)); expected.put(new IntervalWindow(new Instant(5000), new Instant(6001)), set(5000, 5001)); expected.put(new IntervalWindow(new Instant(10000), new Instant(11000)), set(10000)); assertEquals( expected, runWindowFn( Sessions.withGapDuration(Duration.standardSeconds(1)), Arrays.asList(1L, 2L, 1000L, 5000L, 5001L, 10000L))); }
@Test public void testValidOutputTimes() throws Exception { for (long timestamp : Arrays.asList(200, 800, 700)) { WindowFnTestUtils.validateGetOutputTimestamp( Sessions.withGapDuration(Duration.millis(500)), timestamp); } }
@Test public void testInvalidOutputAtEarliest() throws Exception { try { WindowFnTestUtils.validateGetOutputTimestamps( Sessions.withGapDuration(Duration.millis(10)), TimestampCombiner.EARLIEST, ImmutableList.of( (List<Long>) ImmutableList.of(1L, 3L), (List<Long>) ImmutableList.of(0L, 5L, 10L, 15L, 20L))); } catch (AssertionError exc) { assertThat( exc.getMessage(), allOf(containsString("a value greater than"), containsString("was less than"))); } }
@Test public void testValidOutputAtEndTimes() throws Exception { WindowFnTestUtils.validateGetOutputTimestamps( Sessions.withGapDuration(Duration.millis(10)), TimestampCombiner.END_OF_WINDOW, ImmutableList.of( (List<Long>) ImmutableList.of(1L, 3L), (List<Long>) ImmutableList.of(0L, 5L, 10L, 15L, 20L))); }
@Test public void testDisplayData() { Duration gapDuration = Duration.standardMinutes(234); Sessions session = Sessions.withGapDuration(gapDuration); assertThat(DisplayData.from(session), hasDisplayItem("gapDuration", gapDuration)); }