import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
/**
 * Publish 'numMessages' arbitrary events from live users with the provided delay, to a
 * PubSub topic.
 */
public static void publishDataToPubSub(int numMessages, int delayInMillis)
    throws IOException {
  List<PubsubMessage> pubsubMessages = new ArrayList<>();

  for (int i = 0; i < Math.max(1, numMessages); i++) {
    Long currTime = System.currentTimeMillis();
    String message = generateEvent(currTime, delayInMillis);
    PubsubMessage pubsubMessage = new PubsubMessage()
        .encodeData(message.getBytes("UTF-8"));
    pubsubMessage.setAttributes(
        ImmutableMap.of(TIMESTAMP_ATTRIBUTE,
            Long.toString((currTime - delayInMillis) / 1000 * 1000)));
    if (delayInMillis != 0) {
      System.out.println(pubsubMessage.getAttributes());
      System.out.println("late data for: " + message);
    }
    pubsubMessages.add(pubsubMessage);
  }

  PublishRequest publishRequest = new PublishRequest();
  publishRequest.setMessages(pubsubMessages);
  pubsub.projects().topics().publish(topic, publishRequest).execute();
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
/**
 * Publish 'numMessages' arbitrary events from live users with the provided delay, to a
 * PubSub topic.
 */
public static void publishData(int numMessages, int delayInMillis)
    throws IOException {
  List<PubsubMessage> pubsubMessages = new ArrayList<>();

  for (int i = 0; i < Math.max(1, numMessages); i++) {
    Long currTime = System.currentTimeMillis();
    String message = generateEvent(currTime, delayInMillis);
    PubsubMessage pubsubMessage = new PubsubMessage()
            .encodeData(message.getBytes("UTF-8"));
    pubsubMessage.setAttributes(
        ImmutableMap.of(TIMESTAMP_ATTRIBUTE,
            Long.toString((currTime - delayInMillis) / 1000 * 1000)));
    if (delayInMillis != 0) {
      System.out.println(pubsubMessage.getAttributes());
      System.out.println("late data for: " + message);
    }
    pubsubMessages.add(pubsubMessage);
  }

  PublishRequest publishRequest = new PublishRequest();
  publishRequest.setMessages(pubsubMessages);
  pubsub.projects().topics().publish(topic, publishRequest).execute();
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
@Override
public int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages)
    throws IOException {
  List<PubsubMessage> pubsubMessages = new ArrayList<>(outgoingMessages.size());
  for (OutgoingMessage outgoingMessage : outgoingMessages) {
    PubsubMessage pubsubMessage = new PubsubMessage().encodeData(outgoingMessage.elementBytes);
    pubsubMessage.setAttributes(getMessageAttributes(outgoingMessage));
    pubsubMessages.add(pubsubMessage);
  }
  PublishRequest request = new PublishRequest().setMessages(pubsubMessages);
  PublishResponse response = pubsub.projects()
                                   .topics()
                                   .publish(topic.getPath(), request)
                                   .execute();
  return response.getMessageIds().size();
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
@Test
public void publishOneMessage() throws IOException {
  String expectedTopic = TOPIC.getPath();
  PubsubMessage expectedPubsubMessage = new PubsubMessage()
      .encodeData(DATA.getBytes())
      .setAttributes(
          ImmutableMap.<String, String> builder()
                  .put(TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME))
                  .put(ID_ATTRIBUTE, RECORD_ID)
                  .put("k", "v").build());
  PublishRequest expectedRequest = new PublishRequest()
      .setMessages(ImmutableList.of(expectedPubsubMessage));
  PublishResponse expectedResponse = new PublishResponse()
      .setMessageIds(ImmutableList.of(MESSAGE_ID));
  Mockito.when((Object) (mockPubsub.projects()
                              .topics()
                              .publish(expectedTopic, expectedRequest)
                              .execute()))
         .thenReturn(expectedResponse);
  Map<String, String> attrs = new HashMap<>();
  attrs.put("k", "v");
  OutgoingMessage actualMessage = new OutgoingMessage(
          DATA.getBytes(), attrs, MESSAGE_TIME, RECORD_ID);
  int n = client.publish(TOPIC, ImmutableList.of(actualMessage));
  assertEquals(1, n);
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
@Test
public void publishOneMessageWithOnlyTimestampAndIdAttributes() throws IOException {
  String expectedTopic = TOPIC.getPath();
  PubsubMessage expectedPubsubMessage = new PubsubMessage()
      .encodeData(DATA.getBytes())
      .setAttributes(
          ImmutableMap.<String, String> builder()
                  .put(TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME))
                  .put(ID_ATTRIBUTE, RECORD_ID).build());
  PublishRequest expectedRequest = new PublishRequest()
      .setMessages(ImmutableList.of(expectedPubsubMessage));
  PublishResponse expectedResponse = new PublishResponse()
      .setMessageIds(ImmutableList.of(MESSAGE_ID));
  Mockito.when((Object) (mockPubsub.projects()
                              .topics()
                              .publish(expectedTopic, expectedRequest)
                              .execute()))
         .thenReturn(expectedResponse);
  OutgoingMessage actualMessage = new OutgoingMessage(
      DATA.getBytes(), ImmutableMap.<String, String>of(), MESSAGE_TIME, RECORD_ID);
  int n = client.publish(TOPIC, ImmutableList.of(actualMessage));
  assertEquals(1, n);
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
/**
 * Publishes the given message to the given topic.
 *
 * @param client Cloud Pub/Sub client.
 * @param args Command line arguments.
 * @throws IOException when Cloud Pub/Sub API calls fail.
 */
public static void publishMessage(final Pubsub client, final String[] args)
        throws IOException {
    Main.checkArgsLength(args, 4);
    String topic = PubsubUtils.getFullyQualifiedResourceName(
            PubsubUtils.ResourceType.TOPIC, args[0], args[2]);
    String message = args[3];
    PubsubMessage pubsubMessage = new PubsubMessage()
            .encodeData(message.getBytes("UTF-8"));
    List<PubsubMessage> messages = ImmutableList.of(pubsubMessage);
    PublishRequest publishRequest = new PublishRequest();
    publishRequest.setMessages(messages);
    PublishResponse publishResponse = client.projects().topics()
            .publish(topic, publishRequest)
            .execute();
    List<String> messageIds = publishResponse.getMessageIds();
    if (messageIds != null) {
        for (String messageId : messageIds) {
            System.out.println("Published with a message id: " + messageId);
        }
    }
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
/**
 * Publishes the given message to the given topic.
 */
public void publishMessage(String message, String outputTopic) {
  int maxLogMessageLength = 200;
  if (message.length() < maxLogMessageLength) {
    maxLogMessageLength = message.length();
  }
  logger.info("Received ...." + message.substring(0, maxLogMessageLength));

  // Publish message to Pubsub.
  PubsubMessage pubsubMessage = new PubsubMessage();
  pubsubMessage.encodeData(message.getBytes());

  PublishRequest publishRequest = new PublishRequest();
  publishRequest.setTopic(outputTopic).setMessage(pubsubMessage);
  try {
    this.pubsub.topics().publish(publishRequest).execute();
  } catch (java.io.IOException e) {
    logger.warning(e.getStackTrace().toString());
  }
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
/**
 * Publishes the given message to a Cloud Pub/Sub topic.
 */
public void publishMessage(String message, String outputTopic) {
  int maxLogMessageLength = 200;
  if (message.length() < maxLogMessageLength) {
    maxLogMessageLength = message.length();
  }
  logger.info("Received ...." + message.substring(0, maxLogMessageLength));

  // Publish message to Pubsub.
  PubsubMessage pubsubMessage = new PubsubMessage();
  pubsubMessage.encodeData(message.getBytes());

  PublishRequest publishRequest = new PublishRequest();
  publishRequest.setTopic(outputTopic).setMessage(pubsubMessage);
  try {
    this.pubsub.topics().publish(publishRequest).execute();
  } catch (java.io.IOException e) {
    ;
  }
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
@Override
public final void doPost(final HttpServletRequest req,
                         final HttpServletResponse resp)
        throws IOException {
    Pubsub client = PubsubUtils.getClient();
    String message = req.getParameter("message");
    if (!"".equals(message)) {
        String fullTopicName = String.format("projects/%s/topics/%s",
                PubsubUtils.getProjectId(),
                PubsubUtils.getAppTopicName());
        PubsubMessage pubsubMessage = new PubsubMessage();
        pubsubMessage.encodeData(message.getBytes("UTF-8"));
        PublishRequest publishRequest = new PublishRequest();
        publishRequest.setMessages(ImmutableList.of(pubsubMessage));

        client.projects().topics()
                .publish(fullTopicName, publishRequest)
                .execute();
    }
    resp.setStatus(HttpServletResponse.SC_NO_CONTENT);
    resp.getWriter().close();
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
@Override
public void processElement(ProcessContext c) throws IOException {
  if (c.element().isEmpty()) {
    return;
  }
  PubsubMessage pubsubMessage = new PubsubMessage();
  pubsubMessage.encodeData(c.element().getBytes());
  if (timestampLabelKey != null) {
    pubsubMessage.setAttributes(
        ImmutableMap.of(timestampLabelKey, Long.toString(c.timestamp().getMillis())));
  }
  PublishRequest publishRequest = new PublishRequest();
  publishRequest.setMessages(Arrays.asList(pubsubMessage));
  this.pubsub.projects().topics().publish(outputTopic, publishRequest).execute();
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
@Test
public void publishOneMessageWithNoTimestampOrIdAttribute() throws IOException {
  // For this test, create a new PubsubJsonClient without the timestamp attribute
  // or id attribute set.
  client = new PubsubJsonClient(null, null, mockPubsub);

  String expectedTopic = TOPIC.getPath();
  PubsubMessage expectedPubsubMessage = new PubsubMessage()
      .encodeData(DATA.getBytes())
      .setAttributes(
          ImmutableMap.<String, String> builder()
                  .put("k", "v").build());
  PublishRequest expectedRequest = new PublishRequest()
      .setMessages(ImmutableList.of(expectedPubsubMessage));
  PublishResponse expectedResponse = new PublishResponse()
      .setMessageIds(ImmutableList.of(MESSAGE_ID));
  Mockito.when((Object) (mockPubsub.projects()
                              .topics()
                              .publish(expectedTopic, expectedRequest)
                              .execute()))
         .thenReturn(expectedResponse);
  Map<String, String> attrs = new HashMap<>();
  attrs.put("k", "v");
  OutgoingMessage actualMessage = new OutgoingMessage(
          DATA.getBytes(), attrs, MESSAGE_TIME, RECORD_ID);
  int n = client.publish(TOPIC, ImmutableList.of(actualMessage));
  assertEquals(1, n);
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
/**
 * Publishes a message to Google Cloud Pub/Sub.
 * TODO(rshanky) - Provide config option for automatic topic creation on publish/subscribe through
 * Google Cloud Pub/Sub.
 *
 * @param msg the message and attributes to be published is contained in this object.
 * @throws IOException is thrown on Google Cloud Pub/Sub publish(or createTopic) API failure.
 */
@Override
public void publish(PublishMessage msg) throws IOException {
  String publishTopic = createFullGcloudPubsubTopic(createPubSubTopic(msg.getMqttTopic()));
  PubsubMessage pubsubMessage = new PubsubMessage();
  byte[] payload = convertMqttPayloadToGcloudPayload(msg.getMqttPaylaod());
  pubsubMessage.setData(new String(payload));
  // create attributes for the message
  Map<String, String> attributes = ImmutableMap.of(MQTT_CLIENT_ID, msg.getMqttClientId(),
      MQTT_TOPIC_NAME, msg.getMqttTopic(),
      MQTT_MESSAGE_ID, msg.getMqttMessageId() == null ? "" : msg.getMqttMessageId().toString(),
      MQTT_RETAIN, msg.isMqttMessageRetained().toString(),
      PROXY_SERVER_ID, serverName);
  pubsubMessage.setAttributes(attributes);
  // publish message
  List<PubsubMessage> messages = ImmutableList.of(pubsubMessage);
  PublishRequest publishRequest = new PublishRequest().setMessages(messages);
  try {
    pubsub.projects().topics().publish(publishTopic, publishRequest).execute();
  } catch (GoogleJsonResponseException e) {
    if (e.getStatusCode() == RESOURCE_NOT_FOUND) {
      logger.info("Cloud PubSub Topic Not Found");
      createTopic(publishTopic);
      pubsub.projects().topics().publish(publishTopic, publishRequest).execute();
    } else {
      // re-throw the exception so that we do not send a PUBACK
      throw e;
    }
  }
  logger.info("Google Cloud Pub/Sub publish SUCCESS for topic " + publishTopic);
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
public static void main(String[] args)
    throws IOException, GeneralSecurityException {

  Pubsub pubsubClient = ServiceAccountConfiguration.createPubsubClient(
      Settings.getSettings().getServiceAccountEmail(),
      Settings.getSettings().getServiceAccountP12KeyPath());
  String topicName = Settings.getSettings().getTopicName();

  try {
    Topic topic = pubsubClient
        .projects()
        .topics()
        .get(topicName)
        .execute();

    LOG.info("The topic " + topicName + " exists: " + topic.toPrettyString());
  } catch (HttpResponseException e) {
    if (e.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) {
      // The topic doesn't exist
      LOG.info("The topic " + topicName + " doesn't exist, creating it");
      
      // TODO(kirillov): add explicit error handling here
      pubsubClient
          .projects()
          .topics()
          .create(topicName, new Topic())
          .execute();
      LOG.info("The topic " + topicName + " created");
    }
  }

  ImmutableList.Builder<PubsubMessage> listBuilder = ImmutableList.builder();

  EmmPubsub.MdmPushNotification mdmPushNotification = EmmPubsub.MdmPushNotification.newBuilder()
      .setEnterpriseId("12321321")
      .setEventNotificationSentTimestampMillis(System.currentTimeMillis())
      .addProductApprovalEvent(EmmPubsub.ProductApprovalEvent.newBuilder()
          .setApproved(EmmPubsub.ProductApprovalEvent.ApprovalStatus.UNAPPROVED)
          .setProductId("app:com.android.chrome"))
      .build();

  PublishRequest publishRequest = new PublishRequest()
      .setMessages(ImmutableList.of(new PubsubMessage()
          .encodeData(mdmPushNotification.toByteArray())));

  LOG.info("Publishing a request: " + publishRequest.toPrettyString());

  pubsubClient
      .projects()
      .topics()
      .publish(topicName, publishRequest)
      .execute();
}
 

import com.google.api.services.pubsub.model.PublishRequest; //导入依赖的package包/类
/**
 * Connects an IRC channel and publish the chat messages to the given topic.
 *
 * @param client Cloud Pub/Sub client.
 * @param args Command line arguments.
 * @throws IOException when Cloud Pub/Sub API calls fail.
 */
public static void connectIrc(final Pubsub client, final String[] args)
        throws IOException {
    Main.checkArgsLength(args, 5);
    final String server = args[3];
    final String channel = args[4];
    final String topic = PubsubUtils.getFullyQualifiedResourceName(
            PubsubUtils.ResourceType.TOPIC, args[0], args[2]);
    final String nick = String.format("bot-%s", args[0]);
    final Socket socket = new Socket(server, PORT);
    final BufferedWriter writer = new BufferedWriter(
            new OutputStreamWriter(socket.getOutputStream()));
    final BufferedReader reader = new BufferedReader(
            new InputStreamReader(socket.getInputStream()));

    writer.write(String.format("NICK %s\r\n", nick));
    writer.write(String.format("USER %s 8 * : %s\r\n", nick, BOT_NAME));
    writer.flush();

    String line;
    while ((line = reader.readLine()) != null) {
        if (line.contains("004")) {
            System.out.printf("Connected to %s.\n", server);
            break;
        } else if (line.contains("433")) {
            System.err.println("Nickname is already in use.");
            return;
        }
    }

    writer.write(String.format("JOIN %s\r\n", channel));
    writer.flush();

    // A regex pattern for Wikipedia change log as of June 4, 2014
    Pattern pat = Pattern.compile(
            "\\u000314\\[\\[\\u000307(.*)\\u000314\\]\\]\\u0003.*"
                    + "\\u000302(http://[^\\u0003]*)\\u0003");
    while ((line = reader.readLine()) != null) {
        if (line.toLowerCase().startsWith("PING ")) {
            // We must respond to PINGs to avoid being disconnected.
            writer.write("PONG " + line.substring(5) + "\r\n");
            writer.write("PRIVMSG " + channel + " :I got pinged!\r\n");
            writer.flush();
        } else {
            String privmsgMark = "PRIVMSG " + channel + " :";
            int prividx = line.indexOf(privmsgMark);
            if (prividx == -1) {
                continue;
            }
            line = line.substring(prividx + privmsgMark.length(),
                                  line.length());
            PubsubMessage pubsubMessage = new PubsubMessage();
            Matcher matcher = pat.matcher(line);
            if (matcher.find()) {
                String message = String.format("Title: %s, Diff: %s",
                        matcher.group(1), matcher.group(2));
                pubsubMessage.encodeData(message.getBytes("UTF-8"));
            } else {
                pubsubMessage.encodeData(line.getBytes("UTF-8"));
            }
            List<PubsubMessage> messages = ImmutableList.of(pubsubMessage);
            PublishRequest publishRequest = new PublishRequest();
            publishRequest.setMessages(messages);
            client.projects().topics()
                    .publish(topic, publishRequest)
                    .execute();
        }
    }
}