PubsubClient implements Closeable { protected static long extractTimestamp( @Nullable String timestampAttribute, @Nullable String pubsubTimestamp, @Nullable Map<String, String> attributes) { Long timestampMsSinceEpoch; if (Strings.isNullOrEmpty(timestampAttribute)) { timestampMsSinceEpoch = asMsSinceEpoch(pubsubTimestamp); checkArgument( timestampMsSinceEpoch != null, "Cannot interpret PubSub publish timestamp: %s", pubsubTimestamp); } else { String value = attributes == null ? null : attributes.get(timestampAttribute); checkArgument( value != null, "PubSub message is missing a value for timestamp attribute %s", timestampAttribute); timestampMsSinceEpoch = asMsSinceEpoch(value); checkArgument( timestampMsSinceEpoch != null, "Cannot interpret value of attribute %s as timestamp: %s", timestampAttribute, value); } return timestampMsSinceEpoch; }  static ProjectPath projectPathFromPath(String path); static ProjectPath projectPathFromId(String projectId); static SubscriptionPath subscriptionPathFromPath(String path); static SubscriptionPath subscriptionPathFromName(
      String projectId, String subscriptionName); static TopicPath topicPathFromPath(String path); static TopicPath topicPathFromName(String projectId, String topicName); abstract int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages); abstract List<IncomingMessage> pull(
      long requestTimeMsSinceEpoch,
      SubscriptionPath subscription,
      int batchSize,
      boolean returnImmediately); abstract void acknowledge(SubscriptionPath subscription, List<String> ackIds); abstract void modifyAckDeadline(
      SubscriptionPath subscription, List<String> ackIds, int deadlineSeconds); abstract void createTopic(TopicPath topic); abstract void deleteTopic(TopicPath topic); abstract List<TopicPath> listTopics(ProjectPath project); abstract void createSubscription(
      TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds); SubscriptionPath createRandomSubscription(
      ProjectPath project, TopicPath topic, int ackDeadlineSeconds); abstract void deleteSubscription(SubscriptionPath subscription); abstract List<SubscriptionPath> listSubscriptions(ProjectPath project, TopicPath topic); abstract int ackDeadlineSeconds(SubscriptionPath subscription); abstract boolean isEOF();  }
@Test public void noTimestampAttributeReturnsPubsubPublish() { final long time = 987654321L; long timestamp = PubsubClient.extractTimestamp(null, String.valueOf(time), null); assertEquals(time, timestamp); }
@Test public void noTimestampAttributeAndInvalidPubsubPublishThrowsError() { thrown.expect(NumberFormatException.class); PubsubClient.extractTimestamp(null, "not-a-date", null); }
@Test public void timestampAttributeWithNullAttributesThrowsError() { thrown.expect(RuntimeException.class); thrown.expectMessage("PubSub message is missing a value for timestamp attribute myAttribute"); PubsubClient.extractTimestamp("myAttribute", null, null); }
@Test public void timestampAttributeSetWithMissingAttributeThrowsError() { thrown.expect(RuntimeException.class); thrown.expectMessage("PubSub message is missing a value for timestamp attribute myAttribute"); Map<String, String> map = ImmutableMap.of("otherLabel", "whatever"); PubsubClient.extractTimestamp("myAttribute", null, map); }
@Test public void timestampAttributeParsesMillisecondsSinceEpoch() { long time = 1446162101123L; Map<String, String> map = ImmutableMap.of("myAttribute", String.valueOf(time)); long timestamp = PubsubClient.extractTimestamp("myAttribute", null, map); assertEquals(time, timestamp); }