BoundedSourceRunner { @Deprecated public void start() throws Exception { try { InputT boundedSource; if (definition.getUrn().equals(ProcessBundleHandler.JAVA_SOURCE_URN)) { byte[] bytes = definition.getPayload().toByteArray(); @SuppressWarnings("unchecked") InputT boundedSource0 = (InputT) SerializableUtils.deserializeFromByteArray(bytes, definition.toString()); boundedSource = boundedSource0; } else if (definition.getUrn().equals(PTransformTranslation.READ_TRANSFORM_URN)) { ReadPayload readPayload = ReadPayload.parseFrom(definition.getPayload()); boundedSource = (InputT) ReadTranslation.boundedSourceFromProto(readPayload); } else { throw new IllegalArgumentException("Unknown source URN: " + definition.getUrn()); } runReadLoop(WindowedValue.valueInGlobalWindow(boundedSource)); } catch (InvalidProtocolBufferException e) { throw new IOException(String.format("Failed to decode %s", definition.getUrn()), e); } } BoundedSourceRunner(
      PipelineOptions pipelineOptions,
      RunnerApi.FunctionSpec definition,
      Collection<FnDataReceiver<WindowedValue<OutputT>>> consumers); @Deprecated void start(); void runReadLoop(WindowedValue<InputT> value); @Override String toString();  }
@Test public void testStart() throws Exception { List<WindowedValue<Long>> outValues = new ArrayList<>(); Collection<FnDataReceiver<WindowedValue<Long>>> consumers = ImmutableList.of(outValues::add); ByteString encodedSource = ByteString.copyFrom(SerializableUtils.serializeToByteArray(CountingSource.upTo(3))); BoundedSourceRunner<BoundedSource<Long>, Long> runner = new BoundedSourceRunner<>( PipelineOptionsFactory.create(), RunnerApi.FunctionSpec.newBuilder() .setUrn(ProcessBundleHandler.JAVA_SOURCE_URN) .setPayload(encodedSource) .build(), consumers); runner.start(); assertThat( outValues, contains(valueInGlobalWindow(0L), valueInGlobalWindow(1L), valueInGlobalWindow(2L))); }