Refreshing PCollectionView by polling view registry

usage.java

// view creation
PCollectionView<List<MaterializedViewGenerator>> viewGenerators = input
                .apply(new ViewRegistryPollerTransform(opts, windowDuration))
                .apply(View.asList());

// view usage as side input
 input.apply(name, ParDo.of(new ViewMaterializerDoFn(viewGenerators)).withSideInputs(viewGenerators));

ViewRegistryPollerTransform.java

public class ViewRegistryPollerTransform extends PTransform<PCollection<Row>, PCollection<MaterializedViewGenerator>> {

    // configuration
    private AnalyticsPipelineOptions opts;
    private Duration bucketDuration;

    public ViewRegistryPollerTransform(AnalyticsPipelineOptions opts, Duration bucketDuration) {
        this.opts = opts;
        this.bucketDuration = bucketDuration;
    }

    @Override
    public PCollection<MaterializedViewGenerator> apply(PCollection<Row> input) {
      // prepare ticks to refresh the view periodically
      PCollection<Long> ticks = input
                .getPipeline()
                // Produce 1 "tick" per second
                .apply("UnboundedTicks", CountingInput.unbounded().withRate(1, Duration.standardSeconds(1)))
                // Window the ticks into 1-hr windows
                .apply("1HrWindow", Window.into(FixedWindows.of(Duration.standardHours(1))))
                // Use an arbitrary per-window combiner to reduce to 1 element per window
                .apply(Count.globally());

        return ticks.apply("PollViewRegistryService", 
                           ParDo.of(new ListMaterializedViewRegistryFn(this.opts, this.bucketDuration)));
    }
}