LengthPrefixUnknownCoders { @VisibleForTesting static Map<String, Object> forCodec( Map<String, Object> codec, boolean replaceWithByteArrayCoder) { String coderType = (String) codec.get(PropertyNames.OBJECT_TYPE_NAME); if (LENGTH_PREFIX_CODER_TYPE.equals(coderType)) { if (replaceWithByteArrayCoder) { return CloudObjects.asCloudObject( LENGTH_PREFIXED_BYTE_ARRAY_CODER, null); } return codec; } else if (WELL_KNOWN_CODER_TYPES.contains(coderType)) { Map<String, Object> prefixedCodec = new HashMap<>(codec); if (codec.containsKey(PropertyNames.COMPONENT_ENCODINGS)) { List<Map<String, Object>> prefixedComponents = new ArrayList<>(); for (Map<String, Object> component : (Iterable<Map<String, Object>>) codec.get(PropertyNames.COMPONENT_ENCODINGS)) { prefixedComponents.add(forCodec(component, replaceWithByteArrayCoder)); } prefixedCodec.put(PropertyNames.COMPONENT_ENCODINGS, prefixedComponents); } return prefixedCodec; } if (replaceWithByteArrayCoder) { return CloudObjects.asCloudObject(LENGTH_PREFIXED_BYTE_ARRAY_CODER, null); } else { Map<String, Object> prefixedCodec = new HashMap<>(); prefixedCodec.put(PropertyNames.OBJECT_TYPE_NAME, LENGTH_PREFIX_CODER_TYPE); prefixedCodec.put(PropertyNames.COMPONENT_ENCODINGS, ImmutableList.of(codec)); return prefixedCodec; } }  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 testLengthPrefixUnknownCoders() throws Exception { Map<String, Object> lengthPrefixedCoderCloudObject = forCodec(CloudObjects.asCloudObject(windowedValueCoder, null), false); assertEqualsAsJson( CloudObjects.asCloudObject(prefixedWindowedValueCoder, null), lengthPrefixedCoderCloudObject); }
@Test public void testLengthPrefixForLengthPrefixCoder() throws Exception { Coder<WindowedValue<KV<String, Integer>>> windowedValueCoder = WindowedValue.getFullCoder( KvCoder.of(StringUtf8Coder.of(), LengthPrefixCoder.of(VarIntCoder.of())), GlobalWindow.Coder.INSTANCE); Map<String, Object> lengthPrefixedCoderCloudObject = forCodec(CloudObjects.asCloudObject(windowedValueCoder, null), false); Coder<WindowedValue<KV<String, Integer>>> expectedCoder = WindowedValue.getFullCoder( KvCoder.of(StringUtf8Coder.of(), LengthPrefixCoder.of(VarIntCoder.of())), GlobalWindow.Coder.INSTANCE); assertEqualsAsJson( CloudObjects.asCloudObject(expectedCoder, null), lengthPrefixedCoderCloudObject); }
@Test public void testLengthPrefixAndReplaceUnknownCoder() throws Exception { Coder<WindowedValue<KV<String, Integer>>> windowedValueCoder = WindowedValue.getFullCoder( KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()), GlobalWindow.Coder.INSTANCE); Map<String, Object> lengthPrefixedCoderCloudObject = forCodec(CloudObjects.asCloudObject(windowedValueCoder, null), true); assertEqualsAsJson( CloudObjects.asCloudObject(prefixedAndReplacedWindowedValueCoder, null), lengthPrefixedCoderCloudObject); }