CountingSource { static UnboundedCountingSource createUnboundedFrom(long start) { return new UnboundedCountingSource(start, 1, 1L, Duration.ZERO, new NowTimestampFn()); } private  CountingSource(); @Deprecated static BoundedSource<Long> upTo(long numElements); @Deprecated static UnboundedSource<Long, CounterMark> unbounded(); @Deprecated static UnboundedSource<Long, CounterMark> unboundedWithTimestampFn(
      SerializableFunction<Long, Instant> timestampFn);  }
@Test @Category(NeedsRunner.class) public void testUnboundedSourceWithRate() { Duration period = Duration.millis(5); long numElements = 1000L; PCollection<Long> input = p.apply( Read.from( CountingSource.createUnboundedFrom(0) .withTimestampFn(new ValueAsTimestampFn()) .withRate(1, period)) .withMaxNumRecords(numElements)); addCountingAsserts(input, numElements); PCollection<Long> diffs = input .apply("TimestampDiff", ParDo.of(new ElementValueDiff())) .apply("DistinctTimestamps", Distinct.create()); PAssert.thatSingleton(diffs).isEqualTo(0L); Instant started = Instant.now(); p.run(); Instant finished = Instant.now(); Duration expectedDuration = period.multipliedBy((int) numElements); assertThat(started.plus(expectedDuration).isBefore(finished), is(true)); }
@Test @Category(NeedsRunner.class) public void testUnboundedSourceRateSplits() throws Exception { int elementsPerPeriod = 10; Duration period = Duration.millis(5); long numElements = 1000; int numSplits = 10; UnboundedCountingSource initial = CountingSource.createUnboundedFrom(0).withRate(elementsPerPeriod, period); List<? extends UnboundedSource<Long, ?>> splits = initial.split(numSplits, p.getOptions()); assertEquals("Expected exact splitting", numSplits, splits.size()); long elementsPerSplit = numElements / numSplits; assertEquals("Expected even splits", numElements, elementsPerSplit * numSplits); PCollectionList<Long> pcollections = PCollectionList.empty(p); for (int i = 0; i < splits.size(); ++i) { pcollections = pcollections.and( p.apply("split" + i, Read.from(splits.get(i)).withMaxNumRecords(elementsPerSplit))); } PCollection<Long> input = pcollections.apply(Flatten.pCollections()); addCountingAsserts(input, numElements); Instant startTime = Instant.now(); p.run(); Instant endTime = Instant.now(); long expectedMinimumMillis = (numElements * period.getMillis()) / elementsPerPeriod; assertThat(expectedMinimumMillis, lessThan(endTime.getMillis() - startTime.getMillis())); }