SideInputHandler implements ReadyCheckingSideInputReader { @Override public boolean isReady(PCollectionView<?> sideInput, BoundedWindow window) { Set<BoundedWindow> readyWindows = stateInternals.state(StateNamespaces.global(), availableWindowsTags.get(sideInput)).read(); return readyWindows != null && readyWindows.contains(window); } SideInputHandler(
      Collection<PCollectionView<?>> sideInputs, StateInternals stateInternals); void addSideInputValue(PCollectionView<?> sideInput, WindowedValue<Iterable<?>> value); @Nullable @Override T get(PCollectionView<T> view, BoundedWindow window); Iterable<?> getIterable(PCollectionView<T> view, BoundedWindow window); @Override boolean isReady(PCollectionView<?> sideInput, BoundedWindow window); @Override boolean contains(PCollectionView<T> view); @Override boolean isEmpty();  }
@Test public void testIsReady() { SideInputHandler sideInputHandler = new SideInputHandler( ImmutableList.of(view1, view2), InMemoryStateInternals.<Void>forKey(null)); IntervalWindow firstWindow = new IntervalWindow(new Instant(0), new Instant(WINDOW_MSECS_1)); IntervalWindow secondWindow = new IntervalWindow(new Instant(0), new Instant(WINDOW_MSECS_2)); assertFalse(sideInputHandler.isReady(view1, firstWindow)); sideInputHandler.addSideInputValue( view1, valuesInWindow( materializeValuesFor(view1.getPipeline().getOptions(), View.asIterable(), "Hello"), new Instant(0), firstWindow)); assertTrue(sideInputHandler.isReady(view1, firstWindow)); assertFalse(sideInputHandler.isReady(view1, secondWindow)); }
@Test public void testIsReady() { SideInputHandler sideInputHandler = new SideInputHandler( ImmutableList.of(view1, view2), InMemoryStateInternals.<Void>forKey(null)); IntervalWindow firstWindow = new IntervalWindow(new Instant(0), new Instant(WINDOW_MSECS_1)); IntervalWindow secondWindow = new IntervalWindow(new Instant(0), new Instant(WINDOW_MSECS_2)); assertFalse(sideInputHandler.isReady(view1, firstWindow)); sideInputHandler.addSideInputValue( view1, valuesInWindow( materializeValuesFor(View.asIterable(), "Hello"), new Instant(0), firstWindow)); assertTrue(sideInputHandler.isReady(view1, firstWindow)); assertFalse(sideInputHandler.isReady(view1, secondWindow)); }