@Test public void testOutput() { final List<String> validInput = Arrays.asList( "{\"attributeMap\":{},\"payload\":\"\"}", "{\"attributeMap\":{\"uri\":\"/submit/telemetry/ce39b608-f595-4c69-b6a6-f7a436604648" + "/main/Firefox/61.0a1/nightly/20180328030202\"" + "},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"uri\":\"/submit/eng-workflow/hgpush/1/2c3a0767-d84a-4d02-8a92-fa54a3376049\"" + "},\"payload\":\"\"}"); final List<String> invalidInput = Arrays.asList( "{\"attributeMap\":" + "{\"uri\":\"/nonexistent_prefix/ce39b608-f595-4c69-b6a6-f7a436604648" + "/main/Firefox/61.0a1/nightly/20180328030202\"" + "},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"uri\":\"/submit/telemetry/ce39b608-f595-4c69-b6a6-f7a436604648" + "/Firefox/61.0a1/nightly/20180328030202\"" + "},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"uri\":\"/submit/eng-workflow/hgpush/2c3a0767-d84a-4d02-8a92-fa54a3376049\"" + "},\"payload\":\"\"}"); final List<String> expected = Arrays.asList( "{\"attributeMap\":{},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"app_build_id\":\"20180328030202\"" + ",\"app_name\":\"Firefox\"" + ",\"app_update_channel\":\"nightly\"" + ",\"app_version\":\"61.0a1\"" + ",\"document_id\":\"ce39b608-f595-4c69-b6a6-f7a436604648\"" + ",\"document_namespace\":\"telemetry\"" + ",\"document_type\":\"main\"" + ",\"uri\":\"/submit/telemetry/ce39b608-f595-4c69-b6a6-f7a436604648" + "/main/Firefox/61.0a1/nightly/20180328030202\"" + "},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"document_id\":\"2c3a0767-d84a-4d02-8a92-fa54a3376049\"" + ",\"document_namespace\":\"eng-workflow\"" + ",\"document_type\":\"hgpush\"" + ",\"document_version\":\"1\"" + ",\"uri\":\"/submit/eng-workflow/hgpush/1/2c3a0767-d84a-4d02-8a92-fa54a3376049\"" + "},\"payload\":\"\"}"); WithFailures.Result<PCollection<PubsubMessage>, PubsubMessage> parsed = pipeline .apply(Create.of(Iterables.concat(validInput, invalidInput))) .apply("DecodeJsonInput", InputFileFormat.json.decode()) .apply(ParseUri.of()); PCollection<String> output = parsed.output() .apply("EncodeJsonOutput", OutputFileFormat.json.encode()); PAssert.that(output).containsInAnyOrder(expected); PCollection<String> exceptions = parsed.failures() .apply(MapElements.into(TypeDescriptors.strings()) .via(message -> message.getAttribute("exception_class"))); PAssert.that(exceptions).containsInAnyOrder( Arrays.asList("com.mozilla.telemetry.decoder.ParseUri$InvalidUriException", "com.mozilla.telemetry.decoder.ParseUri$UnexpectedPathElementsException", "com.mozilla.telemetry.decoder.ParseUri$UnexpectedPathElementsException")); pipeline.run(); }
public static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage message) -> { message = PubsubConstraints.ensureNonNull(message); final Map<String, String> attributes = new HashMap<>(message.getAttributeMap()); byte[] payload = message.getPayload(); final String uri = attributes.get(Attribute.URI); if (uri == null) { return message; } else if (uri.startsWith(TELEMETRY_URI_PREFIX)) { attributes.put(Attribute.DOCUMENT_NAMESPACE, TELEMETRY); attributes.putAll(zip(TELEMETRY_URI_SUFFIX_ELEMENTS, uri.substring(TELEMETRY_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(GENERIC_URI_PREFIX)) { attributes.putAll(zip(GENERIC_URI_SUFFIX_ELEMENTS, uri.substring(GENERIC_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(StubUri.PREFIX)) { try { payload = StubUri.parse(uri, attributes); } catch (IOException e) { throw new UncheckedIOException(e); } if (attributes.get(Attribute.MESSAGE_ID) != null) { UUID documentId = UUID.nameUUIDFromBytes( attributes.get(Attribute.MESSAGE_ID).getBytes(StandardCharsets.UTF_8)); attributes.put(Attribute.DOCUMENT_ID, documentId.toString().toLowerCase()); } } else { throw new InvalidUriException("Unknown URI prefix"); } attributes.remove(Attribute.MESSAGE_ID); return new PubsubMessage(payload, attributes); }).exceptionsInto(TypeDescriptor.of(PubsubMessage.class)) .exceptionsVia((WithFailures.ExceptionElement<PubsubMessage> ee) -> { try { throw ee.exception(); } catch (UncheckedIOException | InvalidUriException e) { return FailureMessage.of(ParseUri.class.getSimpleName(), ee.element(), ee.exception()); } }); }
ParseUri { public static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage message) -> { message = PubsubConstraints.ensureNonNull(message); final Map<String, String> attributes = new HashMap<>(message.getAttributeMap()); byte[] payload = message.getPayload(); final String uri = attributes.get(Attribute.URI); if (uri == null) { return message; } else if (uri.startsWith(TELEMETRY_URI_PREFIX)) { attributes.put(Attribute.DOCUMENT_NAMESPACE, TELEMETRY); attributes.putAll(zip(TELEMETRY_URI_SUFFIX_ELEMENTS, uri.substring(TELEMETRY_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(GENERIC_URI_PREFIX)) { attributes.putAll(zip(GENERIC_URI_SUFFIX_ELEMENTS, uri.substring(GENERIC_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(StubUri.PREFIX)) { try { payload = StubUri.parse(uri, attributes); } catch (IOException e) { throw new UncheckedIOException(e); } if (attributes.get(Attribute.MESSAGE_ID) != null) { UUID documentId = UUID.nameUUIDFromBytes( attributes.get(Attribute.MESSAGE_ID).getBytes(StandardCharsets.UTF_8)); attributes.put(Attribute.DOCUMENT_ID, documentId.toString().toLowerCase()); } } else { throw new InvalidUriException("Unknown URI prefix"); } attributes.remove(Attribute.MESSAGE_ID); return new PubsubMessage(payload, attributes); }).exceptionsInto(TypeDescriptor.of(PubsubMessage.class)) .exceptionsVia((WithFailures.ExceptionElement<PubsubMessage> ee) -> { try { throw ee.exception(); } catch (UncheckedIOException | InvalidUriException e) { return FailureMessage.of(ParseUri.class.getSimpleName(), ee.element(), ee.exception()); } }); } }
ParseUri { public static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage message) -> { message = PubsubConstraints.ensureNonNull(message); final Map<String, String> attributes = new HashMap<>(message.getAttributeMap()); byte[] payload = message.getPayload(); final String uri = attributes.get(Attribute.URI); if (uri == null) { return message; } else if (uri.startsWith(TELEMETRY_URI_PREFIX)) { attributes.put(Attribute.DOCUMENT_NAMESPACE, TELEMETRY); attributes.putAll(zip(TELEMETRY_URI_SUFFIX_ELEMENTS, uri.substring(TELEMETRY_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(GENERIC_URI_PREFIX)) { attributes.putAll(zip(GENERIC_URI_SUFFIX_ELEMENTS, uri.substring(GENERIC_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(StubUri.PREFIX)) { try { payload = StubUri.parse(uri, attributes); } catch (IOException e) { throw new UncheckedIOException(e); } if (attributes.get(Attribute.MESSAGE_ID) != null) { UUID documentId = UUID.nameUUIDFromBytes( attributes.get(Attribute.MESSAGE_ID).getBytes(StandardCharsets.UTF_8)); attributes.put(Attribute.DOCUMENT_ID, documentId.toString().toLowerCase()); } } else { throw new InvalidUriException("Unknown URI prefix"); } attributes.remove(Attribute.MESSAGE_ID); return new PubsubMessage(payload, attributes); }).exceptionsInto(TypeDescriptor.of(PubsubMessage.class)) .exceptionsVia((WithFailures.ExceptionElement<PubsubMessage> ee) -> { try { throw ee.exception(); } catch (UncheckedIOException | InvalidUriException e) { return FailureMessage.of(ParseUri.class.getSimpleName(), ee.element(), ee.exception()); } }); }  }
ParseUri { public static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage message) -> { message = PubsubConstraints.ensureNonNull(message); final Map<String, String> attributes = new HashMap<>(message.getAttributeMap()); byte[] payload = message.getPayload(); final String uri = attributes.get(Attribute.URI); if (uri == null) { return message; } else if (uri.startsWith(TELEMETRY_URI_PREFIX)) { attributes.put(Attribute.DOCUMENT_NAMESPACE, TELEMETRY); attributes.putAll(zip(TELEMETRY_URI_SUFFIX_ELEMENTS, uri.substring(TELEMETRY_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(GENERIC_URI_PREFIX)) { attributes.putAll(zip(GENERIC_URI_SUFFIX_ELEMENTS, uri.substring(GENERIC_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(StubUri.PREFIX)) { try { payload = StubUri.parse(uri, attributes); } catch (IOException e) { throw new UncheckedIOException(e); } if (attributes.get(Attribute.MESSAGE_ID) != null) { UUID documentId = UUID.nameUUIDFromBytes( attributes.get(Attribute.MESSAGE_ID).getBytes(StandardCharsets.UTF_8)); attributes.put(Attribute.DOCUMENT_ID, documentId.toString().toLowerCase()); } } else { throw new InvalidUriException("Unknown URI prefix"); } attributes.remove(Attribute.MESSAGE_ID); return new PubsubMessage(payload, attributes); }).exceptionsInto(TypeDescriptor.of(PubsubMessage.class)) .exceptionsVia((WithFailures.ExceptionElement<PubsubMessage> ee) -> { try { throw ee.exception(); } catch (UncheckedIOException | InvalidUriException e) { return FailureMessage.of(ParseUri.class.getSimpleName(), ee.element(), ee.exception()); } }); }  static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of(); }
ParseUri { public static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of() { return MapElements.into(TypeDescriptor.of(PubsubMessage.class)).via((PubsubMessage message) -> { message = PubsubConstraints.ensureNonNull(message); final Map<String, String> attributes = new HashMap<>(message.getAttributeMap()); byte[] payload = message.getPayload(); final String uri = attributes.get(Attribute.URI); if (uri == null) { return message; } else if (uri.startsWith(TELEMETRY_URI_PREFIX)) { attributes.put(Attribute.DOCUMENT_NAMESPACE, TELEMETRY); attributes.putAll(zip(TELEMETRY_URI_SUFFIX_ELEMENTS, uri.substring(TELEMETRY_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(GENERIC_URI_PREFIX)) { attributes.putAll(zip(GENERIC_URI_SUFFIX_ELEMENTS, uri.substring(GENERIC_URI_PREFIX.length()).split("/"))); } else if (uri.startsWith(StubUri.PREFIX)) { try { payload = StubUri.parse(uri, attributes); } catch (IOException e) { throw new UncheckedIOException(e); } if (attributes.get(Attribute.MESSAGE_ID) != null) { UUID documentId = UUID.nameUUIDFromBytes( attributes.get(Attribute.MESSAGE_ID).getBytes(StandardCharsets.UTF_8)); attributes.put(Attribute.DOCUMENT_ID, documentId.toString().toLowerCase()); } } else { throw new InvalidUriException("Unknown URI prefix"); } attributes.remove(Attribute.MESSAGE_ID); return new PubsubMessage(payload, attributes); }).exceptionsInto(TypeDescriptor.of(PubsubMessage.class)) .exceptionsVia((WithFailures.ExceptionElement<PubsubMessage> ee) -> { try { throw ee.exception(); } catch (UncheckedIOException | InvalidUriException e) { return FailureMessage.of(ParseUri.class.getSimpleName(), ee.element(), ee.exception()); } }); }  static MapWithFailures<PubsubMessage, PubsubMessage, PubsubMessage> of(); static final String TELEMETRY; static final String TELEMETRY_URI_PREFIX; static final String[] TELEMETRY_URI_SUFFIX_ELEMENTS; static final String GENERIC_URI_PREFIX; static final String[] GENERIC_URI_SUFFIX_ELEMENTS; }