import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
public static void main(String[] args) throws IOException, GeneralSecurityException {
    PubSubWrapper pubsub = PubSubWrapper.getInstance(
            System.getProperty("PRIVATE_KEY_FILE_PATH"),
            System.getProperty("SERVICE_ACCOUNT_EMAIL"),
            PROJECT);

    String subscription = (args.length > 0) ? args[0] : SUBSCRIPTION;
    boolean doAck = args.length < 2 || args[1].toLowerCase().startsWith("ack");

    Subscription sub = pubsub.subscribeTopic(subscription, TOPIC);
    System.out.println("Subscribed " + sub.getName() + " to " + sub.getTopic());
    System.out.println("Ack: " + doAck);
    while (true) {
        List<String> data = pubsub.pullMessage(sub, 1, doAck);
        for (String s : data) {
            System.out.println(s);
        }
    }
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
private void subscribe(Subscription subscription, String cpsSubscriptionName,
    String cpsSubscriptionTopic) throws IOException {
  try {
    pubsub.projects().subscriptions().create(cpsSubscriptionName, subscription).execute();
  } catch (GoogleJsonResponseException e) {
    logger.info("Pubsub Subscribe Error code: " + e.getStatusCode() + "\n" + e.getMessage());
    if (e.getStatusCode() == RESOURCE_CONFLICT) {
      // there is already a subscription and a pull request for this topic.
      // do nothing and return.
      // TODO this condition could change based on the implementation of UNSUBSCRIBE
      logger.info("Cloud PubSub subscription already exists");
    } else if (e.getStatusCode() == RESOURCE_NOT_FOUND) {
      logger.info("Cloud PubSub Topic Not Found");
      createTopic(cpsSubscriptionTopic);
      // possible that subscription name already exists, and might throw an exception.
      // But, we should not treat that as an error.
      subscribe(subscription, cpsSubscriptionName, cpsSubscriptionTopic);
    } else {
      // exception was caused due to some other reason, so we re-throw and do not send a SUBACK.
      // client will re-send the subscription.
      throw e;
    }
  }
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
/**
 * Creates a new subscription.
 *
 * @param client Cloud Pub/Sub client.
 * @param args Arguments as an array of String.
 * @throws IOException when Cloud Pub/Sub API calls fail.
 */
public static void createSubscription(final Pubsub client,
                                      final String[] args)
        throws IOException {
    Main.checkArgsLength(args, 4);
    String subscriptionName = PubsubUtils.getFullyQualifiedResourceName(
            PubsubUtils.ResourceType.SUBSCRIPTION, args[0], args[2]);
    Subscription subscription = new Subscription()
            .setTopic(PubsubUtils.getFullyQualifiedResourceName(
                    PubsubUtils.ResourceType.TOPIC, args[0], args[3]));
    if (args.length == 5) {
        subscription = subscription.setPushConfig(
            new PushConfig().setPushEndpoint(args[4]));
    }
    subscription = client.projects().subscriptions()
            .create(subscriptionName, subscription)
            .execute();
    System.out.printf(
            "Subscription %s was created.\n", subscription.getName());
    System.out.println(subscription.toPrettyString());
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
private void setupPubsubSubscription(String topic, String subscription) throws IOException {
  if (pubsubClient == null) {
    pubsubClient = newPubsubClient(options.as(PubsubOptions.class)).build();
  }
  if (executeNullIfNotFound(pubsubClient.projects().subscriptions().get(subscription)) == null) {
    Subscription subInfo = new Subscription()
      .setAckDeadlineSeconds(60)
      .setTopic(topic);
    pubsubClient.projects().subscriptions().create(subscription, subInfo).execute();
  }
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
@Override
public void createSubscription(
    TopicPath topic, SubscriptionPath subscription,
    int ackDeadlineSeconds) throws IOException {
  Subscription request = new Subscription()
      .setTopic(topic.getPath())
      .setAckDeadlineSeconds(ackDeadlineSeconds);
  pubsub.projects()
        .subscriptions()
        .create(subscription.getPath(), request)
        .execute(); // ignore Subscription result.
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
private synchronized void updateOnSubscribe(String clientId, String mqttTopic,
    String cpsSubscriptionName, String cpsTopic) throws IOException {
  List<String> clientIds = cpsSubscriptionMap.get(cpsTopic);
  if (clientIds == null) {
    // create pubsub subscription
    Subscription subscription = new Subscription()
        .setTopic(cpsTopic) // the name of the topic
        .setAckDeadlineSeconds(SUBSCRIPTION_ACK_DEADLINE); // acknowledgement deadline in seconds
    subscribe(subscription, cpsSubscriptionName, cpsTopic);
    // update subscription maps
    activeSubscriptions.add(cpsSubscriptionName);
    addEntryToClientIdSubscriptionMap(clientId, mqttTopic, cpsTopic);
    addEntryToCpsSubscriptionMap(clientId, cpsTopic, cpsSubscriptionName);
    // schedule pull task for the very first time we have a client Id subscribe to a pubsub topic
    // task must be started after subscription maps are updated
    GcloudPullMessageTask pullTask = new GcloudPullMessageTask.GcloudPullMessageTaskBuilder()
        .withMqttSender(context)
        .withGcloud(this)
        .withPubsub(pubsub)
        .withPubsubExecutor(taskExecutor)
        .withSubscriptionName(cpsSubscriptionName)
        .build();
    taskExecutor.submit(pullTask);
    logger.info("Created Cloud PubSub pulling task for: " + cpsSubscriptionName);
  } else {
    // update subscription maps
    addEntryToClientIdSubscriptionMap(clientId, mqttTopic, cpsTopic);
    addEntryToCpsSubscriptionMap(clientId, cpsTopic, cpsSubscriptionName);
  }
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
/**
 * Lists existing subscriptions within a project.
 *
 * @param client Cloud Pub/Sub client.
 * @param args Arguments as an array of String.
 * @throws IOException when Cloud Pub/Sub API calls fail.
 */
public static void listSubscriptions(final Pubsub client,
                                     final String[] args)
        throws IOException {
    String nextPageToken = null;
    boolean hasSubscriptions = false;
    Pubsub.Projects.Subscriptions.List listMethod =
            client.projects().subscriptions().list("projects/" + args[0]);
    do {
        if (nextPageToken != null) {
            listMethod.setPageToken(nextPageToken);
        }
        ListSubscriptionsResponse response = listMethod.execute();
        if (!response.isEmpty()) {
            for (Subscription subscription : response.getSubscriptions()) {
                hasSubscriptions = true;
                System.out.println(subscription.toPrettyString());
            }
        }
        nextPageToken = response.getNextPageToken();
    } while (nextPageToken != null);
    if (!hasSubscriptions) {
        System.out.println(String.format(
                "There are no subscriptions in the project '%s'.",
                args[0]));
    }
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
/**
 * Creates a Cloud Pub/Sub subscription if it doesn't exist.
 *
 * @param client Pubsub client object.
 * @throws IOException when API calls to Cloud Pub/Sub fails.
 */
private void setupSubscription(final Pubsub client) throws IOException {
    String fullName = String.format("projects/%s/subscriptions/%s",
            PubsubUtils.getProjectId(),
            PubsubUtils.getAppSubscriptionName());

    try {
        client.projects().subscriptions().get(fullName).execute();
    } catch (GoogleJsonResponseException e) {
        if (e.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) {
            // Create the subscription if it doesn't exist
            String fullTopicName = String.format("projects/%s/topics/%s",
                    PubsubUtils.getProjectId(),
                    PubsubUtils.getAppTopicName());
            PushConfig pushConfig = new PushConfig()
                    .setPushEndpoint(PubsubUtils.getAppEndpointUrl());
            Subscription subscription = new Subscription()
                    .setTopic(fullTopicName)
                    .setPushConfig(pushConfig);
            client.projects().subscriptions()
                    .create(fullName, subscription)
                    .execute();
        } else {
            throw e;
        }
    }
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
@Override
public int ackDeadlineSeconds(SubscriptionPath subscription) throws IOException {
  Subscription response = pubsub.projects().subscriptions().get(subscription.getPath()).execute();
  return response.getAckDeadlineSeconds();
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
public static void subscribe(Pubsub pubsub, String topicName, String subscriptionName, Integer ackDeadlineSec)
		throws IOException {
	if (pubsub == null) {
		logger.info("Not creating subscription {} for topic {} since pubsub has not been configured", subscriptionName,
				topicName);
		return;
	}

	Subscription subscription = new Subscription()
			// The name of the topic from which this subscription
			// receives messages
			.setTopic(topicName);

	if (ackDeadlineSec != null) {
		// Ackowledgement deadline in second
		subscription.setAckDeadlineSeconds(ackDeadlineSec);
	}

	try {
		Subscription newSubscription = pubsub.projects().subscriptions().create(subscriptionName, subscription)
				.execute();
		logger.info("Created: " + newSubscription.getName());
	} catch (GoogleJsonResponseException e) {
		final GoogleJsonError details = e.getDetails();
		boolean ignorable = true;

		final Object objectStatus = details.get("status");
		if (objectStatus instanceof String) {
			String status = (String) objectStatus;
			if (!ALREADY_EXISTS.equals(status)) {
				ignorable = false;
			} else {
				logger.info("Subscription {} already exists.", subscriptionName);
			}
		}

		if (!ignorable) {
			throw e;
		}
	}
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
/**
 * Verifies that the subscription with the name defined in settings file actually exists and 
 * points to a correct topic defined in the same settings file. If the subscription doesn't
 * exist, it will be created.
 */
private static void ensureSubscriptionExists(Pubsub client) throws IOException {
  // First we check if the subscription with this name actually exists.
  Subscription subscription = null;

  String topicName = Settings.getSettings().getTopicName();
  String subName = Settings.getSettings().getSubscriptionName();

  LOG.info("Will be using topic name: " + topicName + ", subscription name: " + subName);

  try {
    LOG.info("Trying to get subscription named " + subName);
    subscription = client
        .projects()
        .subscriptions()
        .get(subName)
        .execute();
    
    Preconditions.checkArgument(
        subscription.getTopic().equals(topicName),
        "Subscription %s already exists but points to a topic %s and not %s." +
            "Please specify a different subscription name or delete this subscription",
        subscription.getName(),
        subscription.getTopic(),
        topicName);

    LOG.info("Will be re-using existing subscription: " + subscription.toPrettyString());
  } catch (HttpResponseException e) {

    // Subscription not found
    if (e.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) {
      LOG.info("Subscription doesn't exist, will try to create " + subName);

      // Creating subscription
      subscription = client
          .projects()
          .subscriptions()
          .create(subName, new Subscription()
              .setTopic(topicName)   // Name of the topic it subscribes to
              .setAckDeadlineSeconds(600)
              .setPushConfig(new PushConfig()
                  // FQDN with valid SSL certificate
                  .setPushEndpoint(Settings.getSettings().getPushEndpoint())))
          .execute();

      LOG.info("Created: " + subscription.toPrettyString());
    }
  }
}
 

import com.google.api.services.pubsub.model.Subscription; //导入依赖的package包/类
@Test
public void testGooglePubSubComponent() throws Exception {
    Properties properties = new Properties();
    properties.load(PubSubIntegrationTest.class.getResourceAsStream("/pubsub.properties"));

    String serviceURL = properties.getProperty("test.serviceURL");
    if (System.getenv("DOCKER_HOST") != null) {
        serviceURL = String.format("http://%s:8590", TestUtils.getDockerHost());
    }

    GooglePubsubConnectionFactory connectionFactory = new GooglePubsubConnectionFactory()
        .setServiceAccount(properties.getProperty("service.account"))
        .setServiceAccountKey(properties.getProperty("service.key"))
        .setServiceURL(serviceURL);

    String topicFullName = String.format("projects/%s/topics/%s",
        properties.getProperty("project.id"),
        properties.getProperty("topic.name"));

    String subscriptionFullName = String.format("projects/%s/subscriptions/%s",
        properties.getProperty("project.id"),
        properties.getProperty("subscription.name"));

    Pubsub pubsub = connectionFactory.getDefaultClient();
    pubsub.projects().topics().create(topicFullName, new Topic()).execute();

    Subscription subscription = new Subscription().setTopic(topicFullName).setAckDeadlineSeconds(10);
    pubsub.projects().subscriptions().create(subscriptionFullName, subscription).execute();

    CamelContext camelctx = new DefaultCamelContext();

    GooglePubsubComponent component = camelctx.getComponent("google-pubsub", GooglePubsubComponent.class);
    component.setConnectionFactory(connectionFactory);

    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:send")
            .toF("google-pubsub:%s:%s", properties.getProperty("project.id"), properties.getProperty("topic.name"));

            fromF("google-pubsub:%s:%s", properties.getProperty("project.id"), properties.getProperty("subscription.name"))
            .to("direct:receive");

            from("direct:receive")
            .to("mock:result");
        }
    });

    MockEndpoint mockEndpoint = camelctx.getEndpoint("mock:result", MockEndpoint.class);
    mockEndpoint.expectedBodiesReceivedInAnyOrder("Hello Kermit");

    camelctx.start();
    try {
        Map<String, String> attributes = new HashMap<>();
        attributes.put("ATTRIBUTE-TEST-KEY", "ATTRIBUTE-TEST-VALUE");

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBodyAndHeader("direct:send", "Hello Kermit", GooglePubsubConstants.ATTRIBUTES, attributes);

        mockEndpoint.assertIsSatisfied();
    } finally {
        camelctx.stop();
        pubsub.projects().topics().delete(topicFullName).execute();
        pubsub.projects().subscriptions().delete(subscriptionFullName).execute();
    }
}