sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubClient.java - beam - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.beam.sdk.io.gcp.pubsub;

 import static org.apache.beam.vendor.guava.v20_0.com.google.common.base.Preconditions.checkArgument;
 import static org.apache.beam.vendor.guava.v20_0.com.google.common.base.Preconditions.checkState;

 import com.google.api.client.util.DateTime;
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 import javax.annotation.Nullable;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Objects;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Splitter;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Strings;

 /** An (abstract) helper class for talking to Pubsub via an underlying transport. */
 public abstract class PubsubClient implements Closeable {
   /** Factory for creating clients. */
   public interface PubsubClientFactory extends Serializable {
     /**
      * Construct a new Pubsub client. It should be closed via {@link #close} in order to ensure tidy
      * cleanup of underlying netty resources (or use the try-with-resources construct). Uses {@code
      * options} to derive pubsub endpoints and application credentials. If non-{@literal null}, use
      * {@code timestampAttribute} and {@code idAttribute} to store custom timestamps/ids within
      * message metadata.
      */
     PubsubClient newClient(
         @Nullable String timestampAttribute, @Nullable String idAttribute, PubsubOptions options)
         throws IOException;

     /** Return the display name for this factory. Eg "Json", "gRPC". */
     String getKind();
   }

   /**
    * Return timestamp as ms-since-unix-epoch corresponding to {@code timestamp}. Return {@literal
    * null} if no timestamp could be found. Throw {@link IllegalArgumentException} if timestamp
    * cannot be recognized.
    */
   @Nullable
   private static Long asMsSinceEpoch(@Nullable String timestamp) {
     if (Strings.isNullOrEmpty(timestamp)) {
       return null;
     }
     try {
       // Try parsing as milliseconds since epoch. Note there is no way to parse a
       // string in RFC 3339 format here.
       // Expected IllegalArgumentException if parsing fails; we use that to fall back
       // to RFC 3339.
       return Long.parseLong(timestamp);
     } catch (IllegalArgumentException e1) {
       // Try parsing as RFC3339 string. DateTime.parseRfc3339 will throw an
       // IllegalArgumentException if parsing fails, and the caller should handle.
       return DateTime.parseRfc3339(timestamp).getValue();
     }
   }

   /**
    * Return the timestamp (in ms since unix epoch) to use for a Pubsub message with {@code
    * attributes} and {@code pubsubTimestamp}.
    *
    * <p>If {@code timestampAttribute} is non-{@literal null} then the message attributes must
    * contain that attribute, and the value of that attribute will be taken as the timestamp.
    * Otherwise the timestamp will be taken from the Pubsub publish timestamp {@code
    * pubsubTimestamp}.
    *
    * @throws IllegalArgumentException if the timestamp cannot be recognized as a ms-since-unix-epoch
    *     or RFC3339 time.
    */
   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;
   }

   /** Path representing a cloud project id. */
   public static class ProjectPath implements Serializable {
     private final String projectId;

     /**
      * Creates a {@link ProjectPath} from a {@link String} representation, which must be of the form
      * {@code "projects/" + projectId}.
      */
     ProjectPath(String path) {
       List<String> splits = Splitter.on('/').splitToList(path);
       checkArgument(
           splits.size() == 2 && "projects".equals(splits.get(0)),
           "Malformed project path \"%s\": must be of the form \"projects/\" + <project id>",
           path);
       this.projectId = splits.get(1);
     }

     public String getPath() {
       return String.format("projects/%s", projectId);
     }

     public String getId() {
       return projectId;
     }

     @Override
     public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
       if (o == null || getClass() != o.getClass()) {
         return false;
       }

       ProjectPath that = (ProjectPath) o;

       return projectId.equals(that.projectId);
     }

     @Override
     public int hashCode() {
       return projectId.hashCode();
     }

     @Override
     public String toString() {
       return getPath();
     }
   }

   public static ProjectPath projectPathFromPath(String path) {
     return new ProjectPath(path);
   }

   public static ProjectPath projectPathFromId(String projectId) {
     return new ProjectPath(String.format("projects/%s", projectId));
   }

   /** Path representing a Pubsub subscription. */
   public static class SubscriptionPath implements Serializable {
     private final String projectId;
     private final String subscriptionName;

     SubscriptionPath(String path) {
       List<String> splits = Splitter.on('/').splitToList(path);
       checkState(
           splits.size() == 4
               && "projects".equals(splits.get(0))
               && "subscriptions".equals(splits.get(2)),
           "Malformed subscription path %s: "
               + "must be of the form \"projects/\" + <project id> + \"subscriptions\"",
           path);
       this.projectId = splits.get(1);
       this.subscriptionName = splits.get(3);
     }

     public String getPath() {
       return String.format("projects/%s/subscriptions/%s", projectId, subscriptionName);
     }

     public String getName() {
       return subscriptionName;
     }

     public String getV1Beta1Path() {
       return String.format("/subscriptions/%s/%s", projectId, subscriptionName);
     }

     @Override
     public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
       if (o == null || getClass() != o.getClass()) {
         return false;
       }
       SubscriptionPath that = (SubscriptionPath) o;
       return this.subscriptionName.equals(that.subscriptionName)
           && this.projectId.equals(that.projectId);
     }

     @Override
     public int hashCode() {
       return Objects.hashCode(projectId, subscriptionName);
     }

     @Override
     public String toString() {
       return getPath();
     }
   }

   public static SubscriptionPath subscriptionPathFromPath(String path) {
     return new SubscriptionPath(path);
   }

   public static SubscriptionPath subscriptionPathFromName(
       String projectId, String subscriptionName) {
     return new SubscriptionPath(
         String.format("projects/%s/subscriptions/%s", projectId, subscriptionName));
   }

   /** Path representing a Pubsub topic. */
   public static class TopicPath implements Serializable {
     private final String path;

     TopicPath(String path) {
       this.path = path;
     }

     public String getPath() {
       return path;
     }

     public String getName() {
       List<String> splits = Splitter.on('/').splitToList(path);

       checkState(splits.size() == 4, "Malformed topic path %s", path);
       return splits.get(3);
     }

     public String getV1Beta1Path() {
       List<String> splits = Splitter.on('/').splitToList(path);
       checkState(splits.size() == 4, "Malformed topic path %s", path);
       return String.format("/topics/%s/%s", splits.get(1), splits.get(3));
     }

     @Override
     public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
       if (o == null || getClass() != o.getClass()) {
         return false;
       }
       TopicPath topicPath = (TopicPath) o;
       return path.equals(topicPath.path);
     }

     @Override
     public int hashCode() {
       return path.hashCode();
     }

     @Override
     public String toString() {
       return path;
     }
   }

   public static TopicPath topicPathFromPath(String path) {
     return new TopicPath(path);
   }

   public static TopicPath topicPathFromName(String projectId, String topicName) {
     return new TopicPath(String.format("projects/%s/topics/%s", projectId, topicName));
   }

   /**
    * A message to be sent to Pubsub.
    *
    * <p>NOTE: This class is {@link Serializable} only to support the {@link PubsubTestClient}. Java
    * serialization is never used for non-test clients.
    */
   public static class OutgoingMessage implements Serializable {
     /** Underlying (encoded) element. */
     public final byte[] elementBytes;

     public final Map<String, String> attributes;

     /** Timestamp for element (ms since epoch). */
     public final long timestampMsSinceEpoch;

     /**
      * If using an id attribute, the record id to associate with this record's metadata so the
      * receiver can reject duplicates. Otherwise {@literal null}.
      */
     @Nullable public final String recordId;

     public OutgoingMessage(
         byte[] elementBytes,
         Map<String, String> attributes,
         long timestampMsSinceEpoch,
         @Nullable String recordId) {
       this.elementBytes = elementBytes;
       this.attributes = attributes;
       this.timestampMsSinceEpoch = timestampMsSinceEpoch;
       this.recordId = recordId;
     }

     @Override
     public String toString() {
       return String.format(
           "OutgoingMessage(%db, %dms)", elementBytes.length, timestampMsSinceEpoch);
     }

     @Override
     public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
       if (o == null || getClass() != o.getClass()) {
         return false;
       }

       OutgoingMessage that = (OutgoingMessage) o;

       return timestampMsSinceEpoch == that.timestampMsSinceEpoch
           && Arrays.equals(elementBytes, that.elementBytes)
           && Objects.equal(attributes, that.attributes)
           && Objects.equal(recordId, that.recordId);
     }

     @Override
     public int hashCode() {
       return Objects.hashCode(
           Arrays.hashCode(elementBytes), attributes, timestampMsSinceEpoch, recordId);
     }
   }

   /**
    * A message received from Pubsub.
    *
    * <p>NOTE: This class is {@link Serializable} only to support the {@link PubsubTestClient}. Java
    * serialization is never used for non-test clients.
    */
   static class IncomingMessage implements Serializable {
     /** Underlying (encoded) element. */
     public final byte[] elementBytes;

     public Map<String, String> attributes;

     /**
      * Timestamp for element (ms since epoch). Either Pubsub's processing time, or the custom
      * timestamp associated with the message.
      */
     public final long timestampMsSinceEpoch;

     /** Timestamp (in system time) at which we requested the message (ms since epoch). */
     public final long requestTimeMsSinceEpoch;

     /** Id to pass back to Pubsub to acknowledge receipt of this message. */
     public final String ackId;

     /** Id to pass to the runner to distinguish this message from all others. */
     public final String recordId;

     public IncomingMessage(
         byte[] elementBytes,
         Map<String, String> attributes,
         long timestampMsSinceEpoch,
         long requestTimeMsSinceEpoch,
         String ackId,
         String recordId) {
       this.elementBytes = elementBytes;
       this.attributes = attributes;
       this.timestampMsSinceEpoch = timestampMsSinceEpoch;
       this.requestTimeMsSinceEpoch = requestTimeMsSinceEpoch;
       this.ackId = ackId;
       this.recordId = recordId;
     }

     public IncomingMessage withRequestTime(long requestTimeMsSinceEpoch) {
       return new IncomingMessage(
           elementBytes,
           attributes,
           timestampMsSinceEpoch,
           requestTimeMsSinceEpoch,
           ackId,
           recordId);
     }

     @Override
     public String toString() {
       return String.format(
           "IncomingMessage(%db, %dms)", elementBytes.length, timestampMsSinceEpoch);
     }

     @Override
     public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
       if (o == null || getClass() != o.getClass()) {
         return false;
       }

       IncomingMessage that = (IncomingMessage) o;

       return timestampMsSinceEpoch == that.timestampMsSinceEpoch
           && requestTimeMsSinceEpoch == that.requestTimeMsSinceEpoch
           && ackId.equals(that.ackId)
           && recordId.equals(that.recordId)
           && Arrays.equals(elementBytes, that.elementBytes)
           && Objects.equal(attributes, that.attributes);
     }

     @Override
     public int hashCode() {
       return Objects.hashCode(
           Arrays.hashCode(elementBytes),
           attributes,
           timestampMsSinceEpoch,
           requestTimeMsSinceEpoch,
           ackId,
           recordId);
     }
   }

   /**
    * Publish {@code outgoingMessages} to Pubsub {@code topic}. Return number of messages published.
    */
   public abstract int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages)
       throws IOException;

   /**
    * Request the next batch of up to {@code batchSize} messages from {@code subscription}. Return
    * the received messages, or empty collection if none were available. Does not wait for messages
    * to arrive if {@code returnImmediately} is {@literal true}. Returned messages will record their
    * request time as {@code requestTimeMsSinceEpoch}.
    */
   public abstract List<IncomingMessage> pull(
       long requestTimeMsSinceEpoch,
       SubscriptionPath subscription,
       int batchSize,
       boolean returnImmediately)
       throws IOException;

   /** Acknowldege messages from {@code subscription} with {@code ackIds}. */
   public abstract void acknowledge(SubscriptionPath subscription, List<String> ackIds)
       throws IOException;

   /**
    * Modify the ack deadline for messages from {@code subscription} with {@code ackIds} to be {@code
    * deadlineSeconds} from now.
    */
   public abstract void modifyAckDeadline(
       SubscriptionPath subscription, List<String> ackIds, int deadlineSeconds) throws IOException;

   /** Create {@code topic}. */
   public abstract void createTopic(TopicPath topic) throws IOException;

   /*
    * Delete {@code topic}.
    */
   public abstract void deleteTopic(TopicPath topic) throws IOException;

   /** Return a list of topics for {@code project}. */
   public abstract List<TopicPath> listTopics(ProjectPath project) throws IOException;

   /** Create {@code subscription} to {@code topic}. */
   public abstract void createSubscription(
       TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds) throws IOException;

   /**
    * Create a random subscription for {@code topic}. Return the {@link SubscriptionPath}. It is the
    * responsibility of the caller to later delete the subscription.
    */
   public SubscriptionPath createRandomSubscription(
       ProjectPath project, TopicPath topic, int ackDeadlineSeconds) throws IOException {
     // Create a randomized subscription derived from the topic name.
     String subscriptionName = topic.getName() + "_beam_" + ThreadLocalRandom.current().nextLong();
     SubscriptionPath subscription =
         PubsubClient.subscriptionPathFromName(project.getId(), subscriptionName);
     createSubscription(topic, subscription, ackDeadlineSeconds);
     return subscription;
   }

   /** Delete {@code subscription}. */
   public abstract void deleteSubscription(SubscriptionPath subscription) throws IOException;

   /** Return a list of subscriptions for {@code topic} in {@code project}. */
   public abstract List<SubscriptionPath> listSubscriptions(ProjectPath project, TopicPath topic)
       throws IOException;

   /** Return the ack deadline, in seconds, for {@code subscription}. */
   public abstract int ackDeadlineSeconds(SubscriptionPath subscription) throws IOException;

   /**
    * Return {@literal true} if {@link #pull} will always return empty list. Actual clients will
    * return {@literal false}. Test clients may return {@literal true} to signal that all expected
    * messages have been pulled and the test may complete.
    */
   public abstract boolean isEOF();
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.beam.sdk.io.gcp.pubsub;

	import static org.apache.beam.vendor.guava.v20_0.com.google.common.base.Preconditions.checkArgument;
	import static org.apache.beam.vendor.guava.v20_0.com.google.common.base.Preconditions.checkState;

	import com.google.api.client.util.DateTime;
	import java.io.Closeable;
	import java.io.IOException;
	import java.io.Serializable;
	import java.util.Arrays;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.ThreadLocalRandom;
	import javax.annotation.Nullable;
	import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Objects;
	import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Splitter;
	import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Strings;

	/** An (abstract) helper class for talking to Pubsub via an underlying transport. */
	public abstract class PubsubClient implements Closeable {
	/** Factory for creating clients. */
	public interface PubsubClientFactory extends Serializable {
	/**
	* Construct a new Pubsub client. It should be closed via {@link #close} in order to ensure tidy
	* cleanup of underlying netty resources (or use the try-with-resources construct). Uses {@code
	* options} to derive pubsub endpoints and application credentials. If non-{@literal null}, use
	* {@code timestampAttribute} and {@code idAttribute} to store custom timestamps/ids within
	* message metadata.
	*/
	PubsubClient newClient(
	@Nullable String timestampAttribute, @Nullable String idAttribute, PubsubOptions options)
	throws IOException;

	/** Return the display name for this factory. Eg "Json", "gRPC". */
	String getKind();
	}

	/**
	* Return timestamp as ms-since-unix-epoch corresponding to {@code timestamp}. Return {@literal
	* null} if no timestamp could be found. Throw {@link IllegalArgumentException} if timestamp
	* cannot be recognized.
	*/
	@Nullable
	private static Long asMsSinceEpoch(@Nullable String timestamp) {
	if (Strings.isNullOrEmpty(timestamp)) {
	return null;
	}
	try {
	// Try parsing as milliseconds since epoch. Note there is no way to parse a
	// string in RFC 3339 format here.
	// Expected IllegalArgumentException if parsing fails; we use that to fall back
	// to RFC 3339.
	return Long.parseLong(timestamp);
	} catch (IllegalArgumentException e1) {
	// Try parsing as RFC3339 string. DateTime.parseRfc3339 will throw an
	// IllegalArgumentException if parsing fails, and the caller should handle.
	return DateTime.parseRfc3339(timestamp).getValue();
	}
	}

	/**
	* Return the timestamp (in ms since unix epoch) to use for a Pubsub message with {@code
	* attributes} and {@code pubsubTimestamp}.
	*
	* <p>If {@code timestampAttribute} is non-{@literal null} then the message attributes must
	* contain that attribute, and the value of that attribute will be taken as the timestamp.
	* Otherwise the timestamp will be taken from the Pubsub publish timestamp {@code
	* pubsubTimestamp}.
	*
	* @throws IllegalArgumentException if the timestamp cannot be recognized as a ms-since-unix-epoch
	* or RFC3339 time.
	*/
	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;
	}

	/** Path representing a cloud project id. */
	public static class ProjectPath implements Serializable {
	private final String projectId;

	/**
	* Creates a {@link ProjectPath} from a {@link String} representation, which must be of the form
	* {@code "projects/" + projectId}.
	*/
	ProjectPath(String path) {
	List<String> splits = Splitter.on('/').splitToList(path);
	checkArgument(
	splits.size() == 2 && "projects".equals(splits.get(0)),
	"Malformed project path \"%s\": must be of the form \"projects/\" + <project id>",
	path);
	this.projectId = splits.get(1);
	}

	public String getPath() {
	return String.format("projects/%s", projectId);
	}

	public String getId() {
	return projectId;
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}

	ProjectPath that = (ProjectPath) o;

	return projectId.equals(that.projectId);
	}

	@Override
	public int hashCode() {
	return projectId.hashCode();
	}

	@Override
	public String toString() {
	return getPath();
	}
	}

	public static ProjectPath projectPathFromPath(String path) {
	return new ProjectPath(path);
	}

	public static ProjectPath projectPathFromId(String projectId) {
	return new ProjectPath(String.format("projects/%s", projectId));
	}

	/** Path representing a Pubsub subscription. */
	public static class SubscriptionPath implements Serializable {
	private final String projectId;
	private final String subscriptionName;

	SubscriptionPath(String path) {
	List<String> splits = Splitter.on('/').splitToList(path);
	checkState(
	splits.size() == 4
	&& "projects".equals(splits.get(0))
	&& "subscriptions".equals(splits.get(2)),
	"Malformed subscription path %s: "
	+ "must be of the form \"projects/\" + <project id> + \"subscriptions\"",
	path);
	this.projectId = splits.get(1);
	this.subscriptionName = splits.get(3);
	}

	public String getPath() {
	return String.format("projects/%s/subscriptions/%s", projectId, subscriptionName);
	}

	public String getName() {
	return subscriptionName;
	}

	public String getV1Beta1Path() {
	return String.format("/subscriptions/%s/%s", projectId, subscriptionName);
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}
	SubscriptionPath that = (SubscriptionPath) o;
	return this.subscriptionName.equals(that.subscriptionName)
	&& this.projectId.equals(that.projectId);
	}

	@Override
	public int hashCode() {
	return Objects.hashCode(projectId, subscriptionName);
	}

	@Override
	public String toString() {
	return getPath();
	}
	}

	public static SubscriptionPath subscriptionPathFromPath(String path) {
	return new SubscriptionPath(path);
	}

	public static SubscriptionPath subscriptionPathFromName(
	String projectId, String subscriptionName) {
	return new SubscriptionPath(
	String.format("projects/%s/subscriptions/%s", projectId, subscriptionName));
	}

	/** Path representing a Pubsub topic. */
	public static class TopicPath implements Serializable {
	private final String path;

	TopicPath(String path) {
	this.path = path;
	}

	public String getPath() {
	return path;
	}

	public String getName() {
	List<String> splits = Splitter.on('/').splitToList(path);

	checkState(splits.size() == 4, "Malformed topic path %s", path);
	return splits.get(3);
	}

	public String getV1Beta1Path() {
	List<String> splits = Splitter.on('/').splitToList(path);
	checkState(splits.size() == 4, "Malformed topic path %s", path);
	return String.format("/topics/%s/%s", splits.get(1), splits.get(3));
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}
	TopicPath topicPath = (TopicPath) o;
	return path.equals(topicPath.path);
	}

	@Override
	public int hashCode() {
	return path.hashCode();
	}

	@Override
	public String toString() {
	return path;
	}
	}

	public static TopicPath topicPathFromPath(String path) {
	return new TopicPath(path);
	}

	public static TopicPath topicPathFromName(String projectId, String topicName) {
	return new TopicPath(String.format("projects/%s/topics/%s", projectId, topicName));
	}

	/**
	* A message to be sent to Pubsub.
	*
	* <p>NOTE: This class is {@link Serializable} only to support the {@link PubsubTestClient}. Java
	* serialization is never used for non-test clients.
	*/
	public static class OutgoingMessage implements Serializable {
	/** Underlying (encoded) element. */
	public final byte[] elementBytes;

	public final Map<String, String> attributes;

	/** Timestamp for element (ms since epoch). */
	public final long timestampMsSinceEpoch;

	/**
	* If using an id attribute, the record id to associate with this record's metadata so the
	* receiver can reject duplicates. Otherwise {@literal null}.
	*/
	@Nullable public final String recordId;

	public OutgoingMessage(
	byte[] elementBytes,
	Map<String, String> attributes,
	long timestampMsSinceEpoch,
	@Nullable String recordId) {
	this.elementBytes = elementBytes;
	this.attributes = attributes;
	this.timestampMsSinceEpoch = timestampMsSinceEpoch;
	this.recordId = recordId;
	}

	@Override
	public String toString() {
	return String.format(
	"OutgoingMessage(%db, %dms)", elementBytes.length, timestampMsSinceEpoch);
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}

	OutgoingMessage that = (OutgoingMessage) o;

	return timestampMsSinceEpoch == that.timestampMsSinceEpoch
	&& Arrays.equals(elementBytes, that.elementBytes)
	&& Objects.equal(attributes, that.attributes)
	&& Objects.equal(recordId, that.recordId);
	}

	@Override
	public int hashCode() {
	return Objects.hashCode(
	Arrays.hashCode(elementBytes), attributes, timestampMsSinceEpoch, recordId);
	}
	}

	/**
	* A message received from Pubsub.
	*
	* <p>NOTE: This class is {@link Serializable} only to support the {@link PubsubTestClient}. Java
	* serialization is never used for non-test clients.
	*/
	static class IncomingMessage implements Serializable {
	/** Underlying (encoded) element. */
	public final byte[] elementBytes;

	public Map<String, String> attributes;

	/**
	* Timestamp for element (ms since epoch). Either Pubsub's processing time, or the custom
	* timestamp associated with the message.
	*/
	public final long timestampMsSinceEpoch;

	/** Timestamp (in system time) at which we requested the message (ms since epoch). */
	public final long requestTimeMsSinceEpoch;

	/** Id to pass back to Pubsub to acknowledge receipt of this message. */
	public final String ackId;

	/** Id to pass to the runner to distinguish this message from all others. */
	public final String recordId;

	public IncomingMessage(
	byte[] elementBytes,
	Map<String, String> attributes,
	long timestampMsSinceEpoch,
	long requestTimeMsSinceEpoch,
	String ackId,
	String recordId) {
	this.elementBytes = elementBytes;
	this.attributes = attributes;
	this.timestampMsSinceEpoch = timestampMsSinceEpoch;
	this.requestTimeMsSinceEpoch = requestTimeMsSinceEpoch;
	this.ackId = ackId;
	this.recordId = recordId;
	}

	public IncomingMessage withRequestTime(long requestTimeMsSinceEpoch) {
	return new IncomingMessage(
	elementBytes,
	attributes,
	timestampMsSinceEpoch,
	requestTimeMsSinceEpoch,
	ackId,
	recordId);
	}

	@Override
	public String toString() {
	return String.format(
	"IncomingMessage(%db, %dms)", elementBytes.length, timestampMsSinceEpoch);
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}

	IncomingMessage that = (IncomingMessage) o;

	return timestampMsSinceEpoch == that.timestampMsSinceEpoch
	&& requestTimeMsSinceEpoch == that.requestTimeMsSinceEpoch
	&& ackId.equals(that.ackId)
	&& recordId.equals(that.recordId)
	&& Arrays.equals(elementBytes, that.elementBytes)
	&& Objects.equal(attributes, that.attributes);
	}

	@Override
	public int hashCode() {
	return Objects.hashCode(
	Arrays.hashCode(elementBytes),
	attributes,
	timestampMsSinceEpoch,
	requestTimeMsSinceEpoch,
	ackId,
	recordId);
	}
	}

	/**
	* Publish {@code outgoingMessages} to Pubsub {@code topic}. Return number of messages published.
	*/
	public abstract int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages)
	throws IOException;

	/**
	* Request the next batch of up to {@code batchSize} messages from {@code subscription}. Return
	* the received messages, or empty collection if none were available. Does not wait for messages
	* to arrive if {@code returnImmediately} is {@literal true}. Returned messages will record their
	* request time as {@code requestTimeMsSinceEpoch}.
	*/
	public abstract List<IncomingMessage> pull(
	long requestTimeMsSinceEpoch,
	SubscriptionPath subscription,
	int batchSize,
	boolean returnImmediately)
	throws IOException;

	/** Acknowldege messages from {@code subscription} with {@code ackIds}. */
	public abstract void acknowledge(SubscriptionPath subscription, List<String> ackIds)
	throws IOException;

	/**
	* Modify the ack deadline for messages from {@code subscription} with {@code ackIds} to be {@code
	* deadlineSeconds} from now.
	*/
	public abstract void modifyAckDeadline(
	SubscriptionPath subscription, List<String> ackIds, int deadlineSeconds) throws IOException;

	/** Create {@code topic}. */
	public abstract void createTopic(TopicPath topic) throws IOException;

	/*
	* Delete {@code topic}.
	*/
	public abstract void deleteTopic(TopicPath topic) throws IOException;

	/** Return a list of topics for {@code project}. */
	public abstract List<TopicPath> listTopics(ProjectPath project) throws IOException;

	/** Create {@code subscription} to {@code topic}. */
	public abstract void createSubscription(
	TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds) throws IOException;

	/**
	* Create a random subscription for {@code topic}. Return the {@link SubscriptionPath}. It is the
	* responsibility of the caller to later delete the subscription.
	*/
	public SubscriptionPath createRandomSubscription(
	ProjectPath project, TopicPath topic, int ackDeadlineSeconds) throws IOException {
	// Create a randomized subscription derived from the topic name.
	String subscriptionName = topic.getName() + "_beam_" + ThreadLocalRandom.current().nextLong();
	SubscriptionPath subscription =
	PubsubClient.subscriptionPathFromName(project.getId(), subscriptionName);
	createSubscription(topic, subscription, ackDeadlineSeconds);
	return subscription;
	}

	/** Delete {@code subscription}. */
	public abstract void deleteSubscription(SubscriptionPath subscription) throws IOException;

	/** Return a list of subscriptions for {@code topic} in {@code project}. */
	public abstract List<SubscriptionPath> listSubscriptions(ProjectPath project, TopicPath topic)
	throws IOException;

	/** Return the ack deadline, in seconds, for {@code subscription}. */
	public abstract int ackDeadlineSeconds(SubscriptionPath subscription) throws IOException;

	/**
	* Return {@literal true} if {@link #pull} will always return empty list. Actual clients will
	* return {@literal false}. Test clients may return {@literal true} to signal that all expected
	* messages have been pulled and the test may complete.
	*/
	public abstract boolean isEOF();
	}