diff --git a/docs/_static/js/main.js b/docs/_static/js/main.js
index 11b8202481a07..8f95b772f38a8 100755
--- a/docs/_static/js/main.js
+++ b/docs/_static/js/main.js
@@ -16,7 +16,7 @@ $('.headerlink').parent().each(function() {
 $('.side-nav').children('ul:nth-child(2)').children().each(function() {
   var itemName = $(this).text();
     if (itemName !== 'Datastore' && itemName !== 'Storage' &&
-        itemName !== 'Pub/Sub') {
+        itemName !== 'Pub/Sub' && itemName !== 'Search') {
       $(this).css('padding-left','2em');
   }
 });
diff --git a/docs/index.rst b/docs/index.rst
index 963d2fe1cbcec..0af1385f48b32 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -9,14 +9,20 @@
   datastore-queries
   datastore-transactions
   datastore-batches
-  storage-api
-  storage-blobs
-  storage-buckets
-  storage-acl
+  datastore-dataset
   pubsub-api
   pubsub-usage
   pubsub-subscription
   pubsub-topic
+  search-api
+  search-client
+  search-index
+  search-document
+  search-field
+  storage-api
+  storage-blobs
+  storage-buckets
+  storage-acl
 
 
 Getting started
diff --git a/docs/search-api.rst b/docs/search-api.rst
new file mode 100644
index 0000000000000..4fdab1688204f
--- /dev/null
+++ b/docs/search-api.rst
@@ -0,0 +1,137 @@
+.. toctree::
+  :maxdepth: 1
+  :hidden:
+
+Search
+------
+
+Overview
+~~~~~~~~
+
+Cloud Search allows you to quickly perform full-text and geospatial searches
+against your data without having to spin up your own instances
+and without the hassle of managing and maintaining a search service.
+
+Cloud Search provides a model for indexing your documents
+that contain structured data,
+with documents and indexes saved to a separate persistent store
+optimized for search operations.
+You can search an index, and organize and present your search results.
+The API supports full text matching on string fields
+and allows you to index any number of documents in any number of indexes.
+
+Indexes
+~~~~~~~
+
+Here's an example of how you might deal with indexes::
+
+    >>> from gcloud import search
+    >>> client = search.Client()
+
+    >>> # List all indexes in your project
+    >>> for index in client.list_indexes():
+    ...     print index
+
+    >>> # Create a new index
+    >>> new_index = client.index('index-id-here')
+    >>> new_index.name = 'My new index'
+    >>> new_index.create()
+
+    >>> # Update an existing index
+    >>> index = client.get_index('existing-index-id')
+    >>> print index
+    <Index: Existing Index (existing-index-id)>
+    >>> index.name = 'Modified name'
+    >>> index.update()
+    >>> print index
+    <Index: Modified name (existing-index-id)>
+
+    >>> # Delete an index
+    >>> index = client.get_index('existing-index-id')
+    >>> index.delete()
+
+Documents
+~~~~~~~~~
+
+Documents are the things that you search for.
+The typical process is:
+
+#. Create a document
+#. Add fields to the document
+#. Add the document to an index to be searched for later
+
+Here's an example of how you might deal with documents::
+
+    >>> from gcloud import search
+    >>> client = search.Client()
+
+    >>> # Create a document
+    >>> document = search.Document('document-id')
+
+    >>> # Add a field to the document
+    >>> field = search.Field('fieldname')
+    >>> field.add_value('string')
+    >>> document.add_field(field)
+
+    >>> # Add the document to an index
+    >>> index = client.get_index('existing-index-id')
+    >>> index.add_document(document)
+
+Fields
+~~~~~~
+
+Fields belong to documents and are the data that actually gets searched.
+Each field can have multiple values,
+and there are three different types of tokenization for string values:
+
+- **Atom** (``atom``) means "don't tokenize this string", treat it as one thing
+  to compare against.
+- **Text** (``text``) means "treat this string as normal text" and split words
+  apart to be compared against.
+- **HTML** (``html``) means "treat this string as HTML", understanding the
+  tags, and treating the rest of the content like Text.
+
+You can set this using the ``tokenization`` paramater when adding a field
+value::
+
+    >>> from gcloud import search
+    >>> document = search.Document('document-id')
+    >>> document.add_field(search.Field('field-name', values=[
+    ...     search.Value('britney spears', tokenization='atom'),
+    ...     search.Value('<h1>Britney Spears</h1>', tokenization='html'),
+    ...     ]))
+
+Searching
+~~~~~~~~~
+
+Once you have indexes full of documents, you can search through them by
+issuing a search query.
+
+Here's a simple example of how you might start searching::
+
+    >>> from gcloud import search
+    >>> client = search.Client()
+
+    >>> index = client.get_index('existing-index-id')
+    >>> query = search.Query('britney spears')
+    >>> matching_documents = index.search(query)
+    >>> for document in matching_documents:
+    ...     print document
+
+By default, all queries are sorted by the ``rank`` value you set
+when the documented was created.
+If you want to sort differently, use the ``order_by`` parameter::
+
+    >>> from gcloud import search
+    >>> query = search.Query('britney spears', order_by=['field1', '-field2'])
+
+Note that the ``-`` character before ``field2`` means that
+this query will be sorted ascending by ``field1``
+and then descending by ``field2``.
+
+If you want only want certain fields to be returned in the match,
+you can use the ``fields`` paramater::
+
+    >>> from gcloud import search
+    >>> query = search.Query('britney spears', fields=['field1', 'field2'])
+
diff --git a/docs/search-client.rst b/docs/search-client.rst
new file mode 100644
index 0000000000000..09a79e5d6f203
--- /dev/null
+++ b/docs/search-client.rst
@@ -0,0 +1,19 @@
+.. toctree::
+  :maxdepth: 0
+  :hidden:
+
+Client
+------
+
+.. automodule:: gcloud.search.client
+  :members:
+  :undoc-members:
+  :show-inheritance:
+
+Connection
+~~~~~~~~~~
+
+.. automodule:: gcloud.search.connection
+  :members:
+  :undoc-members:
+  :show-inheritance:
diff --git a/docs/search-document.rst b/docs/search-document.rst
new file mode 100644
index 0000000000000..ebc6c53ed1337
--- /dev/null
+++ b/docs/search-document.rst
@@ -0,0 +1,11 @@
+.. toctree::
+  :maxdepth: 0
+  :hidden:
+
+Document
+--------
+
+.. automodule:: gcloud.search.document
+  :members:
+  :undoc-members:
+  :show-inheritance:
diff --git a/docs/search-field.rst b/docs/search-field.rst
new file mode 100644
index 0000000000000..9dcc2c9bcb885
--- /dev/null
+++ b/docs/search-field.rst
@@ -0,0 +1,19 @@
+.. toctree::
+  :maxdepth: 0
+  :hidden:
+
+Field
+-----
+
+.. automodule:: gcloud.search.field
+  :members:
+  :undoc-members:
+  :show-inheritance:
+
+Value
+~~~~~
+
+.. automodule:: gcloud.search.value
+  :members:
+  :undoc-members:
+  :show-inheritance:
diff --git a/docs/search-index.rst b/docs/search-index.rst
new file mode 100644
index 0000000000000..468d8e0f4e366
--- /dev/null
+++ b/docs/search-index.rst
@@ -0,0 +1,11 @@
+.. toctree::
+  :maxdepth: 0
+  :hidden:
+
+Index
+-----
+
+.. automodule:: gcloud.search.index
+  :members:
+  :undoc-members:
+  :show-inheritance:
diff --git a/docs/search-usage.rst b/docs/search-usage.rst
new file mode 100644
index 0000000000000..e87025e946361
--- /dev/null
+++ b/docs/search-usage.rst
@@ -0,0 +1,314 @@
+Using the API
+=============
+
+Connection / Authorization
+--------------------------
+
+Implicitly use the default client:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> # The search module has the same methods as a client, using the default.
+  >>> search.list_indexes()  # API request
+  []
+
+Configure the default client:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> search.set_project_id('project-id')
+  >>> search.set_credentials(credentials)
+  >>> search.list_indexes()  # API request
+  []
+
+Explicitly use the default client:
+
+.. doctest::
+
+  >>> from gcloud.search import default_client as client
+  >>> # The default_client is equivalent to search.Client()
+  >>> client.list_indexes()  # API request
+  []
+
+Explicitly configure a client:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client(project_id='project-id', credentials=credentials)
+  >>> client.list_indexes()  # API request
+  []
+
+Manage indexes for a project
+----------------------------
+
+Create a new index:
+
+.. doctest::
+
+   >>> from gcloud import search
+   >>> client = search.Client()
+   >>> index = client.create_index('index_id')  # API request
+   >>> index.id
+   'index_id'
+
+Create a new index with a name:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.create_index('index_id', name='Name')  # API request
+  >>> index.name
+  'Name'
+
+Get or create an index:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_or_create_index('index_id')  # API request
+  >>> index.id
+  'index_id'
+
+List the indexes:
+
+.. doctest::
+
+   >>> from gcloud import search
+   >>> client = search.Client()
+   >>> [index.id for index in client.list_indexes()]  # API request
+   ['index_id']
+
+Retrieve an index:
+
+.. doctest::
+
+   >>> from gcloud import search
+   >>> client = search.Client()
+   >>> index = client.get_index('missing_index_id')  # API request
+   >>> index is None
+   True
+   >>> index = client.get_index('index_id')  # API request
+   >>> index.id
+   'index_id'
+
+Get an index without making an API request
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id', check=False)
+  >>> index.id
+  'index_id'
+
+Update an index:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> index.name = 'Name'
+  >>> client.update_index(index)
+
+Delete an index by ID:
+
+.. doctest::
+
+   >>> from gcloud import search
+   >>> client = search.Client()
+   >>> client.delete_index('index_id')  # API request
+
+Delete an index:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> index.id
+  'index_id'
+  >>> client.delete_index(index)  # API request
+
+Manage documents and fields
+---------------------------
+
+Create a document
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> document = search.Document('document_id', rank=0)
+  >>> document.id
+  'document_id'
+
+Add a field to a document
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> document = search.Document('document_id')
+  >>> document.add_field(search.Field('fieldname'))
+
+Add values to a field
+
+.. doctest::
+
+  >>> from datetime import datetime
+  >>> from gcloud import search
+  >>> field = search.Field('fieldname')
+  >>> field.add_value('string')
+  >>> # Tokenization field ignored for non-string values.
+  >>> field.add_value('<h1>string</h1>', tokenization='html')
+  >>> field.add_value('string', tokenization='atom')
+  >>> field.add_value('string', tokenization='text')
+  >>> field.add_value(1234)
+  >>> field.add_value(datetime.now())
+  >>> len(field.values)
+  9
+
+Add values to a field at initialization time
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> field = search.Field('fieldname', values=[
+          'string',
+          search.Value('<h1>string2</h1>', tokenization='html')
+          search.Value('string', tokenization='atom')])
+
+Add a single document to an index:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> document = search.Document('document_id', rank=0)
+  >>> index.add_document(document)  # API request
+
+Add multiple documents to an index:
+
+.. doctest::
+
+   >>> from gcloud import search
+   >>> client = search.Client()
+   >>> index = client.get_index('index_id')  # API request
+   >>> documents = [search.Document('document_id')]
+   >>> index.add_documents(documents)  # API request
+
+Get a single document by ID:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> document = index.get_document('missing_document_id')  # API request
+  >>> document is None
+  True
+  >>> document = index.get_document('document_id')  # API request
+  >>> document.fields
+  []
+
+Delete a document by ID:
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> index.delete_document('document_id')  # API request
+  >>> index.delete_document('missing_document_id')  # API request
+
+Searching
+---------
+
+Create a query
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> query = search.Query('query text here')
+  >>> query.query
+  'query text here'
+
+Specify the fields to return in a query
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> query = search.Query('query text here', fields=['field1', 'field2'])
+  >>> query.fields
+  ['field1', 'field2']
+
+Set the sort order of a query
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> query = search.Query('query text here', order_by='field1')
+  >>> query.order_by
+  'field1'
+  >>> query2 = search.Query('query text here', order_by=['field2', 'field3'])
+  >>> query2.order_by
+  ['field2', 'field3']
+  >>> # Order descending by field1 and ascending by field2
+  >>> query4 = search.Query('query text here', order_by=['-field1', 'field2'])
+  >>> query3.order_by
+  ['-field1', 'field2']
+
+Set custom field expressions on a query
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> query = search.Query('query text here')
+  >>> query.add_field_expression('total_price', '(price + tax)')
+  >>> # We don't do any checks on the expression. These are checked at query time.
+  >>> query.add_field_expression('invalid', 'is_prime(num)')
+  >>> query.add_field_expression('bad_field', '(missing_field + tax)')
+
+Set custom field expressions at initialization time
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> query = search.Query('query text here', field_expressions={
+          'total_price': '(price + tax)'})
+
+Search an index
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> matching = index.search(search.Query('query text here'))  # API request
+  >>> for document in matching:
+  ...   print document.id
+
+Search an index with a limit on number of results
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> matching = index.search(search.Query('query text here'),
+  ...                         limit=42)  # API request
+
+Search an index with a custom page size (advanced)
+
+.. doctest::
+
+  >>> from gcloud import search
+  >>> client = search.Client()
+  >>> index = client.get_index('index_id')  # API request
+  >>> matching = index.search(search.Query('query text here'),
+  ...                         page_size=20)  # API request
diff --git a/gcloud/search/__init__.py b/gcloud/search/__init__.py
new file mode 100644
index 0000000000000..4616e6075b6c2
--- /dev/null
+++ b/gcloud/search/__init__.py
@@ -0,0 +1,27 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""Search API wrapper.
+
+The main concepts with this API are:
+
+- :class:`gcloud.pubsub.topic.Topic` represents an endpoint to which messages
+  can be published using the Cloud Storage Pubsub API.
+
+- :class:`gcloud.pubsub.subscription.Subscription` represents a named
+  subscription (either pull or push) to a topic.
+"""
+
+from gcloud.search.client import Client
+from gcloud.search.connection import SCOPE
diff --git a/gcloud/search/client.py b/gcloud/search/client.py
new file mode 100644
index 0000000000000..bd0321ce0a1cf
--- /dev/null
+++ b/gcloud/search/client.py
@@ -0,0 +1,95 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""gcloud pubsub client for interacting with API."""
+
+
+from gcloud.client import JSONClient
+from gcloud.exceptions import NotFound
+from gcloud.iterator import Iterator
+from gcloud.search.connection import Connection
+from gcloud.search.index import Index
+
+
+class Client(JSONClient):
+    """Client to bundle configuration needed for API requests."""
+
+    _connection_class = Connection
+
+    def list_indexes(self, page_size=None):
+        """List topics for the project associated with this client.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/topics/list
+
+        :type page_size: int
+        :param page_size: maximum number of topics to return, If not passed,
+                          defaults to a value set by the API.
+
+        :rtype: :class:`gcloud.iterator.Iterator`
+        :returns: an :class:`gcloud.iterator.Iterator` of
+                  :class:`gcloud.search.index.Index`
+        """
+        params = {}
+
+        if page_size is not None:
+            params['pageSize'] = page_size
+
+        path = '/projects/%s/indexes' % (self.project,)
+
+        client = self
+
+        class IndexIterator(Iterator):
+            """An iterator over a list of Index resources."""
+
+            def get_items_from_response(self, response):
+                """Get :class:`gcloud.search.index.Index` items from response.
+
+                :type response: dict
+                :param response: The JSON API response for a page of indexes.
+                """
+                for resource in response.get('projects', []):
+                    item = Index.from_api_repr(resource, client=client)
+                    yield item
+
+        return IndexIterator(connection=self.connection, extra_params=params,
+                             path=path)
+
+    def index(self, index_id):
+        """Creates an index bound to the current client.
+
+        :type index_id: string
+        :param index_id: the ID of the index to be constructed.
+
+        :rtype: :class:`gcloud.search.index.Index`
+        :returns: the index created with the current client.
+        """
+        return Index(index_id=index_id, client=self)
+
+    def get_index(self, index_id):
+        """Retrieve an index from the Cloud Search API.
+
+        :type index_id: string
+        :param index_id: the ID of the index to be retrieved.
+
+        :rtype: :class:`gcloud.search.index.Index`
+        :returns: the index retrieved via the current client or ``None`` if
+                  the index with that ID doesn't exist.
+        """
+        try:
+            index = self.index(index_id)
+            index.reload()
+        except NotFound:
+            index = None
+        return index
diff --git a/gcloud/search/connection.py b/gcloud/search/connection.py
new file mode 100644
index 0000000000000..84571c433f48d
--- /dev/null
+++ b/gcloud/search/connection.py
@@ -0,0 +1,39 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""Create / interact with gcloud pubsub connections."""
+
+from gcloud import connection as base_connection
+
+
+SCOPE = ('https://www.googleapis.com/auth/cloudsearch',
+         'https://www.googleapis.com/auth/cloud-platform')
+"""The scopes required for authenticating as a Cloud Pub/Sub consumer."""
+
+
+class Connection(base_connection.JSONConnection):
+    """A connection to Google Cloud Pubsub via the JSON REST API."""
+
+    API_BASE_URL = 'https://cloudsearch.googleapis.com'
+    """The base of the API call URL."""
+
+    API_VERSION = 'v1'
+    """The version of the API, used in building the API call's URL."""
+
+    API_URL_TEMPLATE = '{api_base_url}/{api_version}{path}'
+    """A template for the URL of a particular API call."""
+
+    def __init__(self, credentials=None, http=None):
+        credentials = self._create_scoped_credentials(credentials, SCOPE)
+        super(Connection, self).__init__(credentials=credentials, http=http)
diff --git a/gcloud/search/document.py b/gcloud/search/document.py
new file mode 100644
index 0000000000000..8d8e7789da38d
--- /dev/null
+++ b/gcloud/search/document.py
@@ -0,0 +1,131 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""Define Search Document."""
+
+
+class Document(object):
+    """Subscriptions receive messages published to their topics.
+
+    See:
+    https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions
+
+    :type document_id: string
+    :param document_id: the name of the subscription
+
+    :type fields: :class:`gcloud.pubsub.topic.Topic`
+    :param fields: the topic to which the subscription belongs..
+
+    :type client: int
+    :param client: the deadline (in seconds) by which messages pulled
+                   from the back-end must be acknowledged.
+
+    :type index: string
+    :param index: URL to which messages will be pushed by the back-end.
+                  If not set, the application must pull messages.
+    """
+    def __init__(self, document_id, fields=None, client=None, index=None):
+        self.document_id = document_id
+        self.fields = fields or []
+        self._client = client
+        self.index = index
+
+    @classmethod
+    def from_api_repr(cls, resource, client=None):
+        """Factory:  construct a topic given its API representation
+
+        :type resource: dict
+        :param resource: topic resource representation returned from the API
+
+        :type client: :class:`gcloud.pubsub.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for a topic.
+
+        :rtype: :class:`gcloud.pubsub.subscription.Subscription`
+        :returns: Subscription parsed from ``resource``.
+        """
+        return cls(document_id=resource['id'], client=client)
+
+    @property
+    def client(self):
+        """Retrieve the client for this document.
+
+        This first checks the client set directly, then checks the client set
+        on the index. If no client is found, returns ``None``.
+
+        :rtype: :class:`gcloud.search.client.Client` or ``None``
+        :returns: The client associated with this document.
+        """
+        if self._client:
+            return self._client
+        elif self._index:
+            return self._index.client
+
+    @property
+    def path(self):
+        """URL path for the subscription's APIs"""
+        project = self.topic.project
+        return '/projects/%s/subscriptions/%s' % (project, self.name)
+
+    def _require_client(self, client=None):
+        """Get either a client or raise an exception.
+
+        We need to use this as the various methods could accept a client as a
+        parameter, which we need to evaluate. If the client provided is empty
+        and there is no client set as an instance variable, we'll raise a
+        :class:`ValueError`.
+
+        :type client: :class:`gcloud.search.client.Client`
+        :param client: An optional client to test for existence.
+        """
+        client = client or self.client
+        if not client:
+            raise ValueError('Missing client. You can set the client '
+                             'directly on the document, or indirectly on the '
+                             'index.')
+        return client
+
+    def add_field(self, field):
+        """Add a field to this document.
+
+        :type field: :class:`gcloud.search.field.Field`
+        :param field: The field to add to the document.
+        """
+        pass
+
+    def reload(self, client=None):
+        """API call:  sync local subscription configuration via a GET request
+
+        See
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/get
+
+        :type client: :class:`gcloud.search.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current document's index.
+        """
+        # client = self._require_client(client)
+        # data = client.connection.api_request(method='GET', path=self.path)
+
+    def delete(self, client=None):
+        """API call:  delete the subscription via a DELETE request.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/delete
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        client = self._require_client(client)
+        client.connection.api_request(method='DELETE', path=self.path)
diff --git a/gcloud/search/field.py b/gcloud/search/field.py
new file mode 100644
index 0000000000000..b50a049b3a937
--- /dev/null
+++ b/gcloud/search/field.py
@@ -0,0 +1,59 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""Define Search Field."""
+
+
+from gcloud.search.value import Value
+
+
+class Field(object):
+    """Subscriptions receive messages published to their topics.
+
+    See:
+    https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions
+
+    :type name: string
+    :param name: the name of the field
+
+    :type values: iterable of string or ``None``
+    :param values: the list of values to be associated with this field.
+    """
+    def __init__(self, name, values=None):
+        self.name = name
+        self.values = values or []
+
+    @classmethod
+    def from_api_repr(cls, resource):
+        """Factory:  construct a field given its API representation
+
+        :type resource: dict
+        :param resource: topic resource representation returned from the API
+
+        :rtype: :class:`gcloud.search.field.Field`
+        :returns: Field parsed from ``resource``.
+        """
+        return cls(name=resource['name'])
+
+    def add_value(self, value, tokenization=None):
+        """Add a value to this field.
+
+        :type value: string
+        :param value: The value to add to the field.
+
+        :type tokenization: string
+        :param tokenization: The tokenization type of the value.
+        """
+        self.values.append(Value(value=value,
+                                 tokenization=tokenization))
diff --git a/gcloud/search/index.py b/gcloud/search/index.py
new file mode 100644
index 0000000000000..56bc470f05058
--- /dev/null
+++ b/gcloud/search/index.py
@@ -0,0 +1,335 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""Define Projects."""
+
+from gcloud.exceptions import NotFound
+
+
+class Index(object):
+    """Projects are containers for your work on Google Cloud Platform.
+
+    .. note::
+
+        It's unlikely that you'd need to instantiate this outside the context
+        of a :class:`.client.Client`, so in general, it's best to get a Project
+        from a Resource Manager client.
+
+    To create a new project::
+
+        >>> from gcloud import resource_manager
+        >>> client = resource_manager.Client()
+        >>> project = client.project('purple-spaceship-123')
+        >>> project.name = 'Purple Spaceship Project!'
+        >>> project.create()
+
+    To get an existing project::
+
+        >>> from gcloud import resource_manager
+        >>> client = resource_manager.Client()
+        >>> project = client.get_project('purple-spaceship-123')
+        >>> print project.name
+        Purple Spaceship Project!
+
+    See:
+    https://cloud.google.com/resource-manager/reference/rest/v1beta1/projects
+
+    :type index_id: string
+    :param index_id: the unique ID of the index.
+
+    :type client: :class:`gcloud.search.client.Client`
+    :param client: Client which holds credentials and project
+                   configuration for the index.
+    """
+    def __init__(self, client, index_id):
+        self.client = client
+        self.index_id = index_id
+
+    def __repr__(self):
+        return '<Index: %s>' % (self.index_id,)
+
+    @classmethod
+    def from_api_repr(cls, resource, client):
+        """Factory:  construct a project given its API representation.
+
+        :type resource: dict
+        :param resource: project resource representation returned from the API
+
+        :type client: :class:`gcloud.pubsub.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for the index.
+
+        :rtype: :class:`gcloud.search.index.Index`
+        """
+        project = cls(id=resource['projectId'], client=client)
+        project.set_properties_from_api_repr(resource)
+        return project
+
+    def set_properties_from_api_repr(self, resource):
+        """Update specific properties from its API representation."""
+        self.name = resource.get('name')
+        self.number = resource['projectNumber']
+        self.labels = resource.get('labels', {})
+        self.status = resource['lifecycleState']
+
+    @property
+    def project(self):
+        """Project bound to the index."""
+        return self.client.project
+
+    @property
+    def full_name(self):
+        """Fully-qualified name (ie, ``'projects/purple-spaceship-123'``)."""
+        if not self.project:
+            raise ValueError('Missing project ID!')
+        return 'projects/%s/indexes/%s' % (self.project, self.index_id)
+
+    @property
+    def path(self):
+        """URL for the project (ie, ``'/projects/purple-spaceship-123'``)."""
+        return '/%s' % (self.full_name)
+
+    def _require_client(self, client=None):
+        """Get either a client or raise an exception.
+
+        We need to use this as the various methods could accept a client as a
+        parameter, which we need to evaluate. If the client provided is empty
+        and there is no client set as an instance variable, we'll raise a
+        :class:`gcloud.exceptions.MissingClientError`.
+
+        :type client: :class:`gcloud.resource_manager.client.Client`
+        :param client: An optional client to test for existence.
+        """
+        client = client or self.client
+        if not client:
+            raise ValueError('Missing client.')
+        return client
+
+    def create(self, client=None):
+        """API call:  create the project via a ``POST`` request.
+
+        Example::
+
+            >>> from gcloud import resource_manager
+            >>> client = resource_manager.Client()
+            >>> project = client.project('new-spaceship-123')
+            >>> project.name = 'New Spaceship Project!'
+            >>> project.create()
+
+        See
+        https://cloud.google.com/resource-manager/reference/rest/v1beta1/projects/create
+
+        :type client: :class:`gcloud.resource_manager.client.Client` or None
+        :param client: the client to use.  If not passed, falls back to
+                       the ``client`` attribute.
+        """
+        # Right now this is a no-op as indexes are implicitly created.
+        # Later, this will be a PUT request to the API.
+        # client = self._require_client(client=client)
+        # resp = client.connection.api_request(method='POST', path=self.path)
+        # self.set_properties_from_api_repr(resource=resp)
+
+    def reload(self, client=None):
+        """API call:  reload the index via a ``GET`` request.
+
+        This method will reload the newest metadata for the project.
+
+        .. warning::
+
+            This will overwrite any local changes you've made and not saved!
+
+        Example::
+
+            >>> from gcloud import resource_manager
+            >>> client = resource_manager.Client()
+            >>> project = client.get_project('purple-spaceship-123')
+            >>> project.name = 'Locally changed name'
+            >>> print project
+            <Project: Locally changed name (purple-spaceship-123)>
+            >>> project.reload()
+            >>> print project
+            <project: Purple Spaceship (purple-spaceship-123)>
+
+        See
+        https://cloud.google.com/resource-manager/reference/rest/v1beta1/projects/get
+
+        :type client: :class:`gcloud.resource_manager.client.Client` or None
+        :param client: the client to use.  If not passed, falls back to
+                       the ``client`` attribute.
+        """
+        # client = self._require_client(client=client)
+
+        # We assume the project exists. If it doesn't it will raise a NotFound
+        # exception.
+        # Right now this is a no-op, as there's no extra data with an index.
+        # resp = client.connection.api_request(method='GET', path=self.path)
+        # self.set_properties_from_api_repr(resource=resp)
+
+    def update(self, client=None):
+        """API call:  update the project via a ``PUT`` request.
+
+        Example::
+
+            >>> from gcloud import resource_manager
+            >>> client = resource_manager.Client()
+            >>> project = client.get_project('purple-spaceship-123')
+            >>> project.name = 'New Purple Spaceship'
+            >>> project.labels['environment'] = 'prod'
+            >>> project.update()
+
+        See
+        https://cloud.google.com/resource-manager/reference/rest/v1beta1/projects/update
+
+        :type client: :class:`gcloud.resource_manager.client.Client` or None
+        :param client: the client to use.  If not passed, falls back to
+                       the ``client`` attribute.
+        """
+        client = self._require_client(client=client)
+
+        # Right now this is a no-op, as indexes have no extra data.
+        # data = {'name': self.name, 'labels': self.labels}
+        # resp = client.connection.api_request(method='PUT', path=self.path,
+        #                                      data=data)
+        # self.set_properties_from_api_repr(resp)
+
+    def exists(self, client=None):
+        """API call:  test the existence of a project via a ``GET`` request.
+
+        Example::
+
+            >>> from gcloud import resource_manager
+            >>> client = resource_manager.Client()
+            >>> project = client.project('purple-spaceship-456')
+            >>> project.exists()
+            False
+
+        You can also use the
+        :func:`gcloud.resource_manager.client.Client.get_project`
+        method to check whether a project exists, as it will return ``None``
+        if the project doesn't exist::
+
+            >>> from gcloud import resource_manager
+            >>> client = resource_manager.Client()
+            >>> print client.get_project('purple-spaceship-456')
+            None
+
+        See
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/projects/get
+
+        :type client: :class:`gcloud.resource_manager.client.Client` or None
+        :param client: the client to use.  If not passed, falls back to
+                       the ``client`` attribute.
+        """
+        # Currently there is no way to do this, so this is a no-op.
+        # client = self._require_client(client=client)
+        #
+        # try:
+        #     client.connection.api_request(method='GET', path=self.path)
+        # except NotFound:
+        #     return False
+        # else:
+        #     return True
+
+    def delete(self, client=None, reload=True):
+        """API call:  delete the project via a ``DELETE`` request.
+
+        See:
+        https://cloud.google.com/resource-manager/reference/rest/v1beta1/projects/delete
+
+        This actually changes the status (``lifecycleState``) from ``ACTIVE``
+        to ``DELETE_REQUESTED``.
+        Later (it's not specified when), the project will move into the
+        ``DELETE_IN_PROGRESS`` state, which means the deleting has actually
+        begun.
+
+        Example::
+
+            >>> from gcloud import resource_manager
+            >>> client = resource_manager.Client()
+            >>> project = client.get_project('purple-spaceship-123')
+            >>> project.delete()
+
+        :type client: :class:`gcloud.resource_manager.client.Client` or None
+        :param client: the client to use.  If not passed,
+                       falls back to the ``client`` attribute.
+
+        :type reload: bool
+        :param reload: Whether to reload the project with the latest state.
+                       Default: ``True``.
+        """
+        # This currently is not possible.
+        # client = self._require_client(client)
+        # client.connection.api_request(method='DELETE', path=self.path)
+        #
+        # If the reload flag is True, reload the project.
+        # if reload:
+        #     self.reload()
+
+    def document(self, document_id):
+        """Get a document instance.
+
+        :type document_id: string
+        :param document_id: The unique ID for the document.
+
+        :rtype: :class:`gcloud.search.document.Document`
+        """
+        pass
+
+    def get_document(self, document_id, client=None):
+        """Retrieve a document from the Cloud Search API.
+
+        :type document_id: string
+        :param document_id: The unique ID for the document.
+
+        :type client: :class:`gcloud.search.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for the index.
+
+        :rtype: :class:`gcloud.search.document.Document`
+        """
+        client = self._require_client(client=client)
+        document = self.document(document_id)
+        try:
+            document.reload()
+        except NotFound:
+            document = None
+        return document
+
+    def add_document(self, document, client=None):
+        """Add a document to this index.
+
+        :type document: :class:`gcloud.search.document.Document`
+        :param document: The document to add to the index.
+
+        :type client: :class:`gcloud.search.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for the index.
+        """
+        self.add_document([document], client=client)
+
+    def add_documents(self, documents, client=None):
+        """Add a list of documents to this index.
+
+        :type documents: iterable of :class:`gcloud.search.document.Document`
+        :param documents: The documents to add to the index.
+
+        :type client: :class:`gcloud.search.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for the index.
+        """
+        pass
+
+    def query(self):
+        """Execute a query over this index."""
+        pass
diff --git a/gcloud/search/subscription.py b/gcloud/search/subscription.py
new file mode 100644
index 0000000000000..cb3023a286d77
--- /dev/null
+++ b/gcloud/search/subscription.py
@@ -0,0 +1,263 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""Define API Subscriptions."""
+
+from gcloud.exceptions import NotFound
+from gcloud.pubsub.message import Message
+from gcloud.pubsub.topic import Topic
+
+
+class Subscription(object):
+    """Subscriptions receive messages published to their topics.
+
+    See:
+    https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions
+
+    :type name: string
+    :param name: the name of the subscription
+
+    :type topic: :class:`gcloud.pubsub.topic.Topic`
+    :param topic: the topic to which the subscription belongs..
+
+    :type ack_deadline: int
+    :param ack_deadline: the deadline (in seconds) by which messages pulled
+                         from the back-end must be acknowledged.
+
+    :type push_endpoint: string
+    :param push_endpoint: URL to which messages will be pushed by the back-end.
+                          If not set, the application must pull messages.
+    """
+    def __init__(self, name, topic, ack_deadline=None, push_endpoint=None):
+        self.name = name
+        self.topic = topic
+        self.ack_deadline = ack_deadline
+        self.push_endpoint = push_endpoint
+
+    @classmethod
+    def from_api_repr(cls, resource, client, topics=None):
+        """Factory:  construct a topic given its API representation
+
+        :type resource: dict
+        :param resource: topic resource representation returned from the API
+
+        :type client: :class:`gcloud.pubsub.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for a topic.
+
+        :type topics: dict or None
+        :param topics: A mapping of topic names -> topics.  If not passed,
+                       the subscription will have a newly-created topic.
+
+        :rtype: :class:`gcloud.pubsub.subscription.Subscription`
+        :returns: Subscription parsed from ``resource``.
+        """
+        if topics is None:
+            topics = {}
+        t_name = resource['topic']
+        topic = topics.get(t_name)
+        if topic is None:
+            topic = topics[t_name] = Topic.from_api_repr({'name': t_name},
+                                                         client)
+        _, _, _, name = resource['name'].split('/')
+        ack_deadline = resource.get('ackDeadlineSeconds')
+        push_config = resource.get('pushConfig', {})
+        push_endpoint = push_config.get('pushEndpoint')
+        return cls(name, topic, ack_deadline, push_endpoint)
+
+    @property
+    def path(self):
+        """URL path for the subscription's APIs"""
+        project = self.topic.project
+        return '/projects/%s/subscriptions/%s' % (project, self.name)
+
+    def _require_client(self, client):
+        """Check client or verify over-ride.
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the topic of the
+                       current subscription.
+
+        :rtype: :class:`gcloud.pubsub.client.Client`
+        :returns: The client passed in or the currently bound client.
+        """
+        if client is None:
+            client = self.topic._client
+        return client
+
+    def create(self, client=None):
+        """API call:  create the subscription via a PUT request
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/create
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        data = {'topic': self.topic.full_name}
+
+        if self.ack_deadline is not None:
+            data['ackDeadline'] = self.ack_deadline
+
+        if self.push_endpoint is not None:
+            data['pushConfig'] = {'pushEndpoint': self.push_endpoint}
+
+        client = self._require_client(client)
+        client.connection.api_request(method='PUT', path=self.path, data=data)
+
+    def exists(self, client=None):
+        """API call:  test existence of the subscription via a GET request
+
+        See
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/get
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        client = self._require_client(client)
+        try:
+            client.connection.api_request(method='GET', path=self.path)
+        except NotFound:
+            return False
+        else:
+            return True
+
+    def reload(self, client=None):
+        """API call:  sync local subscription configuration via a GET request
+
+        See
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/get
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        client = self._require_client(client)
+        data = client.connection.api_request(method='GET', path=self.path)
+        self.ack_deadline = data.get('ackDeadline')
+        push_config = data.get('pushConfig', {})
+        self.push_endpoint = push_config.get('pushEndpoint')
+
+    def modify_push_configuration(self, push_endpoint, client=None):
+        """API call:  update the push endpoint for the subscription.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/modifyPushConfig
+
+        :type push_endpoint: string
+        :param push_endpoint: URL to which messages will be pushed by the
+                              back-end.  If None, the application must pull
+                              messages.
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        client = self._require_client(client)
+        data = {}
+        config = data['pushConfig'] = {}
+        if push_endpoint is not None:
+            config['pushEndpoint'] = push_endpoint
+        client.connection.api_request(
+            method='POST', path='%s:modifyPushConfig' % (self.path,),
+            data=data)
+        self.push_endpoint = push_endpoint
+
+    def pull(self, return_immediately=False, max_messages=1, client=None):
+        """API call:  retrieve messages for the subscription.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/pull
+
+        :type return_immediately: boolean
+        :param return_immediately: if True, the back-end returns even if no
+                                   messages are available;  if False, the API
+                                   call blocks until one or more messages are
+                                   available.
+
+        :type max_messages: int
+        :param max_messages: the maximum number of messages to return.
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+
+        :rtype: list of (ack_id, message) tuples
+        :returns: sequence of tuples: ``ack_id`` is the ID to be used in a
+                  subsequent call to :meth:`acknowledge`, and ``message``
+                  is an instance of :class:`gcloud.pubsub.message.Message`.
+        """
+        client = self._require_client(client)
+        data = {'returnImmediately': return_immediately,
+                'maxMessages': max_messages}
+        response = client.connection.api_request(
+            method='POST', path='%s:pull' % (self.path,), data=data)
+        return [(info['ackId'], Message.from_api_repr(info['message']))
+                for info in response.get('receivedMessages', ())]
+
+    def acknowledge(self, ack_ids, client=None):
+        """API call:  acknowledge retrieved messages for the subscription.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/acknowledge
+
+        :type ack_ids: list of string
+        :param ack_ids: ack IDs of messages being acknowledged
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        client = self._require_client(client)
+        data = {'ackIds': ack_ids}
+        client.connection.api_request(
+            method='POST', path='%s:acknowledge' % (self.path,), data=data)
+
+    def modify_ack_deadline(self, ack_id, ack_deadline, client=None):
+        """API call:  update acknowledgement deadline for a retrieved message.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/acknowledge
+
+        :type ack_id: string
+        :param ack_id: ack ID of message being updated
+
+        :type ack_deadline: int
+        :param ack_deadline: new deadline for the message, in seconds
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        client = self._require_client(client)
+        data = {'ackId': ack_id, 'ackDeadlineSeconds': ack_deadline}
+        client.connection.api_request(
+            method='POST', path='%s:modifyAckDeadline' % (self.path,),
+            data=data)
+
+    def delete(self, client=None):
+        """API call:  delete the subscription via a DELETE request.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions/delete
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+        """
+        client = self._require_client(client)
+        client.connection.api_request(method='DELETE', path=self.path)
diff --git a/gcloud/search/test_client.py b/gcloud/search/test_client.py
new file mode 100644
index 0000000000000..f0f1d30a81bed
--- /dev/null
+++ b/gcloud/search/test_client.py
@@ -0,0 +1,342 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+import unittest2
+
+
+class TestClient(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.client import Client
+        return Client
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_defaults(self):
+        from gcloud._testing import _Monkey
+        from gcloud.pubsub import SCOPE
+        from gcloud.pubsub import client
+        from gcloud.pubsub.connection import Connection
+
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+        FUNC_CALLS = []
+
+        def mock_get_proj():
+            FUNC_CALLS.append('_get_production_project')
+            return PROJECT
+
+        def mock_get_credentials():
+            FUNC_CALLS.append('get_credentials')
+            return CREDS
+
+        with _Monkey(client, get_credentials=mock_get_credentials,
+                     _get_production_project=mock_get_proj):
+            client_obj = self._makeOne()
+
+        self.assertEqual(client_obj.project, PROJECT)
+        self.assertTrue(isinstance(client_obj.connection, Connection))
+        self.assertTrue(client_obj.connection._credentials is CREDS)
+        self.assertEqual(client_obj.connection._credentials._scopes, SCOPE)
+        self.assertEqual(FUNC_CALLS,
+                         ['_get_production_project', 'get_credentials'])
+
+    def test_ctor_missing_project(self):
+        from gcloud._testing import _Monkey
+        from gcloud.pubsub import client
+
+        FUNC_CALLS = []
+
+        def mock_get_proj():
+            FUNC_CALLS.append('_get_production_project')
+            return None
+
+        with _Monkey(client, _get_production_project=mock_get_proj):
+            self.assertRaises(ValueError, self._makeOne)
+
+        self.assertEqual(FUNC_CALLS, ['_get_production_project'])
+
+    def test_ctor_explicit(self):
+        from gcloud.pubsub import SCOPE
+        from gcloud.pubsub.connection import Connection
+
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        client_obj = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        self.assertEqual(client_obj.project, PROJECT)
+        self.assertTrue(isinstance(client_obj.connection, Connection))
+        self.assertTrue(client_obj.connection._credentials is CREDS)
+        self.assertEqual(CREDS._scopes, SCOPE)
+
+    def test_from_service_account_json(self):
+        from gcloud._testing import _Monkey
+        from gcloud.pubsub import client
+        from gcloud.pubsub.connection import Connection
+
+        PROJECT = 'PROJECT'
+        KLASS = self._getTargetClass()
+        CREDS = _Credentials()
+        _CALLED = []
+
+        def mock_creds(arg1):
+            _CALLED.append((arg1,))
+            return CREDS
+
+        BOGUS_ARG = object()
+        with _Monkey(client, get_for_service_account_json=mock_creds):
+            client_obj = KLASS.from_service_account_json(
+                BOGUS_ARG, project=PROJECT)
+
+        self.assertEqual(client_obj.project, PROJECT)
+        self.assertTrue(isinstance(client_obj.connection, Connection))
+        self.assertTrue(client_obj.connection._credentials is CREDS)
+        self.assertEqual(_CALLED, [(BOGUS_ARG,)])
+
+    def test_from_service_account_p12(self):
+        from gcloud._testing import _Monkey
+        from gcloud.pubsub import client
+        from gcloud.pubsub.connection import Connection
+
+        PROJECT = 'PROJECT'
+        KLASS = self._getTargetClass()
+        CREDS = _Credentials()
+        _CALLED = []
+
+        def mock_creds(arg1, arg2):
+            _CALLED.append((arg1, arg2))
+            return CREDS
+
+        BOGUS_ARG1 = object()
+        BOGUS_ARG2 = object()
+        with _Monkey(client, get_for_service_account_p12=mock_creds):
+            client_obj = KLASS.from_service_account_p12(
+                BOGUS_ARG1, BOGUS_ARG2, project=PROJECT)
+
+        self.assertEqual(client_obj.project, PROJECT)
+        self.assertTrue(isinstance(client_obj.connection, Connection))
+        self.assertTrue(client_obj.connection._credentials is CREDS)
+        self.assertEqual(_CALLED, [(BOGUS_ARG1, BOGUS_ARG2)])
+
+    def test_list_topics_no_paging(self):
+        from gcloud.pubsub.topic import Topic
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+
+        RETURNED = {'topics': [{'name': TOPIC_PATH}]}
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        topics, next_page_token = CLIENT_OBJ.list_topics()
+        # Test values are correct.
+        self.assertEqual(len(topics), 1)
+        self.assertTrue(isinstance(topics[0], Topic))
+        self.assertEqual(topics[0].name, TOPIC_NAME)
+        self.assertEqual(next_page_token, None)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/topics' % PROJECT)
+        self.assertEqual(req['query_params'], {})
+
+    def test_list_topics_with_paging(self):
+        from gcloud.pubsub.topic import Topic
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        TOKEN1 = 'TOKEN1'
+        TOKEN2 = 'TOKEN2'
+        SIZE = 1
+        RETURNED = {'topics': [{'name': TOPIC_PATH}],
+                    'nextPageToken': TOKEN2}
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        topics, next_page_token = CLIENT_OBJ.list_topics(SIZE, TOKEN1)
+        # Test values are correct.
+        self.assertEqual(len(topics), 1)
+        self.assertTrue(isinstance(topics[0], Topic))
+        self.assertEqual(topics[0].name, TOPIC_NAME)
+        self.assertEqual(next_page_token, TOKEN2)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/topics' % PROJECT)
+        self.assertEqual(req['query_params'],
+                         {'pageSize': SIZE, 'pageToken': TOKEN1})
+
+    def test_list_subscriptions_no_paging(self):
+        from gcloud.pubsub.subscription import Subscription
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        SUB_NAME = 'subscription_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        SUB_INFO = [{'name': SUB_PATH, 'topic': TOPIC_PATH}]
+        RETURNED = {'subscriptions': SUB_INFO}
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        subscriptions, next_page_token = CLIENT_OBJ.list_subscriptions()
+        # Test values are correct.
+        self.assertEqual(len(subscriptions), 1)
+        self.assertTrue(isinstance(subscriptions[0], Subscription))
+        self.assertEqual(subscriptions[0].name, SUB_NAME)
+        self.assertEqual(subscriptions[0].topic.name, TOPIC_NAME)
+        self.assertEqual(next_page_token, None)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/subscriptions' % PROJECT)
+        self.assertEqual(req['query_params'], {})
+
+    def test_list_subscriptions_with_paging(self):
+        from gcloud.pubsub.subscription import Subscription
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        SUB_NAME = 'subscription_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        ACK_DEADLINE = 42
+        PUSH_ENDPOINT = 'https://push.example.com/endpoint'
+        TOKEN1 = 'TOKEN1'
+        TOKEN2 = 'TOKEN2'
+        SIZE = 1
+        SUB_INFO = [{'name': SUB_PATH,
+                     'topic': TOPIC_PATH,
+                     'ackDeadlineSeconds': ACK_DEADLINE,
+                     'pushConfig': {'pushEndpoint': PUSH_ENDPOINT}}]
+        RETURNED = {'subscriptions': SUB_INFO, 'nextPageToken': TOKEN2}
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        subscriptions, next_page_token = CLIENT_OBJ.list_subscriptions(
+            SIZE, TOKEN1)
+        # Test values are correct.
+        self.assertEqual(len(subscriptions), 1)
+        self.assertTrue(isinstance(subscriptions[0], Subscription))
+        self.assertEqual(subscriptions[0].name, SUB_NAME)
+        self.assertEqual(subscriptions[0].topic.name, TOPIC_NAME)
+        self.assertEqual(subscriptions[0].ack_deadline, ACK_DEADLINE)
+        self.assertEqual(subscriptions[0].push_endpoint, PUSH_ENDPOINT)
+        self.assertEqual(next_page_token, TOKEN2)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/subscriptions' % PROJECT)
+        self.assertEqual(req['query_params'],
+                         {'pageSize': SIZE, 'pageToken': TOKEN1})
+
+    def test_list_subscriptions_with_topic_name(self):
+        from gcloud.pubsub.subscription import Subscription
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        SUB_NAME_1 = 'subscription_1'
+        SUB_PATH_1 = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME_1)
+        SUB_NAME_2 = 'subscription_2'
+        SUB_PATH_2 = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME_2)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        SUB_INFO = [{'name': SUB_PATH_1, 'topic': TOPIC_PATH},
+                    {'name': SUB_PATH_2, 'topic': TOPIC_PATH}]
+        TOKEN = 'TOKEN'
+        RETURNED = {'subscriptions': SUB_INFO, 'nextPageToken': TOKEN}
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        subscriptions, next_page_token = CLIENT_OBJ.list_subscriptions(
+            topic_name=TOPIC_NAME)
+        # Test values are correct.
+        self.assertEqual(len(subscriptions), 2)
+        self.assertTrue(isinstance(subscriptions[0], Subscription))
+        self.assertEqual(subscriptions[0].name, SUB_NAME_1)
+        self.assertEqual(subscriptions[0].topic.name, TOPIC_NAME)
+        self.assertTrue(isinstance(subscriptions[1], Subscription))
+        self.assertEqual(subscriptions[1].name, SUB_NAME_2)
+        self.assertEqual(subscriptions[1].topic.name, TOPIC_NAME)
+        self.assertTrue(subscriptions[1].topic is subscriptions[0].topic)
+        self.assertEqual(next_page_token, TOKEN)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'],
+                         '/projects/%s/topics/%s/subscriptions'
+                         % (PROJECT, TOPIC_NAME))
+        self.assertEqual(req['query_params'], {})
+
+    def test_topic(self):
+        PROJECT = 'PROJECT'
+        TOPIC_NAME = 'TOPIC_NAME'
+        CREDS = _Credentials()
+
+        client_obj = self._makeOne(project=PROJECT, credentials=CREDS)
+        new_topic = client_obj.topic(TOPIC_NAME)
+        self.assertEqual(new_topic.name, TOPIC_NAME)
+        self.assertTrue(new_topic._client is client_obj)
+        self.assertEqual(new_topic.project, PROJECT)
+        self.assertEqual(new_topic.full_name,
+                         'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME))
+        self.assertFalse(new_topic.timestamp_messages)
+
+
+class _Credentials(object):
+
+    _scopes = None
+
+    @staticmethod
+    def create_scoped_required():
+        return True
+
+    def create_scoped(self, scope):
+        self._scopes = scope
+        return self
+
+
+class _Connection(object):
+
+    def __init__(self, *responses):
+        self._responses = responses
+        self._requested = []
+
+    def api_request(self, **kw):
+        self._requested.append(kw)
+        response, self._responses = self._responses[0], self._responses[1:]
+        return response
diff --git a/gcloud/search/test_connection.py b/gcloud/search/test_connection.py
new file mode 100644
index 0000000000000..4a8618388e4e5
--- /dev/null
+++ b/gcloud/search/test_connection.py
@@ -0,0 +1,46 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+import unittest2
+
+
+class TestConnection(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.connection import Connection
+        return Connection
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_build_api_url_no_extra_query_params(self):
+        conn = self._makeOne()
+        URI = '/'.join([
+            conn.API_BASE_URL,
+            conn.API_VERSION,
+            'foo',
+        ])
+        self.assertEqual(conn.build_api_url('/foo'), URI)
+
+    def test_build_api_url_w_extra_query_params(self):
+        from six.moves.urllib.parse import parse_qsl
+        from six.moves.urllib.parse import urlsplit
+        conn = self._makeOne()
+        uri = conn.build_api_url('/foo', {'bar': 'baz'})
+        scheme, netloc, path, qs, _ = urlsplit(uri)
+        self.assertEqual('%s://%s' % (scheme, netloc), conn.API_BASE_URL)
+        self.assertEqual(path,
+                         '/'.join(['', conn.API_VERSION, 'foo']))
+        parms = dict(parse_qsl(qs))
+        self.assertEqual(parms['bar'], 'baz')
diff --git a/gcloud/search/test_message.py b/gcloud/search/test_message.py
new file mode 100644
index 0000000000000..38ad240e6199d
--- /dev/null
+++ b/gcloud/search/test_message.py
@@ -0,0 +1,104 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+import unittest2
+
+
+class TestMessage(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.message import Message
+        return Message
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_no_attributes(self):
+        DATA = b'DEADBEEF'
+        MESSAGE_ID = b'12345'
+        message = self._makeOne(data=DATA, message_id=MESSAGE_ID)
+        self.assertEqual(message.data, DATA)
+        self.assertEqual(message.message_id, MESSAGE_ID)
+        self.assertEqual(message.attributes, {})
+
+    def test_ctor_w_attributes(self):
+        DATA = b'DEADBEEF'
+        MESSAGE_ID = b'12345'
+        ATTRS = {'a': 'b'}
+        message = self._makeOne(data=DATA, message_id=MESSAGE_ID,
+                                attributes=ATTRS)
+        self.assertEqual(message.data, DATA)
+        self.assertEqual(message.message_id, MESSAGE_ID)
+        self.assertEqual(message.attributes, ATTRS)
+
+    def test_from_api_repr_no_attributes(self):
+        from base64 import b64encode as b64
+        DATA = b'DEADBEEF'
+        B64_DATA = b64(DATA)
+        MESSAGE_ID = '12345'
+        api_repr = {'data': B64_DATA, 'messageId': MESSAGE_ID}
+        message = self._getTargetClass().from_api_repr(api_repr)
+        self.assertEqual(message.data, DATA)
+        self.assertEqual(message.message_id, MESSAGE_ID)
+        self.assertEqual(message.attributes, {})
+
+    def test_from_api_repr_w_attributes(self):
+        from base64 import b64encode as b64
+        DATA = b'DEADBEEF'
+        B64_DATA = b64(DATA)
+        MESSAGE_ID = '12345'
+        ATTRS = {'a': 'b'}
+        api_repr = {'data': B64_DATA,
+                    'messageId': MESSAGE_ID,
+                    'attributes': ATTRS}
+        message = self._getTargetClass().from_api_repr(api_repr)
+        self.assertEqual(message.data, DATA)
+        self.assertEqual(message.message_id, MESSAGE_ID)
+        self.assertEqual(message.attributes, ATTRS)
+
+    def test_timestamp_no_attributes(self):
+        DATA = b'DEADBEEF'
+        MESSAGE_ID = b'12345'
+        message = self._makeOne(data=DATA, message_id=MESSAGE_ID)
+
+        def _to_fail():
+            return message.timestamp
+
+        self.assertRaises(ValueError, _to_fail)
+
+    def test_timestamp_wo_timestamp_in_attributes(self):
+        DATA = b'DEADBEEF'
+        MESSAGE_ID = b'12345'
+        ATTRS = {'a': 'b'}
+        message = self._makeOne(data=DATA, message_id=MESSAGE_ID,
+                                attributes=ATTRS)
+
+        def _to_fail():
+            return message.timestamp
+
+        self.assertRaises(ValueError, _to_fail)
+
+    def test_timestamp_w_timestamp_in_attributes(self):
+        from datetime import datetime
+        from pytz import utc
+        from gcloud._helpers import _RFC3339_MICROS
+        DATA = b'DEADBEEF'
+        MESSAGE_ID = b'12345'
+        TIMESTAMP = '2015-04-10T18:42:27.131956Z'
+        naive = datetime.strptime(TIMESTAMP, _RFC3339_MICROS)
+        timestamp = naive.replace(tzinfo=utc)
+        ATTRS = {'timestamp': TIMESTAMP}
+        message = self._makeOne(data=DATA, message_id=MESSAGE_ID,
+                                attributes=ATTRS)
+        self.assertEqual(message.timestamp, timestamp)
diff --git a/gcloud/search/test_subscription.py b/gcloud/search/test_subscription.py
new file mode 100644
index 0000000000000..db966ac3c12a9
--- /dev/null
+++ b/gcloud/search/test_subscription.py
@@ -0,0 +1,520 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+import unittest2
+
+
+class TestSubscription(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.subscription import Subscription
+        return Subscription
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_defaults(self):
+        SUB_NAME = 'sub_name'
+        topic = object()
+        subscription = self._makeOne(SUB_NAME, topic)
+        self.assertEqual(subscription.name, SUB_NAME)
+        self.assertTrue(subscription.topic is topic)
+        self.assertEqual(subscription.ack_deadline, None)
+        self.assertEqual(subscription.push_endpoint, None)
+
+    def test_ctor_explicit(self):
+        SUB_NAME = 'sub_name'
+        DEADLINE = 42
+        ENDPOINT = 'https://api.example.com/push'
+        topic = object()
+        subscription = self._makeOne(SUB_NAME, topic, DEADLINE, ENDPOINT)
+        self.assertEqual(subscription.name, SUB_NAME)
+        self.assertTrue(subscription.topic is topic)
+        self.assertEqual(subscription.ack_deadline, DEADLINE)
+        self.assertEqual(subscription.push_endpoint, ENDPOINT)
+
+    def test_from_api_repr_no_topics(self):
+        from gcloud.pubsub.topic import Topic
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        DEADLINE = 42
+        ENDPOINT = 'https://api.example.com/push'
+        resource = {'topic': TOPIC_PATH,
+                    'name': SUB_PATH,
+                    'ackDeadlineSeconds': DEADLINE,
+                    'pushConfig': {'pushEndpoint': ENDPOINT}}
+        klass = self._getTargetClass()
+        client = _Client(project=PROJECT)
+        subscription = klass.from_api_repr(resource, client)
+        self.assertEqual(subscription.name, SUB_NAME)
+        topic = subscription.topic
+        self.assertTrue(isinstance(topic, Topic))
+        self.assertEqual(topic.name, TOPIC_NAME)
+        self.assertEqual(topic.project, PROJECT)
+        self.assertEqual(subscription.ack_deadline, DEADLINE)
+        self.assertEqual(subscription.push_endpoint, ENDPOINT)
+
+    def test_from_api_repr_w_topics_no_topic_match(self):
+        from gcloud.pubsub.topic import Topic
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        DEADLINE = 42
+        ENDPOINT = 'https://api.example.com/push'
+        resource = {'topic': TOPIC_PATH,
+                    'name': SUB_PATH,
+                    'ackDeadlineSeconds': DEADLINE,
+                    'pushConfig': {'pushEndpoint': ENDPOINT}}
+        topics = {}
+        klass = self._getTargetClass()
+        client = _Client(project=PROJECT)
+        subscription = klass.from_api_repr(resource, client, topics=topics)
+        self.assertEqual(subscription.name, SUB_NAME)
+        topic = subscription.topic
+        self.assertTrue(isinstance(topic, Topic))
+        self.assertTrue(topic is topics[TOPIC_PATH])
+        self.assertEqual(topic.name, TOPIC_NAME)
+        self.assertEqual(topic.project, PROJECT)
+        self.assertEqual(subscription.ack_deadline, DEADLINE)
+        self.assertEqual(subscription.push_endpoint, ENDPOINT)
+
+    def test_from_api_repr_w_topics_w_topic_match(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        DEADLINE = 42
+        ENDPOINT = 'https://api.example.com/push'
+        resource = {'topic': TOPIC_PATH,
+                    'name': SUB_PATH,
+                    'ackDeadlineSeconds': DEADLINE,
+                    'pushConfig': {'pushEndpoint': ENDPOINT}}
+        topic = object()
+        topics = {TOPIC_PATH: topic}
+        klass = self._getTargetClass()
+        client = _Client(project=PROJECT)
+        subscription = klass.from_api_repr(resource, client, topics=topics)
+        self.assertEqual(subscription.name, SUB_NAME)
+        self.assertTrue(subscription.topic is topic)
+        self.assertEqual(subscription.ack_deadline, DEADLINE)
+        self.assertEqual(subscription.push_endpoint, ENDPOINT)
+
+    def test_create_pull_wo_ack_deadline_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        BODY = {'topic': TOPIC_PATH}
+        conn = _Connection({'name': SUB_PATH})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.create()
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'PUT')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+        self.assertEqual(req['data'], BODY)
+
+    def test_create_push_w_ack_deadline_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        DEADLINE = 42
+        ENDPOINT = 'https://api.example.com/push'
+        BODY = {'topic': TOPIC_PATH,
+                'ackDeadline': DEADLINE,
+                'pushConfig': {'pushEndpoint': ENDPOINT}}
+        conn1 = _Connection({'name': SUB_PATH})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'name': SUB_PATH})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic, DEADLINE, ENDPOINT)
+        subscription.create(client=CLIENT2)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'PUT')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+        self.assertEqual(req['data'], BODY)
+
+    def test_exists_miss_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        conn = _Connection()
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        self.assertFalse(subscription.exists())
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+        self.assertEqual(req.get('query_params'), None)
+
+    def test_exists_hit_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn1 = _Connection({'name': SUB_PATH, 'topic': TOPIC_PATH})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'name': SUB_PATH, 'topic': TOPIC_PATH})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic)
+        self.assertTrue(subscription.exists(client=CLIENT2))
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+        self.assertEqual(req.get('query_params'), None)
+
+    def test_reload_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        DEADLINE = 42
+        ENDPOINT = 'https://api.example.com/push'
+        conn = _Connection({'name': SUB_PATH,
+                            'topic': TOPIC_PATH,
+                            'ackDeadline': DEADLINE,
+                            'pushConfig': {'pushEndpoint': ENDPOINT}})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.reload()
+        self.assertEqual(subscription.ack_deadline, DEADLINE)
+        self.assertEqual(subscription.push_endpoint, ENDPOINT)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+
+    def test_reload_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        DEADLINE = 42
+        ENDPOINT = 'https://api.example.com/push'
+        conn1 = _Connection()
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'name': SUB_PATH,
+                             'topic': TOPIC_PATH,
+                             'ackDeadline': DEADLINE,
+                             'pushConfig': {'pushEndpoint': ENDPOINT}})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.reload(client=CLIENT2)
+        self.assertEqual(subscription.ack_deadline, DEADLINE)
+        self.assertEqual(subscription.push_endpoint, ENDPOINT)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+
+    def test_modify_push_config_w_endpoint_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ENDPOINT = 'https://api.example.com/push'
+        conn = _Connection({})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.modify_push_configuration(push_endpoint=ENDPOINT)
+        self.assertEqual(subscription.push_endpoint, ENDPOINT)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:modifyPushConfig' % SUB_PATH)
+        self.assertEqual(req['data'],
+                         {'pushConfig': {'pushEndpoint': ENDPOINT}})
+
+    def test_modify_push_config_wo_endpoint_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ENDPOINT = 'https://api.example.com/push'
+        conn1 = _Connection({})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic, push_endpoint=ENDPOINT)
+        subscription.modify_push_configuration(push_endpoint=None,
+                                               client=CLIENT2)
+        self.assertEqual(subscription.push_endpoint, None)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:modifyPushConfig' % SUB_PATH)
+        self.assertEqual(req['data'], {'pushConfig': {}})
+
+    def test_pull_wo_return_immediately_max_messages_w_bound_client(self):
+        import base64
+        from gcloud.pubsub.message import Message
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ACK_ID = 'DEADBEEF'
+        MSG_ID = 'BEADCAFE'
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD)
+        MESSAGE = {'messageId': MSG_ID, 'data': B64}
+        REC_MESSAGE = {'ackId': ACK_ID, 'message': MESSAGE}
+        conn = _Connection({'receivedMessages': [REC_MESSAGE]})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        pulled = subscription.pull()
+        self.assertEqual(len(pulled), 1)
+        ack_id, message = pulled[0]
+        self.assertEqual(ack_id, ACK_ID)
+        self.assertTrue(isinstance(message, Message))
+        self.assertEqual(message.data, PAYLOAD)
+        self.assertEqual(message.message_id, MSG_ID)
+        self.assertEqual(message.attributes, {})
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:pull' % SUB_PATH)
+        self.assertEqual(req['data'],
+                         {'returnImmediately': False, 'maxMessages': 1})
+
+    def test_pull_w_return_immediately_w_max_messages_w_alt_client(self):
+        import base64
+        from gcloud.pubsub.message import Message
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ACK_ID = 'DEADBEEF'
+        MSG_ID = 'BEADCAFE'
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD)
+        MESSAGE = {'messageId': MSG_ID, 'data': B64, 'attributes': {'a': 'b'}}
+        REC_MESSAGE = {'ackId': ACK_ID, 'message': MESSAGE}
+        conn1 = _Connection()
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'receivedMessages': [REC_MESSAGE]})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic)
+        pulled = subscription.pull(return_immediately=True, max_messages=3,
+                                   client=CLIENT2)
+        self.assertEqual(len(pulled), 1)
+        ack_id, message = pulled[0]
+        self.assertEqual(ack_id, ACK_ID)
+        self.assertTrue(isinstance(message, Message))
+        self.assertEqual(message.data, PAYLOAD)
+        self.assertEqual(message.message_id, MSG_ID)
+        self.assertEqual(message.attributes, {'a': 'b'})
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:pull' % SUB_PATH)
+        self.assertEqual(req['data'],
+                         {'returnImmediately': True, 'maxMessages': 3})
+
+    def test_pull_wo_receivedMessages(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        conn = _Connection({})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        pulled = subscription.pull(return_immediately=False)
+        self.assertEqual(len(pulled), 0)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:pull' % SUB_PATH)
+        self.assertEqual(req['data'],
+                         {'returnImmediately': False, 'maxMessages': 1})
+
+    def test_acknowledge_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ACK_ID1 = 'DEADBEEF'
+        ACK_ID2 = 'BEADCAFE'
+        conn = _Connection({})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.acknowledge([ACK_ID1, ACK_ID2])
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:acknowledge' % SUB_PATH)
+        self.assertEqual(req['data'], {'ackIds': [ACK_ID1, ACK_ID2]})
+
+    def test_acknowledge_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ACK_ID1 = 'DEADBEEF'
+        ACK_ID2 = 'BEADCAFE'
+        conn1 = _Connection({})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.acknowledge([ACK_ID1, ACK_ID2], client=CLIENT2)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:acknowledge' % SUB_PATH)
+        self.assertEqual(req['data'], {'ackIds': [ACK_ID1, ACK_ID2]})
+
+    def test_modify_ack_deadline_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ACK_ID = 'DEADBEEF'
+        DEADLINE = 42
+        conn = _Connection({})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.modify_ack_deadline(ACK_ID, DEADLINE)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:modifyAckDeadline' % SUB_PATH)
+        self.assertEqual(req['data'],
+                         {'ackId': ACK_ID, 'ackDeadlineSeconds': DEADLINE})
+
+    def test_modify_ack_deadline_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        ACK_ID = 'DEADBEEF'
+        DEADLINE = 42
+        conn1 = _Connection({})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.modify_ack_deadline(ACK_ID, DEADLINE, client=CLIENT2)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:modifyAckDeadline' % SUB_PATH)
+        self.assertEqual(req['data'],
+                         {'ackId': ACK_ID, 'ackDeadlineSeconds': DEADLINE})
+
+    def test_delete_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        conn = _Connection({})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.delete()
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'DELETE')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+
+    def test_delete_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'sub_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        conn1 = _Connection({})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic)
+        subscription.delete(client=CLIENT2)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'DELETE')
+        self.assertEqual(req['path'], '/%s' % SUB_PATH)
+
+
+class _Connection(object):
+
+    def __init__(self, *responses):
+        self._responses = responses
+        self._requested = []
+
+    def api_request(self, **kw):
+        from gcloud.exceptions import NotFound
+        self._requested.append(kw)
+
+        try:
+            response, self._responses = self._responses[0], self._responses[1:]
+        except:
+            raise NotFound('miss')
+        else:
+            return response
+
+
+class _Topic(object):
+
+    def __init__(self, name, client):
+        self.name = name
+        self._client = client
+        self.project = client.project
+        self.full_name = 'projects/%s/topics/%s' % (client.project, name)
+        self.path = '/projects/%s/topics/%s' % (client.project, name)
+
+
+class _Client(object):
+
+    def __init__(self, project, connection=None):
+        self.project = project
+        self.connection = connection
diff --git a/gcloud/search/test_topic.py b/gcloud/search/test_topic.py
new file mode 100644
index 0000000000000..4d4942db6a23d
--- /dev/null
+++ b/gcloud/search/test_topic.py
@@ -0,0 +1,550 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+import unittest2
+
+
+class TestTopic(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.topic import Topic
+        return Topic
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_w_explicit_timestamp(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        CLIENT = _Client(project=PROJECT)
+        topic = self._makeOne(TOPIC_NAME,
+                              client=CLIENT,
+                              timestamp_messages=True)
+        self.assertEqual(topic.name, TOPIC_NAME)
+        self.assertEqual(topic.project, PROJECT)
+        self.assertEqual(topic.full_name,
+                         'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME))
+        self.assertTrue(topic.timestamp_messages)
+
+    def test_from_api_repr(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        CLIENT = _Client(project=PROJECT)
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        resource = {'name': PATH}
+        klass = self._getTargetClass()
+        topic = klass.from_api_repr(resource, client=CLIENT)
+        self.assertEqual(topic.name, TOPIC_NAME)
+        self.assertTrue(topic._client is CLIENT)
+        self.assertEqual(topic.project, PROJECT)
+        self.assertEqual(topic.full_name, PATH)
+
+    def test_from_api_repr_with_bad_client(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT1 = 'PROJECT1'
+        PROJECT2 = 'PROJECT2'
+        CLIENT = _Client(project=PROJECT1)
+        PATH = 'projects/%s/topics/%s' % (PROJECT2, TOPIC_NAME)
+        resource = {'name': PATH}
+        klass = self._getTargetClass()
+        self.assertRaises(ValueError, klass.from_api_repr,
+                          resource, client=CLIENT)
+
+    def test_create_w_bound_client(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn = _Connection({'name': PATH})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+        topic.create()
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'PUT')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
+    def test_create_w_alternate_client(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn1 = _Connection({'name': PATH})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'name': PATH})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT1)
+        topic.create(client=CLIENT2)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'PUT')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
+    def test_exists_miss_w_bound_client(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn = _Connection()
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+        self.assertFalse(topic.exists())
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
+    def test_exists_hit_w_alternate_client(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn1 = _Connection({'name': PATH})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'name': PATH})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT1)
+        self.assertTrue(topic.exists(client=CLIENT2))
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
+    def test_publish_single_bytes_wo_attrs_w_bound_client(self):
+        import base64
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD).decode('ascii')
+        MSGID = 'DEADBEEF'
+        MESSAGE = {'data': B64,
+                   'attributes': {}}
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn = _Connection({'messageIds': [MSGID]})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+        msgid = topic.publish(PAYLOAD)
+        self.assertEqual(msgid, MSGID)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:publish' % PATH)
+        self.assertEqual(req['data'], {'messages': [MESSAGE]})
+
+    def test_publish_single_bytes_wo_attrs_w_add_timestamp_alt_client(self):
+        import base64
+        import datetime
+        from gcloud.pubsub import topic as MUT
+        from gcloud._helpers import _RFC3339_MICROS
+        from gcloud._testing import _Monkey
+        NOW = datetime.datetime.utcnow()
+
+        def _utcnow():
+            return NOW
+
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD).decode('ascii')
+        MSGID = 'DEADBEEF'
+        MESSAGE = {'data': B64,
+                   'attributes': {'timestamp': NOW.strftime(_RFC3339_MICROS)}}
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn1 = _Connection({'messageIds': [MSGID]})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'messageIds': [MSGID]})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT1,
+                              timestamp_messages=True)
+        with _Monkey(MUT, _NOW=_utcnow):
+            msgid = topic.publish(PAYLOAD, client=CLIENT2)
+
+        self.assertEqual(msgid, MSGID)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:publish' % PATH)
+        self.assertEqual(req['data'], {'messages': [MESSAGE]})
+
+    def test_publish_single_bytes_w_add_timestamp_w_ts_in_attrs(self):
+        import base64
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD).decode('ascii')
+        MSGID = 'DEADBEEF'
+        OVERRIDE = '2015-04-10T16:46:22.868399Z'
+        MESSAGE = {'data': B64,
+                   'attributes': {'timestamp': OVERRIDE}}
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn = _Connection({'messageIds': [MSGID]})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT,
+                              timestamp_messages=True)
+        msgid = topic.publish(PAYLOAD, timestamp=OVERRIDE)
+        self.assertEqual(msgid, MSGID)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:publish' % PATH)
+        self.assertEqual(req['data'], {'messages': [MESSAGE]})
+
+    def test_publish_single_w_attrs(self):
+        import base64
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD).decode('ascii')
+        MSGID = 'DEADBEEF'
+        MESSAGE = {'data': B64,
+                   'attributes': {'attr1': 'value1', 'attr2': 'value2'}}
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn = _Connection({'messageIds': [MSGID]})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+        msgid = topic.publish(PAYLOAD, attr1='value1', attr2='value2')
+        self.assertEqual(msgid, MSGID)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:publish' % PATH)
+        self.assertEqual(req['data'], {'messages': [MESSAGE]})
+
+    def test_publish_multiple_w_bound_client(self):
+        import base64
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PAYLOAD1 = b'This is the first message text'
+        PAYLOAD2 = b'This is the second message text'
+        B64_1 = base64.b64encode(PAYLOAD1)
+        B64_2 = base64.b64encode(PAYLOAD2)
+        MSGID1 = 'DEADBEEF'
+        MSGID2 = 'BEADCAFE'
+        MESSAGE1 = {'data': B64_1.decode('ascii'),
+                    'attributes': {}}
+        MESSAGE2 = {'data': B64_2.decode('ascii'),
+                    'attributes': {'attr1': 'value1', 'attr2': 'value2'}}
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+        with topic.batch() as batch:
+            batch.publish(PAYLOAD1)
+            batch.publish(PAYLOAD2, attr1='value1', attr2='value2')
+        self.assertEqual(list(batch), [MSGID1, MSGID2])
+        self.assertEqual(list(batch.messages), [])
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:publish' % PATH)
+        self.assertEqual(req['data'], {'messages': [MESSAGE1, MESSAGE2]})
+
+    def test_publish_multiple_w_alternate_client(self):
+        import base64
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PAYLOAD1 = b'This is the first message text'
+        PAYLOAD2 = b'This is the second message text'
+        B64_1 = base64.b64encode(PAYLOAD1)
+        B64_2 = base64.b64encode(PAYLOAD2)
+        MSGID1 = 'DEADBEEF'
+        MSGID2 = 'BEADCAFE'
+        MESSAGE1 = {'data': B64_1.decode('ascii'),
+                    'attributes': {}}
+        MESSAGE2 = {'data': B64_2.decode('ascii'),
+                    'attributes': {'attr1': 'value1', 'attr2': 'value2'}}
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn1 = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT1)
+        with topic.batch(client=CLIENT2) as batch:
+            batch.publish(PAYLOAD1)
+            batch.publish(PAYLOAD2, attr1='value1', attr2='value2')
+        self.assertEqual(list(batch), [MSGID1, MSGID2])
+        self.assertEqual(list(batch.messages), [])
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s:publish' % PATH)
+        self.assertEqual(req['data'], {'messages': [MESSAGE1, MESSAGE2]})
+
+    def test_publish_multiple_error(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PAYLOAD1 = b'This is the first message text'
+        PAYLOAD2 = b'This is the second message text'
+        MSGID1 = 'DEADBEEF'
+        MSGID2 = 'BEADCAFE'
+        conn = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT = _Client(project=PROJECT)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+        try:
+            with topic.batch() as batch:
+                batch.publish(PAYLOAD1)
+                batch.publish(PAYLOAD2, attr1='value1', attr2='value2')
+                raise _Bugout()
+        except _Bugout:
+            pass
+        self.assertEqual(list(batch), [])
+        self.assertEqual(len(conn._requested), 0)
+
+    def test_delete_w_bound_client(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn = _Connection({})
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+        topic.delete()
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'DELETE')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
+    def test_delete_w_alternate_client(self):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        conn1 = _Connection({})
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection({})
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT1)
+        topic.delete(client=CLIENT2)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'DELETE')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
+
+class TestBatch(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.topic import Batch
+        return Batch
+
+    def _makeOne(self, *args, **kwargs):
+        return self._getTargetClass()(*args, **kwargs)
+
+    def test_ctor_defaults(self):
+        topic = _Topic()
+        CLIENT = _Client(project='PROJECT')
+        batch = self._makeOne(topic, CLIENT)
+        self.assertTrue(batch.topic is topic)
+        self.assertTrue(batch.client is CLIENT)
+        self.assertEqual(len(batch.messages), 0)
+        self.assertEqual(len(batch.message_ids), 0)
+
+    def test___iter___empty(self):
+        topic = _Topic()
+        client = object()
+        batch = self._makeOne(topic, client)
+        self.assertEqual(list(batch), [])
+
+    def test___iter___non_empty(self):
+        topic = _Topic()
+        client = object()
+        batch = self._makeOne(topic, client)
+        batch.message_ids[:] = ['ONE', 'TWO', 'THREE']
+        self.assertEqual(list(batch), ['ONE', 'TWO', 'THREE'])
+
+    def test_publish_bytes_wo_attrs(self):
+        import base64
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD).decode('ascii')
+        MESSAGE = {'data': B64,
+                   'attributes': {}}
+        connection = _Connection()
+        CLIENT = _Client(project='PROJECT', connection=connection)
+        topic = _Topic()
+        batch = self._makeOne(topic, client=CLIENT)
+        batch.publish(PAYLOAD)
+        self.assertEqual(len(connection._requested), 0)
+        self.assertEqual(batch.messages, [MESSAGE])
+
+    def test_publish_bytes_w_add_timestamp(self):
+        import base64
+        PAYLOAD = b'This is the message text'
+        B64 = base64.b64encode(PAYLOAD).decode('ascii')
+        MESSAGE = {'data': B64,
+                   'attributes': {'timestamp': 'TIMESTAMP'}}
+        connection = _Connection()
+        CLIENT = _Client(project='PROJECT', connection=connection)
+        topic = _Topic(timestamp_messages=True)
+        batch = self._makeOne(topic, client=CLIENT)
+        batch.publish(PAYLOAD)
+        self.assertEqual(len(connection._requested), 0)
+        self.assertEqual(batch.messages, [MESSAGE])
+
+    def test_commit_w_bound_client(self):
+        import base64
+        PAYLOAD1 = b'This is the first message text'
+        PAYLOAD2 = b'This is the second message text'
+        B64_1 = base64.b64encode(PAYLOAD1)
+        B64_2 = base64.b64encode(PAYLOAD2)
+        MSGID1 = 'DEADBEEF'
+        MSGID2 = 'BEADCAFE'
+        MESSAGE1 = {'data': B64_1.decode('ascii'),
+                    'attributes': {}}
+        MESSAGE2 = {'data': B64_2.decode('ascii'),
+                    'attributes': {'attr1': 'value1', 'attr2': 'value2'}}
+        conn = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT = _Client(project='PROJECT', connection=conn)
+        topic = _Topic()
+        batch = self._makeOne(topic, client=CLIENT)
+        batch.publish(PAYLOAD1)
+        batch.publish(PAYLOAD2, attr1='value1', attr2='value2')
+        batch.commit()
+        self.assertEqual(list(batch), [MSGID1, MSGID2])
+        self.assertEqual(list(batch.messages), [])
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '%s:publish' % topic.path)
+        self.assertEqual(req['data'], {'messages': [MESSAGE1, MESSAGE2]})
+
+    def test_commit_w_alternate_client(self):
+        import base64
+        PAYLOAD1 = b'This is the first message text'
+        PAYLOAD2 = b'This is the second message text'
+        B64_1 = base64.b64encode(PAYLOAD1)
+        B64_2 = base64.b64encode(PAYLOAD2)
+        MSGID1 = 'DEADBEEF'
+        MSGID2 = 'BEADCAFE'
+        MESSAGE1 = {'data': B64_1.decode('ascii'),
+                    'attributes': {}}
+        MESSAGE2 = {'data': B64_2.decode('ascii'),
+                    'attributes': {'attr1': 'value1', 'attr2': 'value2'}}
+        conn1 = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT1 = _Client(project='PROJECT', connection=conn1)
+        conn2 = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT2 = _Client(project='PROJECT', connection=conn2)
+        topic = _Topic()
+        batch = self._makeOne(topic, client=CLIENT1)
+        batch.publish(PAYLOAD1)
+        batch.publish(PAYLOAD2, attr1='value1', attr2='value2')
+        batch.commit(client=CLIENT2)
+        self.assertEqual(list(batch), [MSGID1, MSGID2])
+        self.assertEqual(list(batch.messages), [])
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '%s:publish' % topic.path)
+        self.assertEqual(req['data'], {'messages': [MESSAGE1, MESSAGE2]})
+
+    def test_context_mgr_success(self):
+        import base64
+        PAYLOAD1 = b'This is the first message text'
+        PAYLOAD2 = b'This is the second message text'
+        B64_1 = base64.b64encode(PAYLOAD1)
+        B64_2 = base64.b64encode(PAYLOAD2)
+        MSGID1 = 'DEADBEEF'
+        MSGID2 = 'BEADCAFE'
+        MESSAGE1 = {'data': B64_1.decode('ascii'),
+                    'attributes': {}}
+        MESSAGE2 = {'data': B64_2.decode('ascii'),
+                    'attributes': {'attr1': 'value1', 'attr2': 'value2'}}
+        conn = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT = _Client(project='PROJECT', connection=conn)
+        topic = _Topic()
+        batch = self._makeOne(topic, client=CLIENT)
+
+        with batch as other:
+            batch.publish(PAYLOAD1)
+            batch.publish(PAYLOAD2, attr1='value1', attr2='value2')
+
+        self.assertTrue(other is batch)
+        self.assertEqual(list(batch), [MSGID1, MSGID2])
+        self.assertEqual(list(batch.messages), [])
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '%s:publish' % topic.path)
+        self.assertEqual(req['data'], {'messages': [MESSAGE1, MESSAGE2]})
+
+    def test_context_mgr_failure(self):
+        import base64
+        PAYLOAD1 = b'This is the first message text'
+        PAYLOAD2 = b'This is the second message text'
+        B64_1 = base64.b64encode(PAYLOAD1)
+        B64_2 = base64.b64encode(PAYLOAD2)
+        MSGID1 = 'DEADBEEF'
+        MSGID2 = 'BEADCAFE'
+        MESSAGE1 = {'data': B64_1.decode('ascii'),
+                    'attributes': {}}
+        MESSAGE2 = {'data': B64_2.decode('ascii'),
+                    'attributes': {'attr1': 'value1', 'attr2': 'value2'}}
+        conn = _Connection({'messageIds': [MSGID1, MSGID2]})
+        CLIENT = _Client(project='PROJECT', connection=conn)
+        topic = _Topic()
+        batch = self._makeOne(topic, client=CLIENT)
+
+        try:
+            with batch as other:
+                batch.publish(PAYLOAD1)
+                batch.publish(PAYLOAD2, attr1='value1', attr2='value2')
+                raise _Bugout()
+        except _Bugout:
+            pass
+
+        self.assertTrue(other is batch)
+        self.assertEqual(list(batch), [])
+        self.assertEqual(list(batch.messages), [MESSAGE1, MESSAGE2])
+        self.assertEqual(len(conn._requested), 0)
+
+
+class _Connection(object):
+
+    def __init__(self, *responses):
+        self._responses = responses
+        self._requested = []
+
+    def api_request(self, **kw):
+        from gcloud.exceptions import NotFound
+        self._requested.append(kw)
+
+        try:
+            response, self._responses = self._responses[0], self._responses[1:]
+        except:
+            raise NotFound('miss')
+        else:
+            return response
+
+
+class _Topic(object):
+
+    def __init__(self, name="NAME", project="PROJECT",
+                 timestamp_messages=False):
+        self.path = '/projects/%s/topics/%s' % (project, name)
+        self.timestamp_messages = timestamp_messages
+
+    def _timestamp_message(self, attrs):
+        if self.timestamp_messages:
+            attrs['timestamp'] = 'TIMESTAMP'
+
+
+class _Client(object):
+
+    def __init__(self, project, connection=None):
+        self.project = project
+        self.connection = connection
+
+
+class _Bugout(Exception):
+    pass
diff --git a/gcloud/search/value.py b/gcloud/search/value.py
new file mode 100644
index 0000000000000..5efc56c328f38
--- /dev/null
+++ b/gcloud/search/value.py
@@ -0,0 +1,44 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# Licensed 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.
+
+"""Define Cloud Search Values."""
+
+
+class Value(object):
+    """Value objects hold search values and tokenization parameters.
+
+    See:
+    https://cloud.google.com/pubsub/reference/rest/v1beta2/projects/subscriptions
+
+    :type value: string
+    :param value: the string value
+
+    :type tokenization: string
+    :param tokenization: the tokenization format for string values.
+    """
+    def __init__(self, value, tokenization=None):
+        self.value = value
+        self.tokenization = tokenization
+
+    @classmethod
+    def from_api_repr(cls, resource):
+        """Factory:  construct a topic given its API representation
+
+        :type resource: dict
+        :param resource: topic resource representation returned from the API
+
+        :rtype: :class:`gcloud.search.value.Value`
+        :returns: Value parsed from ``resource``.
+        """
+        return cls(value=resource['value'])