@Test public void testStubUri() { final List<String> input = Streams.stream(Iterables.concat( VALID_V6_CASES.keySet(), VALID_V7_CASES.keySet(), VALID_V8_CASES.keySet(), Arrays.asList( "v6/" + String.join("/", Collections.nCopies(35, " ")), "v7/" + String.join("/", Collections.nCopies(36, " ")), "v8/" + String.join("/", Collections.nCopies(38, " ")), "", "v6", "v7", "v8"))) .map(v -> "{\"attributeMap\":{\"message_id\":\"00000000001\",\"uri\":\"/stub/" + v + "\"},\"payload\":\"\"}") .collect(Collectors.toList()); final List<String> expectedPayloads = Streams.stream(Iterables.concat( VALID_V6_CASES.values(), VALID_V7_CASES.values(), VALID_V8_CASES.values())) .map(v -> sortJSON("{\"installer_type\":\"stub\",\"installer_version\":\"\"," + v + "}")) .collect(Collectors.toList()); final List<String> expectedAttributes = Collections.nCopies(expectedPayloads.size(), "{\"document_id\":\"d67f0826-d4c0-3a7e-bea5-861616a822b2\"," + "\"document_namespace\":\"firefox-installer\"," + "\"document_type\":\"install\",\"document_version\":\"1\"}"); final List<String> expectedExceptions = Arrays.asList( "com.mozilla.telemetry.decoder.ParseUri$StubUri$InvalidIntegerException", "com.mozilla.telemetry.decoder.ParseUri$StubUri$InvalidIntegerException", "com.mozilla.telemetry.decoder.ParseUri$StubUri$InvalidIntegerException", "com.mozilla.telemetry.decoder.ParseUri$StubUri$UnknownPingVersionException", "com.mozilla.telemetry.decoder.ParseUri$UnexpectedPathElementsException", "com.mozilla.telemetry.decoder.ParseUri$UnexpectedPathElementsException", "com.mozilla.telemetry.decoder.ParseUri$UnexpectedPathElementsException"); WithFailures.Result<PCollection<PubsubMessage>, PubsubMessage> parsed = pipeline .apply(Create.of(input)).apply("DecodeJsonInput", InputFileFormat.json.decode()) .apply(ParseUri.of()); PCollection<PubsubMessage> output = parsed.output(); PCollection<String> payloads = output .apply("PayloadString", MapElements.into(TypeDescriptors.strings()).via(message -> { return sortJSON(new String(message.getPayload(), StandardCharsets.UTF_8)); })); PAssert.that(payloads).containsInAnyOrder(expectedPayloads); PCollection<String> attributesSansUri = output .apply("AttributesSansUri", MapElements.into(TypeDescriptors.strings()).via(message -> { Map<String, String> attributes = new HashMap<>(message.getAttributeMap()); attributes.remove("uri"); try { return Json.asString(attributes); } catch (IOException e) { throw new UncheckedIOException(e); } })); PAssert.that(attributesSansUri).containsInAnyOrder(expectedAttributes); PCollection<String> uriExceptions = parsed.failures() .apply("ExceptionClass", MapElements.into(TypeDescriptors.strings()) .via(message -> message.getAttribute("exception_class"))); PAssert.that(uriExceptions).containsInAnyOrder(expectedExceptions); ValueProvider<String> schemas = pipeline.newProvider("schemas.tar.gz"); PCollection<String> schemaExceptions = output.apply(ParsePayload.of(schemas)).failures() .apply("EncodeJsonErrors", OutputFileFormat.json.encode()); PAssert.that(schemaExceptions).empty(); 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; }