import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
  Publisher publisher = this.publisher;
  // construct a pubsub message from the payload
  final String payload = req.getParameter("payload");
  Message message = new Message(null);
  message.setData(payload);
  PubsubMessage pubsubMessage =
      PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8(payload))
          .putAttributes("sourceLang", req.getParameter("sourceLang"))
          .putAttributes("targetLang", req.getParameter("targetLang"))
          .build();
  String topicId = System.getenv("PUBSUB_TOPIC");
  // create a publisher on the topic
  if (publisher == null) {
    this.publisher = publisher = Publisher.newBuilder(
        TopicName.of(ServiceOptions.getDefaultProjectId(), topicId))
        .build();
  }

  publisher.publish(pubsubMessage);
  // redirect to home page
  resp.sendRedirect("/");
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
private void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
	Map<String, Object> messageHeaders = new HashMap<>();

	message.getAttributesMap().forEach(messageHeaders::put);

	if (this.ackMode == AckMode.MANUAL) {
		// Send the consumer downstream so user decides on when to ack/nack.
		messageHeaders.put(GcpHeaders.ACKNOWLEDGEMENT, consumer);
	}

	try {
		sendMessage(this.messageConverter.toMessage(
				message.getData().toStringUtf8(),
				new MessageHeaders(messageHeaders)));
	}
	catch (RuntimeException re) {
		if (this.ackMode == AckMode.AUTO) {
			consumer.nack();
		}
		throw re;
	}

	if (this.ackMode == AckMode.AUTO) {
		consumer.ack();
	}
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public void receiveMessage(final PubsubMessage message, final AckReplyConsumer consumer) {
  recordMessageLatency(
      Integer.parseInt(message.getAttributesMap().get("clientId")),
      Integer.parseInt(message.getAttributesMap().get("sequenceNumber")),
      System.currentTimeMillis() - Long.parseLong(message.getAttributesMap().get("sendTime")));
  consumer.ack();
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public void doPost(HttpServletRequest req, HttpServletResponse resp)
    throws IOException, ServletException {
  Publisher publisher = this.publisher;
  try {
    String topicId = System.getenv("PUBSUB_TOPIC");
    // create a publisher on the topic
    if (publisher == null) {
      publisher = Publisher.newBuilder(
          TopicName.of(ServiceOptions.getDefaultProjectId(), topicId))
          .build();
    }
    // construct a pubsub message from the payload
    final String payload = req.getParameter("payload");
    PubsubMessage pubsubMessage =
        PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8(payload)).build();

    publisher.publish(pubsubMessage);
    // redirect to home page
    resp.sendRedirect("/");
  } catch (Exception e) {
    resp.sendError(HttpStatus.SC_INTERNAL_SERVER_ERROR, e.getMessage());
  }
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Test
public void servletPublishesPayloadMessage() throws Exception {
  assertNotNull(System.getenv("PUBSUB_TOPIC"));
  HttpServletRequest request = mock(HttpServletRequest.class);
  when(request.getParameter("payload")).thenReturn("test-message");

  HttpServletResponse response = mock(HttpServletResponse.class);
  Publisher publisher = mock(Publisher.class);
  PubsubMessage message = PubsubMessage.newBuilder()
      .setData(ByteString.copyFromUtf8("test-message")).build();
  when(publisher.publish(eq(message))).thenReturn(SettableApiFuture.create());
  PubSubPublish pubSubPublish = new PubSubPublish(publisher);
  // verify content of published test message
  pubSubPublish.doPost(request, response);
  verify(publisher, times(1)).publish(eq(message));
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
/** Receive messages over a subscription. */
public static void main(String... args) throws Exception {
  // set subscriber id, eg. my-sub
  String subscriptionId = args[0];
  SubscriptionName subscriptionName = SubscriptionName.of(PROJECT_ID, subscriptionId);
  Subscriber subscriber = null;
  try {
    // create a subscriber bound to the asynchronous message receiver
    subscriber =
        Subscriber.newBuilder(subscriptionName, new MessageReceiverExample()).build();
    subscriber.startAsync().awaitRunning();
    // Continue to listen to messages
    while (true) {
      PubsubMessage message = messages.take();
      System.out.println("Message Id: " + message.getMessageId());
      System.out.println("Data: " + message.getData().toStringUtf8());
    }
  } finally {
    if (subscriber != null) {
      subscriber.stopAsync();
    }
  }
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public void doPost(HttpServletRequest req, HttpServletResponse resp)
    throws IOException, ServletException {
  Publisher publisher = this.publisher;
  try {
    String topicId = System.getenv("PUBSUB_TOPIC");
    // create a publisher on the topic
    if (publisher == null) {
      publisher = Publisher.defaultBuilder(
          TopicName.create(ServiceOptions.getDefaultProjectId(), topicId))
          .build();
    }
    // construct a pubsub message from the payload
    final String payload = req.getParameter("payload");
    PubsubMessage pubsubMessage =
        PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8(payload)).build();

    publisher.publish(pubsubMessage);
    // redirect to home page
    resp.sendRedirect("/");
  } catch (Exception e) {
    resp.sendError(HttpStatus.SC_INTERNAL_SERVER_ERROR, e.getMessage());
  }
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
private void parseMessage(PubsubMessage message) {
  LOG.trace("parseMessage called");
  try (DataParser parser = parserFactory.getParser(message.getMessageId(), message.getData().toByteArray())) {
    Record r;
    while ((r = tryParse(parser, message)) != null) {
      setHeaders(message, r);

      batchMaker.addRecord(r);
      ++currentRecordCount;
    }
  } catch (DataParserException | IOException e) {
    LOG.error(Errors.PUBSUB_05.getMessage(), e.toString(), e);
    // Create a raw record of the gRPC message data, set attributes as headers, and use the messageId for the recordId
    Record errorRecord = context.createRecord(message.getMessageId(), message.getData().toByteArray(), MIME_GRPC);
    setHeaders(message, errorRecord);
    context.reportError(new OnRecordErrorException(errorRecord, Errors.PUBSUB_05, e.toString()));
  }
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
private void publish(Record record) throws StageException {
  ByteArrayOutputStream os = new ByteArrayOutputStream();
  try (DataGenerator generator = generatorFactory.getGenerator(os)) {
    generator.write(record);
  } catch (IOException | DataGeneratorException e) {
    errorRecordHandler.onError(new OnRecordErrorException(record, Errors.PUBSUB_06, e.toString(), e));
    return;
  }

  ByteString data = ByteString.copyFrom(os.toByteArray());

  Map<String, String> attributes = new HashMap<>();
  Record.Header header = record.getHeader();
  header.getAttributeNames().forEach(k -> attributes.put(k, header.getAttribute(k)));

  PubsubMessage message = PubsubMessage.newBuilder().setData(data).putAllAttributes(attributes).build();

  ApiFuture<String> messageIdFuture = publisher.publish(message);
  pendingMessages.add(new PendingMessage(record, messageIdFuture));
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public ListenableFuture<String> publish(final String topic, ByteString payload,
		Map<String, String> headers) {
	PubsubMessage.Builder pubsubMessageBuilder = PubsubMessage.newBuilder().setData(payload);

	if (headers != null) {
		pubsubMessageBuilder.putAllAttributes(headers);
	}

	return publish(topic, pubsubMessageBuilder.build());
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
/**
 * Pulls messages synchronously, on demand, using the pull request in argument.
 *
 * <p>
 * This method acknowledges all received messages.
 * @param pullRequest pull request containing the subscription name
 * @return the list of {@link PubsubMessage} containing the headers and payload
 */
private List<PubsubMessage> pull(PullRequest pullRequest, RetrySettings retrySettings) {
	Assert.notNull(pullRequest, "The pull request cannot be null.");

	try {
		SubscriberStub subscriber = this.subscriberFactory.createSubscriberStub(retrySettings);
		Assert.notNull(subscriber, "A SubscriberStub is needed to execute the pull request.");

		PullResponse pullResponse =	subscriber.pullCallable().call(pullRequest);

		// Ack received messages.
		if (pullResponse.getReceivedMessagesCount() > 0) {
			List<String> ackIds = pullResponse.getReceivedMessagesList().stream()
					.map(ReceivedMessage::getAckId)
					.collect(Collectors.toList());

			AcknowledgeRequest acknowledgeRequest = AcknowledgeRequest.newBuilder()
					.setSubscriptionWithSubscriptionName(
							pullRequest.getSubscriptionAsSubscriptionName())
					.addAllAckIds(ackIds)
					.build();

			subscriber.acknowledgeCallable().call(acknowledgeRequest);
		}

		return pullResponse.getReceivedMessagesList().stream()
				.map(ReceivedMessage::getMessage)
				.collect(Collectors.toList());
	}
	catch (Exception ioe) {
		throw new PubSubException("Error pulling messages from subscription "
				+ pullRequest.getSubscription() + ".", ioe);
	}
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Before
public void setUp() {
	this.pubSubTemplate = new PubSubTemplate(this.mockPublisherFactory,
			this.mockSubscriberFactory);
	when(this.mockPublisherFactory.createPublisher("testTopic"))
			.thenReturn(this.mockPublisher);
	this.settableApiFuture = SettableApiFuture.create();
	when(this.mockPublisher.publish(isA(PubsubMessage.class)))
			.thenReturn(this.settableApiFuture);

	when(this.mockSubscriberFactory.createSubscriber(
			eq("testSubscription"), isA(MessageReceiver.class)))
			.thenReturn(this.mockSubscriber);
	when(this.mockSubscriber.startAsync()).thenReturn(mock(ApiService.class));

	this.pubsubMessage = PubsubMessage.newBuilder().setData(
			ByteString.copyFrom("permanating".getBytes())).build();
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Test
public void testPublish_String() {
	this.pubSubTemplate.publish("testTopic", "testPayload", null);

	verify(this.mockPublisher, times(1))
			.publish(isA(PubsubMessage.class));
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Test
public void testPublish_Bytes() {
	this.pubSubTemplate.publish("testTopic", "testPayload".getBytes(), null);

	verify(this.mockPublisher, times(1))
			.publish(isA(PubsubMessage.class));
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public void publishTask(final String queue, final String task) {
    Topic topic = getTopic(queue);
    ByteString data = ByteString.copyFromUtf8(task);
    TopicName topicName = topic.getNameAsTopicName();

    try {
        PubsubMessage psmessage = PubsubMessage.newBuilder().setData(data).build();
        Publisher publisher = randomHouse.get(topicName);
        if (publisher == null) {
            logger.trace("No publisher found for " + topicName + " - creating");
            Builder builder = Publisher.defaultBuilder(topicName);
            // The default executor provider creates an insane number of threads.
            if (executor != null) builder.setExecutorProvider(executor);
            publisher = builder.build();
            randomHouse.put(topicName, publisher);
        } else {
            logger.trace("Existing publisher found for " + topicName);
        }

        ApiFuture<String> messageIdFuture = publisher.publish(psmessage);

        if (executor != null) messageIdFuture.addListener(listener, executor.getExecutor());

    } catch (IOException e) {
        String error = String.format("Cannot send message to topic %s:\n%s", topic.getName(), ExceptionToString.format(e));
        logger.error(error);
        throw RaptureExceptionFactory.create(HttpURLConnection.HTTP_INTERNAL_ERROR, error, e);
    }
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
/** Publish all the messages in a partition and store the Future's for each publish request. */
private void publishMessagesForPartition(
    String topic, Integer partition, List<PubsubMessage> messages) {
  // Get a map containing all futures per partition for the passed in topic.
  Map<Integer, OutstandingFuturesForPartition> outstandingFuturesForTopic =
      allOutstandingFutures.get(topic);
  if (outstandingFuturesForTopic == null) {
    outstandingFuturesForTopic = new HashMap<>();
    allOutstandingFutures.put(topic, outstandingFuturesForTopic);
  }
  // Get the object containing the outstanding futures for this topic and partition..
  OutstandingFuturesForPartition outstandingFutures = outstandingFuturesForTopic.get(partition);
  if (outstandingFutures == null) {
    outstandingFutures = new OutstandingFuturesForPartition();
    outstandingFuturesForTopic.put(partition, outstandingFutures);
  }
  int startIndex = 0;
  int endIndex = Math.min(CPS_MAX_MESSAGES_PER_REQUEST, messages.size());
  PublishRequest.Builder builder = PublishRequest.newBuilder();
  // Publish all the messages for this partition in batches.
  while (startIndex < messages.size()) {
    PublishRequest request =
        builder.setTopic(cpsTopic).addAllMessages(messages.subList(startIndex, endIndex)).build();
    builder.clear();
    log.trace("Publishing: " + (endIndex - startIndex) + " messages");
    outstandingFutures.futures.add(publisher.publish(request));
    startIndex = endIndex;
    endIndex = Math.min(endIndex + CPS_MAX_MESSAGES_PER_REQUEST, messages.size());
  }
  messages.clear();
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
/**
 * Get some PubsubMessage's which correspond to the SinkRecord's created in {@link
 * #getSampleRecords()}.
 */
private List<PubsubMessage> getPubsubMessagesFromSampleRecords() {
  List<PubsubMessage> messages = new ArrayList<>();
  Map<String, String> attributes = new HashMap<>();
  attributes.put(ConnectorUtils.CPS_MESSAGE_KEY_ATTRIBUTE, KAFKA_MESSAGE_KEY);
  messages.add(
      PubsubMessage.newBuilder().putAllAttributes(attributes).setData(KAFKA_MESSAGE1).build());
  messages.add(
      PubsubMessage.newBuilder().putAllAttributes(attributes).setData(KAFKA_MESSAGE2).build());
  return messages;
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public void send(
    final Destination destination,
    final Message message,
    final int deliveryMode,
    final int priority,
    final long timeToLive,
    final CompletionListener completionListener) throws JMSException {
  if (isClosed()) {
    throw new IllegalStateException("Producer has been closed.");
  }

  if (!getDestination().equals(destination)) {
    throw new IllegalArgumentException("Destination [" + destination
        + "] is invalid. Expected [" + getDestination() + "].");
  }

  final RpcFuture<String> messageIdFuture = publisher.publish(
      PubsubMessage.newBuilder()
          .setData(ByteString.copyFromUtf8(message.getBody(String.class)))
          .build());

  messageIdFuture.addCallback(
      new RpcFutureCallback<String>() {
        @Override public void onSuccess(final String messageId) {
          LOGGER.fine(String.format("%s has been sent successfully.", messageId));
          if (null != completionListener) {
            completionListener.onCompletion(message);
          }
        }

        @Override public void onFailure(final Throwable thrown) {
          LOGGER.log(Level.SEVERE, "Message sending error:", thrown);
          if (null != completionListener) {
            completionListener.onException(message, (Exception) thrown);
          }
        }
      });
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public ListenableFuture<RunResult> doRun() {
  AtomicInteger numPending = new AtomicInteger(batchSize);
  final SettableFuture<RunResult> done = SettableFuture.create();
  String sendTime = String.valueOf(System.currentTimeMillis());
  if (!outstandingBytes.tryAcquire(batchSize * messageSize)) {
    return Futures.immediateFailedFuture(new Exception("Flow control limits reached."));
  }
  for (int i = 0; i < batchSize; i++) {
    ApiFutures.addCallback(publisher
        .publish(
            PubsubMessage.newBuilder()
            .setData(payload)
            .putAttributes("sendTime", sendTime)
            .putAttributes("clientId", id.toString())
            .putAttributes(
                "sequenceNumber", Integer.toString(sequenceNumber.getAndIncrement()))
            .build()), new ApiFutureCallback<String>() {
          @Override
          public void onSuccess(String messageId) {
            outstandingBytes.release(messageSize);
            if (numPending.decrementAndGet() == 0) {
              done.set(RunResult.fromBatchSize(batchSize));
            }
          }

          @Override
          public void onFailure(Throwable t) {
            outstandingBytes.release(messageSize);
            done.setException(t);
          }
        });
  }
  return done;
}
 

import com.google.pubsub.v1.PubsubMessage; //导入依赖的package包/类
@Override
public int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages)
    throws IOException {
  PublishRequest.Builder request = PublishRequest.newBuilder()
                                                 .setTopic(topic.getPath());
  for (OutgoingMessage outgoingMessage : outgoingMessages) {
    PubsubMessage.Builder message =
        PubsubMessage.newBuilder()
                     .setData(ByteString.copyFrom(outgoingMessage.elementBytes));

    if (outgoingMessage.attributes != null) {
      message.putAllAttributes(outgoingMessage.attributes);
    }

    if (timestampAttribute != null) {
      message.getMutableAttributes()
             .put(timestampAttribute, String.valueOf(outgoingMessage.timestampMsSinceEpoch));
    }

    if (idAttribute != null && !Strings.isNullOrEmpty(outgoingMessage.recordId)) {
      message.getMutableAttributes().put(idAttribute, outgoingMessage.recordId);
    }

    request.addMessages(message);
  }

  PublishResponse response = publisherStub().publish(request.build());
  return response.getMessageIdsCount();
}