// handover to queue, do not block the grpc thread
private void processElementWithSdkHarness(WindowedValue<InputT> element) throws Exception {
    checkState(
        stageBundleFactory != null, "%s not yet prepared", StageBundleFactory.class.getName());
    checkState(
        stateRequestHandler != null, "%s not yet prepared", StateRequestHandler.class.getName());
    OutputReceiverFactory receiverFactory =
        new OutputReceiverFactory() {
          @Override
          public FnDataReceiver<OutputT> create(String pCollectionId) {
            return (receivedElement) -> {
              // handover to queue, do not block the grpc thread
              outputQueue.put(KV.of(pCollectionId, receivedElement));
            };
          }
        };
    try (RemoteBundle bundle =
        stageBundleFactory.getBundle(receiverFactory, stateRequestHandler, progressHandler)) {
      logger.debug(String.format("Sending value: %s", element));
      // TODO(BEAM-4681): Add support to Flink to support portable timers.
      Iterables.getOnlyElement(bundle.getInputReceivers().values()).accept(element);
      // TODO: it would be nice to emit results as they arrive, can thread wait non-blocking?
    }
    // RemoteBundle close blocks until all results are received
    KV<String, OutputT> result;
    while ((result = outputQueue.poll()) != null) {
      outputManager.output(outputMap.get(result.getKey()), (WindowedValue) result.getValue());
    }
  }
public FnDataReceiver<OutputT> create(String pCollectionId) {
return (receivedElement) -> {
// handover to queue, do not block the grpc thread
outputQueue.put(KV.of(pCollectionId, receivedElement));
};
private void processElementWithSdkHarness(WindowedValue<InputT> element) throws Exception {
checkState(
stageBundleFactory != null, "%s not yet prepared", StageBundleFactory.class.getName());
checkState(
stateRequestHandler != null, "%s not yet prepared", StateRequestHandler.class.getName());
OutputReceiverFactory receiverFactory =
new OutputReceiverFactory() {
@Override
public FnDataReceiver<OutputT> create(String pCollectionId) {
return (receivedElement) -> {
// handover to queue, do not block the grpc thread
outputQueue.put(KV.of(pCollectionId, receivedElement));
};
}
};
try (RemoteBundle bundle =
stageBundleFactory.getBundle(receiverFactory, stateRequestHandler, progressHandler)) {
logger.debug(String.format("Sending value: %s", element));
// TODO(BEAM-4681): Add support to Flink to support portable timers.
Iterables.getOnlyElement(bundle.getInputReceivers().values()).accept(element);
// TODO: it would be nice to emit results as they arrive, can thread wait non-blocking?
private void processElementWithSdkHarness(WindowedValue<InputT> element) throws Exception {
checkState(
stageBundleFactory != null, "%s not yet prepared", StageBundleFactory.class.getName());
checkState(
stateRequestHandler != null, "%s not yet prepared", StateRequestHandler.class.getName());
OutputReceiverFactory receiverFactory =
new OutputReceiverFactory() {
@Override
public FnDataReceiver<OutputT> create(String pCollectionId) {
return (receivedElement) -> {
// handover to queue, do not block the grpc thread
outputQueue.put(KV.of(pCollectionId, receivedElement));
};
}
};
try (RemoteBundle bundle =
stageBundleFactory.getBundle(receiverFactory, stateRequestHandler, progressHandler)) {
logger.debug(String.format("Sending value: %s", element));
// TODO(BEAM-4681): Add support to Flink to support portable timers.
Iterables.getOnlyElement(bundle.getInputReceivers().values()).accept(element);
// TODO: it would be nice to emit results as they arrive, can thread wait non-blocking?
}
// RemoteBundle close blocks until all results are received
KV<String, OutputT> result;
while ((result = outputQueue.poll()) != null) {
outputManager.output(outputMap.get(result.getKey()), (WindowedValue) result.getValue());
}
}