LengthPrefixUnknownCoders { @VisibleForTesting static Function<Node, Node> forInstructionOutputNode(MutableNetwork<Node, Edge> network) { return new TypeSafeNodeFunction<InstructionOutputNode>(InstructionOutputNode.class) { @Override public InstructionOutputNode typedApply(InstructionOutputNode input) { InstructionOutput cloudOutput = input.getInstructionOutput(); if (network.predecessors(input).stream().anyMatch(RemoteGrpcPortNode.class::isInstance) || network.successors(input).stream().anyMatch(RemoteGrpcPortNode.class::isInstance)) { try { cloudOutput = forInstructionOutput(cloudOutput, false ); } catch (Exception e) { throw new RuntimeException( String.format( "Failed to replace unknown coder with " + "LengthPrefixCoder for : {%s}", input.getInstructionOutput()), e); } } return InstructionOutputNode.create(cloudOutput, input.getPcollectionId()); } }; }  static MutableNetwork<Node, Edge> andReplaceForRunnerNetwork(
      MutableNetwork<Node, Edge> network); static MutableNetwork<Node, Edge> forSdkNetwork(MutableNetwork<Node, Edge> network); static List<SideInputInfo> forSideInputInfos(
      List<SideInputInfo> sideInputInfos, boolean replaceWithByteArrayCoder);  }
@Test public void testLengthPrefixForInstructionOutputNodeWithGrpcNodeSuccessor() { MutableNetwork<Node, Edge> network = createEmptyNetwork(); network.addNode(instructionOutputNode); network.addNode(grpcPortNode); network.addEdge(grpcPortNode, instructionOutputNode, DefaultEdge.create()); assertEqualsAsJson( CloudObjects.asCloudObject(prefixedWindowedValueCoder, null), ((InstructionOutputNode) forInstructionOutputNode(network).apply(instructionOutputNode)) .getInstructionOutput() .getCodec()); }
@Test public void testLengthPrefixForInstructionOutputNodeWithGrpcNodePredecessor() { MutableNetwork<Node, Edge> network = createEmptyNetwork(); network.addNode(instructionOutputNode); network.addNode(grpcPortNode); network.addEdge(instructionOutputNode, grpcPortNode, DefaultEdge.create()); assertEqualsAsJson( CloudObjects.asCloudObject(prefixedWindowedValueCoder, null), ((InstructionOutputNode) forInstructionOutputNode(network).apply(instructionOutputNode)) .getInstructionOutput() .getCodec()); }
@Test public void testLengthPrefixForInstructionOutputNodeWithNonGrpcNodeNeighbor() { MutableNetwork<Node, Edge> network = createEmptyNetwork(); ParallelInstructionNode readNode = createReadNode("read", "source", windowedValueCoder); network.addNode(instructionOutputNode); network.addNode(readNode); network.addEdge(readNode, instructionOutputNode, DefaultEdge.create()); assertEqualsAsJson( CloudObjects.asCloudObject(windowedValueCoder, null), ((InstructionOutputNode) forInstructionOutputNode(network).apply(instructionOutputNode)) .getInstructionOutput() .getCodec()); }