Блог

Я пытаюсь публиковать сообщения в теме, используя Google Cloud Pub / Sub. Я следовал этому руководству. Мне успешно удалось создать Topic с помощью следующего метода.

     public static Topic createTopic(String topic) throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create(topicAdminSettings)) {
      ProjectTopicName topicName = ProjectTopicName.of(projectId, topic);
      return topicAdminClient.createTopic(topicName);
    }
  }
  

Следующий метод возвращает topic-id из вновь созданного Topic .

     public static String getTopicId(String topicName) throws IOException {
    String topic_id = null;
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create(topicAdminSettings)) {
      ListTopicsRequest listTopicsRequest =
          ListTopicsRequest.newBuilder().setProject(ProjectName.format(projectId)).build();
      ListTopicsPagedResponse response = topicAdminClient.listTopics(listTopicsRequest);
      Iterable<Topic> topics = response.iterateAll();
      for (Topic topic : topics) {
        // return the topic id for the given topic
         if (topic.getName().toLowerCase().contains(topicName.toLowerCase())) {
           topic_id = topic.getName();
         }
      }
    }
    return topic_id;
  }
  

Но когда я пытаюсь опубликовать сообщения, используя следующий метод

     public static void publishMessages(String topic) throws Exception {
    String topicId = getTopicId(topic);
    ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
    Publisher publisher = null;
    try {
      // Create a publisher instance with default settings bound to the topic
      publisher = Publisher.newBuilder(topicName).build();

      List<String> messages = Arrays.asList("first message", "second message");

      for (final String message : messages) {
        ByteString data = ByteString.copyFromUtf8(message);
        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

        // Once published, returns a server-assigned message id (unique within the topic)
        ApiFuture<String> future = publisher.publish(pubsubMessage);

        // Add an asynchronous callback to handle success / failure
        ApiFutures.addCallback(
            future,
            new ApiFutureCallback<String>() {

              @Override
              public void onFailure(Throwable throwable) {
                if (throwable instanceof ApiException) {
                  ApiException apiException = ((ApiException) throwable);
                  // details on the API exception
                  System.out.println(apiException.getStatusCode().getCode());
                  System.out.println(apiException.isRetryable());
                }
                System.out.println("Error publishing message : "   message);
              }

              @Override
              public void onSuccess(String messageId) {
                // Once published, returns server-assigned message ids (unique within the topic)
                System.out.println(messageId);
              }
            },
            MoreExecutors.directExecutor());
      }
    } finally {
      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }

  }
  

Я получаю следующее исключение

 Exception in thread "main" com.google.api.pathtemplate.ValidationException: Invalid character "/" in path section "projects/deft-idiom-234709/topics/test-topic".
at com.google.api.pathtemplate.PathTemplate.encodeUrl(PathTemplate.java:924)
at com.google.api.pathtemplate.PathTemplate.instantiate(PathTemplate.java:721)
at com.google.api.pathtemplate.PathTemplate.instantiate(PathTemplate.java:646)
at com.google.api.pathtemplate.PathTemplate.instantiate(PathTemplate.java:657)
at com.google.pubsub.v1.ProjectTopicName.toString(ProjectTopicName.java:119)
at com.google.cloud.pubsub.v1.Publisher.newBuilder(Publisher.java:460)
at pubsub.TopicAndPubSub.publishMessages(TopicAndPubSub.java:73)
at pubsub.TopicAndPubSub.main(TopicAndPubSub.java:121)
  

Это весь класс

 import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.core.FixedCredentialsProvider;
import com.google.api.gax.rpc.ApiException;
import com.google.auth.oauth2.ServiceAccountCredentials;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.cloud.pubsub.v1.TopicAdminClient.ListTopicsPagedResponse;
import com.google.cloud.pubsub.v1.TopicAdminSettings;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.ListTopicsRequest;
import com.google.pubsub.v1.ProjectName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.Topic;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class TopicAndPubSub {

  private static ServiceAccountCredentials creds;
  private static TopicAdminSettings topicAdminSettings;
  private static String projectId;

  static {
    try {
      creds = ServiceAccountCredentials.fromStream(new FileInputStream("C:/cred/Key.json"));
      topicAdminSettings = TopicAdminSettings.newBuilder()
          .setCredentialsProvider(FixedCredentialsProvider.create(creds)).build();
      projectId = creds.getProjectId();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  public static Topic createTopic(String topic) throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create(topicAdminSettings)) {
      ProjectTopicName topicName = ProjectTopicName.of(projectId, topic);
      return topicAdminClient.createTopic(topicName);
    }
  }

  public static String getTopicId(String topicName) throws IOException {
    String topic_id = null;
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create(topicAdminSettings)) {
      ListTopicsRequest listTopicsRequest =
          ListTopicsRequest.newBuilder().setProject(ProjectName.format(projectId)).build();
      ListTopicsPagedResponse response = topicAdminClient.listTopics(listTopicsRequest);
      Iterable<Topic> topics = response.iterateAll();
      for (Topic topic : topics) {
        // return the topic id for the given topic
         if (topic.getName().toLowerCase().contains(topicName.toLowerCase())) {
           topic_id = topic.getName();
         }
      }
    }
    return topic_id;
  }

  public static void publishMessages(String topic) throws Exception {
    String topicId = getTopicId(topic);
    ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
    Publisher publisher = null;
    try {
      // Create a publisher instance with default settings bound to the topic
      publisher = Publisher.newBuilder(topicName).build();

      List<String> messages = Arrays.asList("first message", "second message");

      for (final String message : messages) {
        ByteString data = ByteString.copyFromUtf8(message);
        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

        // Once published, returns a server-assigned message id (unique within the topic)
        ApiFuture<String> future = publisher.publish(pubsubMessage);

        // Add an asynchronous callback to handle success / failure
        ApiFutures.addCallback(
            future,
            new ApiFutureCallback<String>() {

              @Override
              public void onFailure(Throwable throwable) {
                if (throwable instanceof ApiException) {
                  ApiException apiException = ((ApiException) throwable);
                  // details on the API exception
                  System.out.println(apiException.getStatusCode().getCode());
                  System.out.println(apiException.isRetryable());
                }
                System.out.println("Error publishing message : "   message);
              }

              @Override
              public void onSuccess(String messageId) {
                // Once published, returns server-assigned message ids (unique within the topic)
                System.out.println(messageId);
              }
            },
            MoreExecutors.directExecutor());
      }
    } finally {
      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }

  }

  public static void main(String[] args) throws Exception {
    publishMessages("test-topic");
  }

}
  

Кажется, я не могу это исправить. Кто-нибудь, пожалуйста, может помочь?!

Оказывается, мне нужно было создать, Publisher используя setCredentialsProvider . Пришлось изменить следующее в publishMessages методе, начиная с

 publisher = Publisher.newBuilder(topicName).build();
  

Для

 publisher = Publisher.newBuilder(
      topicName)
      .setCredentialsProvider(FixedCredentialsProvider.create(creds))
      .build();
  

Теперь работает так, как ожидалось!