Watch { public static <InputT, OutputT> Growth<InputT, OutputT, OutputT> growthOf( Growth.PollFn<InputT, OutputT> pollFn, Requirements requirements) { return new AutoValue_Watch_Growth.Builder<InputT, OutputT, OutputT>() .setTerminationPerInput(Growth.never()) .setPollFn(Contextful.of(pollFn, requirements)) .setOutputKeyFn(null) .build(); }  static Growth<InputT, OutputT, OutputT> growthOf(
      Growth.PollFn<InputT, OutputT> pollFn, Requirements requirements); static Growth<InputT, OutputT, OutputT> growthOf(
      Growth.PollFn<InputT, OutputT> pollFn); static Growth<InputT, OutputT, KeyT> growthOf(
      Contextful<Growth.PollFn<InputT, OutputT>> pollFn,
      SerializableFunction<OutputT, KeyT> outputKeyFn);  }
@Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testSinglePollMultipleInputs() { PCollection<KV<String, String>> res = p.apply(Create.of("a", "b")) .apply( Watch.growthOf( new PollFn<String, String>() { @Override public PollResult<String> apply(String element, Context c) throws Exception { return PollResult.complete( Instant.now(), Arrays.asList(element + ".foo", element + ".bar")); } }) .withPollInterval(Duration.ZERO)); PAssert.that(res) .containsInAnyOrder( Arrays.asList( KV.of("a", "a.foo"), KV.of("a", "a.bar"), KV.of("b", "b.foo"), KV.of("b", "b.bar"))); p.run(); }
@Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testSinglePollMultipleInputsWithSideInput() { final PCollectionView<String> moo = p.apply("moo", Create.of("moo")).apply("moo singleton", View.asSingleton()); final PCollectionView<String> zoo = p.apply("zoo", Create.of("zoo")).apply("zoo singleton", View.asSingleton()); PCollection<KV<String, String>> res = p.apply("input", Create.of("a", "b")) .apply( Watch.growthOf( new PollFn<String, String>() { @Override public PollResult<String> apply(String element, Context c) throws Exception { return PollResult.complete( Instant.now(), Arrays.asList( element + " " + c.sideInput(moo) + " " + c.sideInput(zoo))); } }, requiresSideInputs(moo, zoo)) .withPollInterval(Duration.ZERO)); PAssert.that(res) .containsInAnyOrder(Arrays.asList(KV.of("a", "a moo zoo"), KV.of("b", "b moo zoo"))); p.run(); }
@Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testMultiplePollsWithKeyExtractor() { List<KV<Integer, String>> polls = Arrays.asList( KV.of(0, "0"), KV.of(10, "10"), KV.of(20, "20"), KV.of(30, "30"), KV.of(40, "40"), KV.of(40, "40.1"), KV.of(20, "20.1"), KV.of(50, "50"), KV.of(10, "10.1"), KV.of(10, "10.2"), KV.of(60, "60"), KV.of(70, "70"), KV.of(60, "60.1"), KV.of(80, "80"), KV.of(40, "40.2"), KV.of(90, "90"), KV.of(90, "90.1")); List<Integer> expected = Arrays.asList(0, 10, 20, 30, 40, 50, 60, 70, 80, 90); PCollection<Integer> res = p.apply(Create.of("a")) .apply( Watch.growthOf( Contextful.of( new TimedPollFn<String, KV<Integer, String>>( polls, standardSeconds(1) , standardSeconds(3) , standardSeconds(30) ), Requirements.empty()), KV::getKey) .withTerminationPerInput(Growth.afterTotalOf(standardSeconds(5))) .withPollInterval(Duration.millis(100)) .withOutputCoder(KvCoder.of(VarIntCoder.of(), StringUtf8Coder.of())) .withOutputKeyCoder(VarIntCoder.of())) .apply("Drop input", Values.create()) .apply("Drop auxiliary string", Keys.create()); PAssert.that(res).containsInAnyOrder(expected); p.run(); }
@Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testMultiplePollsStopAfterTimeSinceNewOutput() { List<Integer> all = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); PCollection<Integer> res = p.apply(Create.of("a")) .apply( Watch.growthOf( new TimedPollFn<String, Integer>( all, standardSeconds(1) , standardSeconds(1000) , standardSeconds(30) )) .withTerminationPerInput(afterTimeSinceNewOutput(standardSeconds(3))) .withPollInterval(Duration.millis(300)) .withOutputCoder(VarIntCoder.of())) .apply("Drop input", Values.create()); PAssert.that(res).containsInAnyOrder(all); p.run(); }
@Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testSinglePollWithManyResults() { final long numResults = 3000; PCollection<KV<String, Integer>> res = p.apply(Create.of("a")) .apply( Watch.growthOf( new PollFn<String, KV<String, Integer>>() { @Override public PollResult<KV<String, Integer>> apply(String element, Context c) throws Exception { String pollId = UUID.randomUUID().toString(); List<KV<String, Integer>> output = Lists.newArrayList(); for (int i = 0; i < numResults; ++i) { output.add(KV.of(pollId, i)); } return PollResult.complete(Instant.now(), output); } }) .withTerminationPerInput(Growth.afterTotalOf(standardSeconds(1))) .withPollInterval(Duration.millis(1)) .withOutputCoder(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))) .apply("Drop input", Values.create()); PAssert.that("Poll called only once", res.apply(Keys.create())) .satisfies( pollIds -> { assertEquals(1, Sets.newHashSet(pollIds).size()); return null; }); PAssert.that("Yields all expected results", res.apply("Drop poll id", Values.create())) .satisfies( input -> { assertEquals( "Total number of results mismatches", numResults, Lists.newArrayList(input).size()); assertEquals("Results are not unique", numResults, Sets.newHashSet(input).size()); return null; }); p.run(); }
@Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) @Ignore("https: public void testMultiplePollsWithManyResults() { final long numResults = 3000; List<Integer> all = Lists.newArrayList(); for (int i = 0; i < numResults; ++i) { all.add(i); } PCollection<TimestampedValue<Integer>> res = p.apply(Create.of("a")) .apply( Watch.growthOf( new TimedPollFn<String, Integer>( all, standardSeconds(1) , standardSeconds(3) , standardSeconds(30) )) .withPollInterval(Duration.millis(500)) .withOutputCoder(VarIntCoder.of())) .apply(Reify.timestampsInValue()) .apply("Drop timestamped input", Values.create()); PAssert.that(res) .satisfies( outputs -> { Function<TimestampedValue<Integer>, Integer> extractValueFn = new Function<TimestampedValue<Integer>, Integer>() { @Nullable @Override public Integer apply(@Nullable TimestampedValue<Integer> input) { return input.getValue(); } }; Function<TimestampedValue<Integer>, Instant> extractTimestampFn = new Function<TimestampedValue<Integer>, Instant>() { @Nullable @Override public Instant apply(@Nullable TimestampedValue<Integer> input) { return input.getTimestamp(); } }; Ordering<TimestampedValue<Integer>> byTimestamp = Ordering.natural().onResultOf(extractTimestampFn); assertTrue("Outputs must be in timestamp order", byTimestamp.isOrdered(outputs)); assertEquals( "Yields all expected values", numResults, Sets.newHashSet( StreamSupport.stream(outputs.spliterator(), false) .map(extractValueFn::apply) .collect(Collectors.toList())) .size()); assertThat( "Poll called more than once", Sets.newHashSet( StreamSupport.stream(outputs.spliterator(), false) .map(extractTimestampFn::apply) .collect(Collectors.toList())) .size(), greaterThan(1)); return null; }); p.run(); }
@Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testMultiplePollsWithManyResults() { final long numResults = 3000; List<Integer> all = Lists.newArrayList(); for (int i = 0; i < numResults; ++i) { all.add(i); } PCollection<TimestampedValue<Integer>> res = p.apply(Create.of("a")) .apply( Watch.growthOf( new TimedPollFn<String, Integer>( all, standardSeconds(1) , standardSeconds(3) , standardSeconds(30) )) .withPollInterval(Duration.millis(500)) .withOutputCoder(VarIntCoder.of())) .apply(Reify.timestampsInValue()) .apply("Drop timestamped input", Values.create()); PAssert.that(res) .satisfies( outputs -> { Function<TimestampedValue<Integer>, Integer> extractValueFn = new Function<TimestampedValue<Integer>, Integer>() { @Nullable @Override public Integer apply(@Nullable TimestampedValue<Integer> input) { return input.getValue(); } }; Function<TimestampedValue<Integer>, Instant> extractTimestampFn = new Function<TimestampedValue<Integer>, Instant>() { @Nullable @Override public Instant apply(@Nullable TimestampedValue<Integer> input) { return input.getTimestamp(); } }; Ordering<TimestampedValue<Integer>> byTimestamp = Ordering.natural().onResultOf(extractTimestampFn); assertTrue("Outputs must be in timestamp order", byTimestamp.isOrdered(outputs)); assertEquals( "Yields all expected values", numResults, Sets.newHashSet( StreamSupport.stream(outputs.spliterator(), false) .map(extractValueFn::apply) .collect(Collectors.toList())) .size()); assertThat( "Poll called more than once", Sets.newHashSet( StreamSupport.stream(outputs.spliterator(), false) .map(extractTimestampFn::apply) .collect(Collectors.toList())) .size(), greaterThan(1)); return null; }); p.run(); }