FixedWindows extends PartitioningWindowFn<Object, IntervalWindow> { public static FixedWindows of(Duration size) { return new FixedWindows(size, Duration.ZERO); } private  FixedWindows(Duration size, Duration offset); static FixedWindows of(Duration size); FixedWindows withOffset(Duration offset); @Override IntervalWindow assignWindow(Instant timestamp); @Override void populateDisplayData(DisplayData.Builder builder); @Override Coder<IntervalWindow> windowCoder(); @Override boolean isCompatible(WindowFn<?, ?> other); @Override void verifyCompatibility(WindowFn<?, ?> other); Duration getSize(); Duration getOffset(); @Override boolean equals(@Nullable Object object); @Override int hashCode();  }
@Test public void testSimpleFixedWindow() throws Exception { Map<IntervalWindow, Set<String>> expected = new HashMap<>(); expected.put(new IntervalWindow(new Instant(0), new Instant(10)), set(1, 2, 5, 9)); expected.put(new IntervalWindow(new Instant(10), new Instant(20)), set(10, 11)); expected.put(new IntervalWindow(new Instant(100), new Instant(110)), set(100)); assertEquals( expected, runWindowFn( FixedWindows.of(new Duration(10)), Arrays.asList(1L, 2L, 5L, 9L, 10L, 11L, 100L))); }
@Test public void testDefaultWindowMappingFn() { PartitioningWindowFn<?, ?> windowFn = FixedWindows.of(Duration.standardMinutes(20L)); WindowMappingFn<?> mapping = windowFn.getDefaultWindowMappingFn(); assertThat( mapping.getSideInputWindow( new BoundedWindow() { @Override public Instant maxTimestamp() { return new Instant(100L); } }), equalTo( new IntervalWindow( new Instant(0L), new Instant(0L).plus(Duration.standardMinutes(20L))))); assertThat(mapping.maximumLookback(), equalTo(Duration.ZERO)); }
@Test public void testDefaultWindowMappingFnGlobalWindow() { PartitioningWindowFn<?, ?> windowFn = FixedWindows.of(Duration.standardMinutes(20L)); WindowMappingFn<?> mapping = windowFn.getDefaultWindowMappingFn(); thrown.expect(IllegalArgumentException.class); mapping.getSideInputWindow(GlobalWindow.INSTANCE); }
@Test public void testValidOutputTimes() throws Exception { for (long timestamp : Arrays.asList(200, 800, 700)) { WindowFnTestUtils.validateGetOutputTimestamp(FixedWindows.of(new Duration(500)), timestamp); } }