BeamFnDataGrpcClient implements BeamFnDataClient { @Override public <T> CloseableFnDataReceiver<T> send( Endpoints.ApiServiceDescriptor apiServiceDescriptor, LogicalEndpoint outputLocation, Coder<T> coder) { BeamFnDataGrpcMultiplexer client = getClientFor(apiServiceDescriptor); LOG.debug("Creating output consumer for {}", outputLocation); return BeamFnDataBufferingOutboundObserver.forLocation( options, outputLocation, coder, client.getOutboundObserver()); } BeamFnDataGrpcClient(
      PipelineOptions options,
      Function<Endpoints.ApiServiceDescriptor, ManagedChannel> channelFactory,
      OutboundObserverFactory outboundObserverFactory); @Override InboundDataClient receive(
      ApiServiceDescriptor apiServiceDescriptor,
      LogicalEndpoint inputLocation,
      Coder<T> coder,
      FnDataReceiver<T> consumer); @Override CloseableFnDataReceiver<T> send(
      Endpoints.ApiServiceDescriptor apiServiceDescriptor,
      LogicalEndpoint outputLocation,
      Coder<T> coder);  }
@Test public void testForOutboundConsumer() throws Exception { CountDownLatch waitForInboundServerValuesCompletion = new CountDownLatch(2); Collection<BeamFnApi.Elements> inboundServerValues = new ConcurrentLinkedQueue<>(); CallStreamObserver<BeamFnApi.Elements> inboundServerObserver = TestStreams.withOnNext( (BeamFnApi.Elements t) -> { inboundServerValues.add(t); waitForInboundServerValuesCompletion.countDown(); }) .build(); Endpoints.ApiServiceDescriptor apiServiceDescriptor = Endpoints.ApiServiceDescriptor.newBuilder() .setUrl(this.getClass().getName() + "-" + UUID.randomUUID().toString()) .build(); Server server = InProcessServerBuilder.forName(apiServiceDescriptor.getUrl()) .addService( new BeamFnDataGrpc.BeamFnDataImplBase() { @Override public StreamObserver<BeamFnApi.Elements> data( StreamObserver<BeamFnApi.Elements> outboundObserver) { return inboundServerObserver; } }) .build(); server.start(); try { ManagedChannel channel = InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build(); BeamFnDataGrpcClient clientFactory = new BeamFnDataGrpcClient( PipelineOptionsFactory.fromArgs( new String[] {"--experiments=data_buffer_size_limit=20"}) .create(), (Endpoints.ApiServiceDescriptor descriptor) -> channel, OutboundObserverFactory.trivial()); try (CloseableFnDataReceiver<WindowedValue<String>> consumer = clientFactory.send(apiServiceDescriptor, ENDPOINT_A, CODER)) { consumer.accept(valueInGlobalWindow("ABC")); consumer.accept(valueInGlobalWindow("DEF")); consumer.accept(valueInGlobalWindow("GHI")); } waitForInboundServerValuesCompletion.await(); assertThat(inboundServerValues, contains(ELEMENTS_A_1, ELEMENTS_A_2)); } finally { server.shutdownNow(); } }