PubsubIO { public static Read<String> readStrings() { return Read.newBuilder( (PubsubMessage message) -> new String(message.getPayload(), StandardCharsets.UTF_8)) .setCoder(StringUtf8Coder.of()) .build(); } private  PubsubIO(); static Read<PubsubMessage> readMessages(); static Read<PubsubMessage> readMessagesWithMessageId(); static Read<PubsubMessage> readMessagesWithAttributes(); static Read<PubsubMessage> readMessagesWithAttributesAndMessageId(); static Read<String> readStrings(); static Read<T> readProtos(Class<T> messageClass); static Read<T> readAvros(Class<T> clazz); static Read<T> readMessagesWithCoderAndParseFn(
      Coder<T> coder, SimpleFunction<PubsubMessage, T> parseFn); @Experimental(Kind.SCHEMAS) static Read<GenericRecord> readAvroGenericRecords(org.apache.avro.Schema avroSchema); @Experimental(Kind.SCHEMAS) static Read<T> readAvrosWithBeamSchema(Class<T> clazz); static Write<PubsubMessage> writeMessages(); static Write<String> writeStrings(); static Write<T> writeProtos(Class<T> messageClass); static Write<T> writeAvros(Class<T> clazz);  }
@Test public void testTopicValidationSuccess() throws Exception { PubsubIO.readStrings().fromTopic("projects/my-project/topics/abc"); PubsubIO.readStrings().fromTopic("projects/my-project/topics/ABC"); PubsubIO.readStrings().fromTopic("projects/my-project/topics/AbC-DeF"); PubsubIO.readStrings().fromTopic("projects/my-project/topics/AbC-1234"); PubsubIO.readStrings().fromTopic("projects/my-project/topics/AbC-1234-_.~%+-_.~%+-_.~%+-abc"); PubsubIO.readStrings() .fromTopic( new StringBuilder() .append("projects/my-project/topics/A-really-long-one-") .append( "111111111111111111111111111111111111111111111111111111111111111111111111111111111") .append( "111111111111111111111111111111111111111111111111111111111111111111111111111111111") .append( "11111111111111111111111111111111111111111111111111111111111111111111111111") .toString()); }
@Test public void testTopicValidationBadCharacter() throws Exception { thrown.expect(IllegalArgumentException.class); PubsubIO.readStrings().fromTopic("projects/my-project/topics/abc-*-abc"); }
@Test public void testTopicValidationTooLong() throws Exception { thrown.expect(IllegalArgumentException.class); PubsubIO.readStrings() .fromTopic( new StringBuilder() .append("projects/my-project/topics/A-really-long-one-") .append( "111111111111111111111111111111111111111111111111111111111111111111111111111111111") .append( "111111111111111111111111111111111111111111111111111111111111111111111111111111111") .append( "1111111111111111111111111111111111111111111111111111111111111111111111111111") .toString()); }
@Test public void testReadTopicDisplayData() { String topic = "projects/project/topics/topic"; PubsubIO.Read<String> read = PubsubIO.readStrings() .fromTopic(StaticValueProvider.of(topic)) .withTimestampAttribute("myTimestamp") .withIdAttribute("myId"); DisplayData displayData = DisplayData.from(read); assertThat(displayData, hasDisplayItem("topic", topic)); assertThat(displayData, hasDisplayItem("timestampAttribute", "myTimestamp")); assertThat(displayData, hasDisplayItem("idAttribute", "myId")); }
@Test public void testReadSubscriptionDisplayData() { String subscription = "projects/project/subscriptions/subscription"; PubsubIO.Read<String> read = PubsubIO.readStrings() .fromSubscription(StaticValueProvider.of(subscription)) .withTimestampAttribute("myTimestamp") .withIdAttribute("myId"); DisplayData displayData = DisplayData.from(read); assertThat(displayData, hasDisplayItem("subscription", subscription)); assertThat(displayData, hasDisplayItem("timestampAttribute", "myTimestamp")); assertThat(displayData, hasDisplayItem("idAttribute", "myId")); }
@Test public void testNullTopic() { String subscription = "projects/project/subscriptions/subscription"; PubsubIO.Read<String> read = PubsubIO.readStrings().fromSubscription(StaticValueProvider.of(subscription)); assertNull(read.getTopicProvider()); assertNotNull(read.getSubscriptionProvider()); assertNotNull(DisplayData.from(read)); }
@Test public void testNullSubscription() { String topic = "projects/project/topics/topic"; PubsubIO.Read<String> read = PubsubIO.readStrings().fromTopic(StaticValueProvider.of(topic)); assertNotNull(read.getTopicProvider()); assertNull(read.getSubscriptionProvider()); assertNotNull(DisplayData.from(read)); }
@Test public void testValueProviderSubscription() { StaticValueProvider<String> provider = StaticValueProvider.of("projects/project/subscriptions/subscription"); Read<String> pubsubRead = PubsubIO.readStrings().fromSubscription(provider); Pipeline.create().apply(pubsubRead); assertThat(pubsubRead.getSubscriptionProvider(), not(nullValue())); assertThat(pubsubRead.getSubscriptionProvider().isAccessible(), is(true)); assertThat(pubsubRead.getSubscriptionProvider().get().asPath(), equalTo(provider.get())); }
@Test public void testRuntimeValueProviderSubscription() { TestPipeline pipeline = TestPipeline.create(); ValueProvider<String> subscription = pipeline.newProvider("projects/project/subscriptions/subscription"); Read<String> pubsubRead = PubsubIO.readStrings().fromSubscription(subscription); pipeline.apply(pubsubRead); assertThat(pubsubRead.getSubscriptionProvider(), not(nullValue())); assertThat(pubsubRead.getSubscriptionProvider().isAccessible(), is(false)); }
@Test public void testValueProviderTopic() { StaticValueProvider<String> provider = StaticValueProvider.of("projects/project/topics/topic"); Read<String> pubsubRead = PubsubIO.readStrings().fromTopic(provider); Pipeline.create().apply(pubsubRead); assertThat(pubsubRead.getTopicProvider(), not(nullValue())); assertThat(pubsubRead.getTopicProvider().isAccessible(), is(true)); assertThat(pubsubRead.getTopicProvider().get().asPath(), equalTo(provider.get())); }
@Test public void testRuntimeValueProviderTopic() { TestPipeline pipeline = TestPipeline.create(); ValueProvider<String> topic = pipeline.newProvider("projects/project/topics/topic"); Read<String> pubsubRead = PubsubIO.readStrings().fromTopic(topic); pipeline.apply(pubsubRead); assertThat(pubsubRead.getTopicProvider(), not(nullValue())); assertThat(pubsubRead.getTopicProvider().isAccessible(), is(false)); }
@Test public void testPrimitiveReadDisplayData() { DisplayDataEvaluator evaluator = DisplayDataEvaluator.create(); Set<DisplayData> displayData; PubsubIO.Read<String> baseRead = PubsubIO.readStrings(); PubsubIO.Read<String> read = baseRead.fromSubscription("projects/project/subscriptions/subscription"); displayData = evaluator.displayDataForPrimitiveSourceTransforms(read); assertThat( "PubsubIO.Read should include the subscription in its primitive display data", displayData, hasItem(hasDisplayItem("subscription"))); read = baseRead.fromTopic("projects/project/topics/topic"); displayData = evaluator.displayDataForPrimitiveSourceTransforms(read); assertThat( "PubsubIO.Read should include the topic in its primitive display data", displayData, hasItem(hasDisplayItem("topic"))); }
@Test public void testReadWithPubsubGrpcClientFactory() { String topic = "projects/project/topics/topic"; PubsubIO.Read<String> read = PubsubIO.readStrings() .fromTopic(StaticValueProvider.of(topic)) .withClientFactory(PubsubGrpcClient.FACTORY) .withTimestampAttribute("myTimestamp") .withIdAttribute("myId"); DisplayData displayData = DisplayData.from(read); assertThat(displayData, hasDisplayItem("topic", topic)); assertThat(displayData, hasDisplayItem("timestampAttribute", "myTimestamp")); assertThat(displayData, hasDisplayItem("idAttribute", "myId")); }