SideInputContainer { public ReadyCheckingSideInputReader createReaderForViews( Collection<PCollectionView<?>> newContainedViews) { if (!containedViews.containsAll(newContainedViews)) { Set<PCollectionView<?>> currentlyContained = ImmutableSet.copyOf(containedViews); Set<PCollectionView<?>> newRequested = ImmutableSet.copyOf(newContainedViews); throw new IllegalArgumentException( "Can't create a SideInputReader with unknown views " + Sets.difference(newRequested, currentlyContained)); } return new SideInputContainerSideInputReader(newContainedViews); } private  SideInputContainer(
      Collection<PCollectionView<?>> containedViews,
      LoadingCache<PCollectionViewWindow<?>, AtomicReference<Iterable<? extends WindowedValue<?>>>>
          viewByWindows); static SideInputContainer create(
      final EvaluationContext context, Collection<PCollectionView<?>> containedViews); ReadyCheckingSideInputReader createReaderForViews(
      Collection<PCollectionView<?>> newContainedViews); void write(PCollectionView<?> view, Iterable<? extends WindowedValue<?>> values);  }
@Test public void getNotReadyThrows() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("not ready"); container.createReaderForViews(ImmutableList.of(mapView)).get(mapView, GlobalWindow.INSTANCE); }
@Test public void withViewsForViewNotInContainerFails() { PCollection<KV<String, String>> input = pipeline.apply(Create.empty(new TypeDescriptor<KV<String, String>>() {})); PCollectionView<Map<String, Iterable<String>>> newView = input.apply(View.asMultimap()); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("unknown views"); thrown.expectMessage(newView.toString()); container.createReaderForViews(ImmutableList.of(newView)); }
@Test public void getOnReaderForViewNotInReaderFails() { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("unknown view: " + iterableView.toString()); container .createReaderForViews(ImmutableList.of(mapView)) .get(iterableView, GlobalWindow.INSTANCE); }
@Test public void finishOnPendingViewsSetsEmptyElements() throws Exception { immediatelyInvokeCallback(mapView, SECOND_WINDOW); Future<Map<String, Integer>> mapFuture = getFutureOfView( container.createReaderForViews(ImmutableList.of(mapView)), mapView, SECOND_WINDOW); assertThat(mapFuture.get().isEmpty(), is(true)); }
@Test public void isReadyInEmptyReaderThrows() { ReadyCheckingSideInputReader reader = container.createReaderForViews(ImmutableList.of()); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("does not contain"); thrown.expectMessage(ImmutableList.of().toString()); reader.isReady(mapView, GlobalWindow.INSTANCE); }
@Test public void isReadyForEmptyWindowTrue() throws Exception { CountDownLatch onComplete = new CountDownLatch(1); immediatelyInvokeCallback(mapView, GlobalWindow.INSTANCE); CountDownLatch latch = invokeLatchedCallback(singletonView, GlobalWindow.INSTANCE, onComplete); ReadyCheckingSideInputReader reader = container.createReaderForViews(ImmutableList.of(mapView, singletonView)); assertThat(reader.isReady(mapView, GlobalWindow.INSTANCE), is(true)); assertThat(reader.isReady(singletonView, GlobalWindow.INSTANCE), is(false)); latch.countDown(); if (!onComplete.await(1500L, TimeUnit.MILLISECONDS)) { fail("Callback to set empty values did not complete!"); } assertThat(reader.isReady(singletonView, GlobalWindow.INSTANCE), is(false)); reader = container.createReaderForViews(ImmutableList.of(mapView, singletonView)); assertThat(reader.isReady(singletonView, GlobalWindow.INSTANCE), is(true)); }