PubsubUnboundedSource extends PTransform<PBegin, PCollection<PubsubMessage>> { public @Nullable SubscriptionPath getSubscription() { return subscription == null ? null : subscription.get(); } @VisibleForTesting  PubsubUnboundedSource(
      Clock clock,
      PubsubClientFactory pubsubFactory,
      @Nullable ValueProvider<ProjectPath> project,
      @Nullable ValueProvider<TopicPath> topic,
      @Nullable ValueProvider<SubscriptionPath> subscription,
      @Nullable String timestampAttribute,
      @Nullable String idAttribute,
      boolean needsAttributes,
      boolean needsMessageId);  PubsubUnboundedSource(
      PubsubClientFactory pubsubFactory,
      @Nullable ValueProvider<ProjectPath> project,
      @Nullable ValueProvider<TopicPath> topic,
      @Nullable ValueProvider<SubscriptionPath> subscription,
      @Nullable String timestampAttribute,
      @Nullable String idAttribute,
      boolean needsAttributes);  PubsubUnboundedSource(
      Clock clock,
      PubsubClientFactory pubsubFactory,
      @Nullable ValueProvider<ProjectPath> project,
      @Nullable ValueProvider<TopicPath> topic,
      @Nullable ValueProvider<SubscriptionPath> subscription,
      @Nullable String timestampAttribute,
      @Nullable String idAttribute,
      boolean needsAttributes);  PubsubUnboundedSource(
      PubsubClientFactory pubsubFactory,
      @Nullable ValueProvider<ProjectPath> project,
      @Nullable ValueProvider<TopicPath> topic,
      @Nullable ValueProvider<SubscriptionPath> subscription,
      @Nullable String timestampAttribute,
      @Nullable String idAttribute,
      boolean needsAttributes,
      boolean needsMessageId); @Nullable ProjectPath getProject(); @Nullable TopicPath getTopic(); @Nullable ValueProvider<TopicPath> getTopicProvider(); @Nullable SubscriptionPath getSubscription(); @Nullable ValueProvider<SubscriptionPath> getSubscriptionProvider(); @Nullable String getTimestampAttribute(); @Nullable String getIdAttribute(); boolean getNeedsAttributes(); boolean getNeedsMessageId(); @Override PCollection<PubsubMessage> expand(PBegin input);  }
@Test public void noSubscriptionSplitGeneratesSubscription() throws Exception { TopicPath topicPath = PubsubClient.topicPathFromName("my_project", "my_topic"); factory = PubsubTestClient.createFactoryForCreateSubscription(); PubsubUnboundedSource source = new PubsubUnboundedSource( factory, StaticValueProvider.of(PubsubClient.projectPathFromId("my_project")), StaticValueProvider.of(topicPath), null , null , null , false ); assertThat(source.getSubscription(), nullValue()); assertThat(source.getSubscription(), nullValue()); PipelineOptions options = PipelineOptionsFactory.create(); List<PubsubSource> splits = (new PubsubSource(source)).split(3, options); assertThat(splits, hasSize(greaterThan(0))); for (PubsubSource split : splits) { assertThat(split, equalTo(splits.get(0))); } assertThat(splits.get(0).subscriptionPath, not(nullValue())); }
@Test public void noSubscriptionNoSplitGeneratesSubscription() throws Exception { TopicPath topicPath = PubsubClient.topicPathFromName("my_project", "my_topic"); factory = PubsubTestClient.createFactoryForCreateSubscription(); PubsubUnboundedSource source = new PubsubUnboundedSource( factory, StaticValueProvider.of(PubsubClient.projectPathFromId("my_project")), StaticValueProvider.of(topicPath), null , null , null , false ); assertThat(source.getSubscription(), nullValue()); assertThat(source.getSubscription(), nullValue()); PipelineOptions options = PipelineOptionsFactory.create(); PubsubSource actualSource = new PubsubSource(source); PubsubReader reader = actualSource.createReader(options, null); SubscriptionPath createdSubscription = reader.subscription; assertThat(createdSubscription, not(nullValue())); PubsubCheckpoint checkpoint = reader.getCheckpointMark(); assertThat(checkpoint.subscriptionPath, equalTo(createdSubscription.getPath())); checkpoint.finalizeCheckpoint(); PubsubCheckpoint deserCheckpoint = CoderUtils.clone(actualSource.getCheckpointMarkCoder(), checkpoint); assertThat(checkpoint.subscriptionPath, not(nullValue())); assertThat(checkpoint.subscriptionPath, equalTo(deserCheckpoint.subscriptionPath)); PubsubReader readerFromOriginal = actualSource.createReader(options, checkpoint); PubsubReader readerFromDeser = actualSource.createReader(options, deserCheckpoint); assertThat(readerFromOriginal.subscription, equalTo(createdSubscription)); assertThat(readerFromDeser.subscription, equalTo(createdSubscription)); }