ImpulseEvaluatorFactory implements TransformEvaluatorFactory { @Nullable @Override public <InputT> TransformEvaluator<InputT> forApplication( AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) { return (TransformEvaluator<InputT>) new ImpulseEvaluator(ctxt, (AppliedPTransform) application); } ImpulseEvaluatorFactory(EvaluationContext ctxt); @Nullable @Override TransformEvaluator<InputT> forApplication(
      AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle); @Override void cleanup();  }
@Test public void testImpulse() throws Exception { Pipeline p = Pipeline.create(); PCollection<byte[]> impulseOut = p.apply(Impulse.create()); AppliedPTransform<?, ?, ?> impulseApplication = DirectGraphs.getProducer(impulseOut); ImpulseEvaluatorFactory factory = new ImpulseEvaluatorFactory(context); WindowedValue<ImpulseShard> inputShard = WindowedValue.valueInGlobalWindow(new ImpulseShard()); CommittedBundle<ImpulseShard> inputShardBundle = bundleFactory.<ImpulseShard>createRootBundle().add(inputShard).commit(Instant.now()); when(context.createBundle(impulseOut)).thenReturn(bundleFactory.createBundle(impulseOut)); TransformEvaluator<ImpulseShard> evaluator = factory.forApplication(impulseApplication, inputShardBundle); evaluator.processElement(inputShard); TransformResult<ImpulseShard> result = evaluator.finishBundle(); assertThat( "Exactly one output from a single ImpulseShard", Iterables.size(result.getOutputBundles()), equalTo(1)); UncommittedBundle<?> outputBundle = result.getOutputBundles().iterator().next(); CommittedBundle<?> committedOutputBundle = outputBundle.commit(Instant.now()); assertThat( committedOutputBundle.getMinimumTimestamp(), equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE)); assertThat(committedOutputBundle.getPCollection(), equalTo(impulseOut)); assertThat( "Should only be one impulse element", Iterables.size(committedOutputBundle.getElements()), equalTo(1)); assertThat( committedOutputBundle.getElements().iterator().next().getWindows(), contains(GlobalWindow.INSTANCE)); assertArrayEquals( "Output should be an empty byte array", new byte[0], (byte[]) committedOutputBundle.getElements().iterator().next().getValue()); }