@Test public void testOutput() { final List<String> validInput = Arrays.asList( "{\"attributeMap\":{\"document_namespace\":\"telemetry\"" + ",\"document_type\": \"xfocsp-error-report\"},\"payload\":\"\"}", "{\"attributeMap\":{\"document_namespace\":\"telemetry\"" + ",\"document_type\": \"main\"},\"payload\":\"\"}"); final List<String> expected = Arrays.asList( "{\"attributeMap\":{\"document_namespace\":\"xfocsp-error-report\"" + ",\"document_type\":\"xfocsp-error-report\"},\"payload\":\"\"}", "{\"attributeMap\":{\"document_namespace\":\"telemetry\"" + ",\"document_type\":\"main\"},\"payload\":\"\"}"); PCollection<PubsubMessage> result = pipeline.apply(Create.of(validInput)) .apply("DecodeJsonInput", InputFileFormat.json.decode()) .apply("RerouteDocuments", RerouteDocuments.of()); PCollection<String> output = result .apply("EncodeJsonOutput", OutputFileFormat.json.encode()); PAssert.that(output).containsInAnyOrder(expected); pipeline.run(); }
public static PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage m) -> { if (ParseUri.TELEMETRY.equals(m.getAttribute(Attribute.DOCUMENT_NAMESPACE)) && XFOCSP_ERROR_REPORT.equals(m.getAttribute(Attribute.DOCUMENT_TYPE))) { Map<String, String> attributes = new HashMap<>(m.getAttributeMap()); attributes.put(Attribute.DOCUMENT_NAMESPACE, XFOCSP_ERROR_REPORT); return new PubsubMessage(m.getPayload(), attributes); } else { return m; } }); }
RerouteDocuments { public static PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage m) -> { if (ParseUri.TELEMETRY.equals(m.getAttribute(Attribute.DOCUMENT_NAMESPACE)) && XFOCSP_ERROR_REPORT.equals(m.getAttribute(Attribute.DOCUMENT_TYPE))) { Map<String, String> attributes = new HashMap<>(m.getAttributeMap()); attributes.put(Attribute.DOCUMENT_NAMESPACE, XFOCSP_ERROR_REPORT); return new PubsubMessage(m.getPayload(), attributes); } else { return m; } }); } }
RerouteDocuments { public static PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage m) -> { if (ParseUri.TELEMETRY.equals(m.getAttribute(Attribute.DOCUMENT_NAMESPACE)) && XFOCSP_ERROR_REPORT.equals(m.getAttribute(Attribute.DOCUMENT_TYPE))) { Map<String, String> attributes = new HashMap<>(m.getAttributeMap()); attributes.put(Attribute.DOCUMENT_NAMESPACE, XFOCSP_ERROR_REPORT); return new PubsubMessage(m.getPayload(), attributes); } else { return m; } }); }  }
RerouteDocuments { public static PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage m) -> { if (ParseUri.TELEMETRY.equals(m.getAttribute(Attribute.DOCUMENT_NAMESPACE)) && XFOCSP_ERROR_REPORT.equals(m.getAttribute(Attribute.DOCUMENT_TYPE))) { Map<String, String> attributes = new HashMap<>(m.getAttributeMap()); attributes.put(Attribute.DOCUMENT_NAMESPACE, XFOCSP_ERROR_REPORT); return new PubsubMessage(m.getPayload(), attributes); } else { return m; } }); }  static PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> of(); }
RerouteDocuments { public static PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage m) -> { if (ParseUri.TELEMETRY.equals(m.getAttribute(Attribute.DOCUMENT_NAMESPACE)) && XFOCSP_ERROR_REPORT.equals(m.getAttribute(Attribute.DOCUMENT_TYPE))) { Map<String, String> attributes = new HashMap<>(m.getAttributeMap()); attributes.put(Attribute.DOCUMENT_NAMESPACE, XFOCSP_ERROR_REPORT); return new PubsubMessage(m.getPayload(), attributes); } else { return m; } }); }  static PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> of();  }