import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
/**
 * Attempts to send the received pubsub message to localhost as an MQTT message.
 *
 * @param msg the message received from cloud pubsub.
 */
private void handlePubsubMessage(ReceivedMessage msg, long ackDeadlineTimer) {
  Future<Boolean> checkStatusTask;
  logger.info("Sending MQTT message to localhost");
  try {
    checkStatusTask = mqttSender.publishToSubscribers(createMqttPublishMessage(msg));
  } catch (IOException e) {
    // failed to send the message. do not ack the pubsub message and wait for re-delivery.
    logger.info("MQTT send error: " + e.getMessage());
    return;
  }
  // create a task for renewing the pubsub message if needed
  GcloudRenewMessageTask renewTask = new GcloudRenewMessageTask.GcloudRenewMessageTaskBuilder()
      .withAckDeadlineTimer(ackDeadlineTimer)
      .withAckId(msg.getAckId())
      .withFutureTask(checkStatusTask)
      .withPubsub(pubsub)
      .withPubsubExecutor(pubsubExecutor)
      .withSubscriptionName(subscriptionName)
      .withTaskDelayTime(ACK_CHECK_DELAY)
      .build();
  pubsubExecutor.schedule(renewTask, ACK_CHECK_DELAY, TimeUnit.SECONDS);
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
private List<ReceivedMessage> getMessagesFromPubsub(String subscriptionName) throws IOException {
  PullRequest request = new PullRequest().setReturnImmediately(true).setMaxMessages(BATCH_SIZE);
  List<ReceivedMessage> msgs = pubsub.projects().subscriptions().pull(subscriptionName, request)
      .execute().getReceivedMessages();
  if (msgs == null) {
    return null;
  }
  // separate received messages into those that were published to pubsub from localhost
  // and those published from other servers
  List<ReceivedMessage> nonLocalMsgs = new LinkedList<ReceivedMessage>();
  List<String> localMessageAckIds = new LinkedList<>();
  for (ReceivedMessage msg : msgs) {
    String msgServerId = msg.getMessage().getAttributes().get(GcloudPubsub.PROXY_SERVER_ID);
    String serverId = InetAddress.getLocalHost().getCanonicalHostName();
    if (msgServerId.equals(serverId)) {
      logger.info("Discarding Pubsub message -- Message was published to pubsub from localhost");
      localMessageAckIds.add(msg.getAckId());
    } else {
      nonLocalMsgs.add(msg);
    }
  }
  if (!localMessageAckIds.isEmpty()) {
    ackPubsubMessage(localMessageAckIds);
  }
  return nonLocalMsgs;
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
@ProcessElement
public void processElement(ProcessContext context) {
    int num = 0;
    Instant endTime = Instant.now().plus(maxTime);
    while (num < maxNum && Instant.now().isBefore(endTime)) {
        try {
            List<String> ackIds = new ArrayList<>();
            for (ReceivedMessage receivedMessage : client.pull(spec.getDatasetProperties().subscription.getValue(),
                    maxNum - num)) {
                context.output(
                        new PubsubMessage(receivedMessage.getMessage().decodeData(), ImmutableMap.<String, String> of()));
                ackIds.add(receivedMessage.getAckId());
                num++;
            }
            if (ack && !ackIds.isEmpty()) { // do not call this when getSample, else the message will be removed
                client.ack(spec.getDatasetProperties().subscription.getValue(), ackIds);
            }
        } catch (IOException e) {
            throw TalendRuntimeException.createUnexpectedException(e);
        }
    }
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
@Test
public void pullOneMessage() throws IOException {
  String expectedSubscription = SUBSCRIPTION.getPath();
  PullRequest expectedRequest =
      new PullRequest().setReturnImmediately(true).setMaxMessages(10);
  PubsubMessage expectedPubsubMessage = new PubsubMessage()
      .setMessageId(MESSAGE_ID)
      .encodeData(DATA.getBytes())
      .setPublishTime(String.valueOf(PUB_TIME))
      .setAttributes(
          ImmutableMap.of(TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME),
              ID_ATTRIBUTE, RECORD_ID));
  ReceivedMessage expectedReceivedMessage =
      new ReceivedMessage().setMessage(expectedPubsubMessage)
                           .setAckId(ACK_ID);
  PullResponse expectedResponse =
      new PullResponse().setReceivedMessages(ImmutableList.of(expectedReceivedMessage));
  Mockito.when((Object) (mockPubsub.projects()
                             .subscriptions()
                             .pull(expectedSubscription, expectedRequest)
                             .execute()))
         .thenReturn(expectedResponse);
  List<IncomingMessage> acutalMessages = client.pull(REQ_TIME, SUBSCRIPTION, 10, true);
  assertEquals(1, acutalMessages.size());
  IncomingMessage actualMessage = acutalMessages.get(0);
  assertEquals(ACK_ID, actualMessage.ackId);
  assertEquals(DATA, new String(actualMessage.elementBytes));
  assertEquals(RECORD_ID, actualMessage.recordId);
  assertEquals(REQ_TIME, actualMessage.requestTimeMsSinceEpoch);
  assertEquals(MESSAGE_TIME, actualMessage.timestampMsSinceEpoch);
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
private PublishMessage createMqttPublishMessage(ReceivedMessage msg) {
  byte [] mqttPayload = BaseEncoding.base64().decode(msg.getMessage().getData());
  Map<String, String> mqttAttributes = msg.getMessage().getAttributes();
  PublishMessage mqttMsg = new PublishMessage.PublishMessageBuilder()
      .withPayload(mqttPayload)
      .withTopic(mqttAttributes.get(GcloudPubsub.MQTT_TOPIC_NAME))
      .withRetain(Boolean.valueOf(mqttAttributes.get(GcloudPubsub.MQTT_RETAIN)))
      .build();
  return mqttMsg;
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
private void outputCsv(Pipeline pipeline) throws IOException {
    String testID = "csvBasicTest" + new Random().nextInt();
    final String fieldDelimited = ";";

    List<Person> expectedPersons = Person.genRandomList(testID, maxRecords);
    List<String> expectedMessages = new ArrayList<>();
    List<String[]> sendMessages = new ArrayList<>();
    for (Person person : expectedPersons) {
        expectedMessages.add(person.toCSV(fieldDelimited));
        sendMessages.add(person.toCSV(fieldDelimited).split(fieldDelimited));
    }

    PubSubOutputRuntime outputRuntime = new PubSubOutputRuntime();
    outputRuntime.initialize(runtimeContainer,
            createOutput(createDatasetFromCSV(createDatastore(), topicName, fieldDelimited)));

    PCollection<IndexedRecord> records = (PCollection<IndexedRecord>) pipeline.apply(Create.of(sendMessages))
            .apply((PTransform) ConvertToIndexedRecord.of());

    records.setCoder(LazyAvroCoder.of()).apply(outputRuntime);

    pipeline.run().waitUntilFinish();

    List<String> actual = new ArrayList<>();
    while (true) {
        List<ReceivedMessage> messages = client.pull(subscriptionName, maxRecords);
        List<String> ackIds = new ArrayList<>();
        for (ReceivedMessage message : messages) {
            actual.add(new String(message.getMessage().decodeData()));
            ackIds.add(message.getAckId());
        }
        client.ack(subscriptionName, ackIds);
        if (actual.size() >= maxRecords) {
            break;
        }
    }
    assertThat(actual, containsInAnyOrder(expectedMessages.toArray()));
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
private void outputAvro(Pipeline pipeline) throws IOException {
    String testID = "avroBasicTest" + new Random().nextInt();

    List<Person> expectedPersons = Person.genRandomList(testID, maxRecords);
    List<String> expectedMessages = new ArrayList<>();
    List<IndexedRecord> sendMessages = new ArrayList<>();
    for (Person person : expectedPersons) {
        expectedMessages.add(person.toAvroRecord().toString());
        sendMessages.add(person.toAvroRecord());
    }

    PubSubOutputRuntime outputRuntime = new PubSubOutputRuntime();
    outputRuntime.initialize(runtimeContainer,
            createOutput(createDatasetFromAvro(createDatastore(), topicName, Person.schema.toString())));

    PCollection<IndexedRecord> output = (PCollection<IndexedRecord>) pipeline.apply(Create.of(sendMessages));
    output.apply(outputRuntime);

    pipeline.run().waitUntilFinish();

    List<String> actual = new ArrayList<>();
    while (true) {
        List<ReceivedMessage> messages = client.pull(subscriptionName, maxRecords);
        List<String> ackIds = new ArrayList<>();
        for (ReceivedMessage message : messages) {
            actual.add(Person.desFromAvroBytes(message.getMessage().decodeData()).toAvroRecord().toString());
            ackIds.add(message.getAckId());
        }
        client.ack(subscriptionName, ackIds);
        if (actual.size() >= maxRecords) {
            break;
        }
    }
    assertThat(actual, containsInAnyOrder(expectedMessages.toArray()));
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
/**
 * Keeps pulling messages from the given 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 pullMessages(final Pubsub client, final String[] args)
        throws IOException {
    Main.checkArgsLength(args, 3);
    String subscriptionName = PubsubUtils.getFullyQualifiedResourceName(
            PubsubUtils.ResourceType.SUBSCRIPTION, args[0], args[2]);
    PullRequest pullRequest = new PullRequest()
            .setReturnImmediately(false)
            .setMaxMessages(Main.BATCH_SIZE);

    do {
        PullResponse pullResponse;
        pullResponse = client.projects().subscriptions()
                .pull(subscriptionName, pullRequest)
                .execute();
        List<String> ackIds = new ArrayList<>(Main.BATCH_SIZE);
        List<ReceivedMessage> receivedMessages =
                pullResponse.getReceivedMessages();
        if (receivedMessages != null) {
            for (ReceivedMessage receivedMessage : receivedMessages) {
                PubsubMessage pubsubMessage =
                        receivedMessage.getMessage();
                if (pubsubMessage != null
                        && pubsubMessage.decodeData() != null) {
                    System.out.println(
                            new String(pubsubMessage.decodeData(),
                                    "UTF-8"));
                }
                ackIds.add(receivedMessage.getAckId());
            }
            AcknowledgeRequest ackRequest = new AcknowledgeRequest();
            ackRequest.setAckIds(ackIds);
            client.projects().subscriptions()
                    .acknowledge(subscriptionName, ackRequest)
                    .execute();
        }
    } while (System.getProperty(Main.LOOP_ENV_NAME) != null);
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
@Override
public List<IncomingMessage> pull(
    long requestTimeMsSinceEpoch,
    SubscriptionPath subscription,
    int batchSize,
    boolean returnImmediately) throws IOException {
  PullRequest request = new PullRequest()
      .setReturnImmediately(returnImmediately)
      .setMaxMessages(batchSize);
  PullResponse response = pubsub.projects()
                                .subscriptions()
                                .pull(subscription.getPath(), request)
                                .execute();
  if (response.getReceivedMessages() == null || response.getReceivedMessages().size() == 0) {
    return ImmutableList.of();
  }
  List<IncomingMessage> incomingMessages = new ArrayList<>(response.getReceivedMessages().size());
  for (ReceivedMessage message : response.getReceivedMessages()) {
    PubsubMessage pubsubMessage = message.getMessage();
    @Nullable Map<String, String> attributes = pubsubMessage.getAttributes();

    // Payload.
    byte[] elementBytes = pubsubMessage.decodeData();

    // Timestamp.
    long timestampMsSinceEpoch =
        extractTimestamp(timestampAttribute, message.getMessage().getPublishTime(), attributes);

    // Ack id.
    String ackId = message.getAckId();
    checkState(!Strings.isNullOrEmpty(ackId));

    // Record id, if any.
    @Nullable String recordId = null;
    if (idAttribute != null && attributes != null) {
      recordId = attributes.get(idAttribute);
    }
    if (Strings.isNullOrEmpty(recordId)) {
      // Fall back to the Pubsub provided message id.
      recordId = pubsubMessage.getMessageId();
    }

    incomingMessages.add(new IncomingMessage(elementBytes, attributes, timestampMsSinceEpoch,
                                             requestTimeMsSinceEpoch, ackId, recordId));
  }

  return incomingMessages;
}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
@Override
public void run() {
	try {
		while (!pubsubReceiver.isStopped()) {

			final PullRequest pullRequest = new PullRequest().setReturnImmediately(false)
					.setMaxMessages(pubsubReceiver.getBatchSize());
			PullResponse pullResponse;
			pullResponse = pubsubClient.projects().subscriptions().pull(pubsubReceiver.getSubscription(), pullRequest)
					.execute();
			final List<ReceivedMessage> receivedMessages = pullResponse.getReceivedMessages();

			if (CollectionUtils.isNotEmpty(receivedMessages)) {

				final List<String> messages = new ArrayList<>(pubsubReceiver.getBatchSize());
				final List<String> ackIds = new ArrayList<>(pubsubReceiver.getBatchSize());
				for (final ReceivedMessage receivedMessage : receivedMessages) {
					final PubsubMessage pubsubMessage = receivedMessage.getMessage();

					if (pubsubMessage != null) {
						if (decodeData) {
							messages.add(new String(pubsubMessage.decodeData(), StandardCharsets.UTF_8));
						} else {
							messages.add(pubsubMessage.getData());
						}
					}
					ackIds.add(receivedMessage.getAckId());
				}

				if (CollectionUtils.isNotEmpty(messages)) {
					pubsubReceiver.store(messages.iterator());
				}

				final AcknowledgeRequest ackRequest = new AcknowledgeRequest().setAckIds(ackIds);
				pubsubClient.projects().subscriptions().acknowledge(pubsubReceiver.getSubscription(), ackRequest)
						.execute();
			}
		}
	} catch (Throwable t) {
		pubsubReceiver.restart(
				"Error while fetching messages from pubsub for subscription " + pubsubReceiver.getSubscription(), t);
	}

}
 

import com.google.api.services.pubsub.model.ReceivedMessage; //导入依赖的package包/类
private void createTopicSub(Pipeline pipeline) throws IOException {
    String testID = "createTopicSubTest" + new Random().nextInt();

    final String newTopicName = "tcomp-pubsub-createTopicSub" + uuid;

    final String newSubName = "tcomp-pubsub-createTopicSub-sub" + uuid;

    final String fieldDelimited = ";";

    List<Person> expectedPersons = Person.genRandomList(testID, maxRecords);
    List<String> expectedMessages = new ArrayList<>();
    List<String[]> sendMessages = new ArrayList<>();
    for (Person person : expectedPersons) {
        expectedMessages.add(person.toCSV(fieldDelimited));
        sendMessages.add(person.toCSV(fieldDelimited).split(fieldDelimited));
    }

    PubSubOutputRuntime outputRuntime = new PubSubOutputRuntime();
    PubSubOutputProperties outputProperties = createOutput(
            addSubscriptionForDataset(createDatasetFromCSV(createDatastore(), newTopicName, fieldDelimited), newSubName));
    outputProperties.topicOperation.setValue(PubSubOutputProperties.TopicOperation.CREATE_IF_NOT_EXISTS);
    outputRuntime.initialize(runtimeContainer, outputProperties);

    PCollection<IndexedRecord> records = (PCollection<IndexedRecord>) pipeline.apply(Create.of(sendMessages))
            .apply((PTransform) ConvertToIndexedRecord.of());

    records.setCoder(LazyAvroCoder.of()).apply(outputRuntime);

    pipeline.run().waitUntilFinish();

    List<String> actual = new ArrayList<>();
    while (true) {
        List<ReceivedMessage> messages = client.pull(newSubName, maxRecords);
        List<String> ackIds = new ArrayList<>();
        for (ReceivedMessage message : messages) {
            actual.add(new String(message.getMessage().decodeData()));
            ackIds.add(message.getAckId());
        }
        client.ack(newSubName, ackIds);
        if (actual.size() >= maxRecords) {
            break;
        }
    }

    client.deleteSubscription(newSubName);
    client.deleteTopic(newTopicName);
    assertThat(actual, containsInAnyOrder(expectedMessages.toArray()));
}