PubSub Publisher

pubsub/google/cloud/pubsub.py

@@ -0,0 +1,26 @@
+# Copyright 2017, 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.
+
+from __future__ import absolute_import
+
+from google.cloud.pubsub_v1 import PublisherClient
+from google.cloud.pubsub_v1 import SubscriberClient
+from google.cloud.pubsub_v1 import types
+
+
+__all__ = (
+    'PublisherClient',
+    'SubscriberClient',
+    'types',
+)

pubsub/google/cloud/pubsub_v1/__init__.py

@@ -0,0 +1,25 @@
+# Copyright 2017, 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.
+
+from __future__ import absolute_import
+
+from google.cloud.pubsub_v1 import types
+from google.cloud.pubsub_v1.publisher import PublisherClient
+from google.cloud.pubsub_v1.subscriber import SubscriberClient
+
+__all__ = (
+    'PublisherClient',
+    'SubscriberClient',
+    'types',
+)

pubsub/google/cloud/pubsub_v1/_gapic.py

@@ -0,0 +1,73 @@
+# Copyright 2017, 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.
+
+from __future__ import absolute_import
+
+import functools
+
+
+def add_methods(source_class, blacklist=()):
+    """Add wrapped versions of the `api` member's methods to the class.
+
+    Any methods passed in `blacklist` are not added.
+    Additionally, any methods explicitly defined on the wrapped class are
+    not added.
+    """
+    def wrap(wrapped_fx):
+        """Wrap a GAPIC method; preserve its name and docstring."""
+        # If this is a static or class method, then we need to *not*
+        # send self as the first argument.
+        #
+        # Similarly, for instance methods, we need to send self.api rather
+        # than self, since that is where the actual methods were declared.
+        instance_method = True
+        self = getattr(wrapped_fx, '__self__', None)
+        if issubclass(type(self), type):
+            instance_method = False
+
+        # Okay, we have figured out what kind of method this is; send
+        # down the correct wrapper function.
+        if instance_method:
+            fx = lambda self, *a, **kw: wrapped_fx(self.api, *a, **kw)
+            return functools.wraps(wrapped_fx)(fx)
+        fx = lambda self, *a, **kw: wrapped_fx(*a, **kw)
+        return functools.wraps(wrapped_fx)(fx)
+
+    def actual_decorator(cls):
+        # Reflectively iterate over most of the methods on the source class
+        # (the GAPIC) and make wrapped versions available on this client.
+        for name in dir(source_class):
+            # Ignore all private and magic methods.
+            if name.startswith('_'):
+                continue
+
+            # Ignore anything on our blacklist.
+            if name in blacklist:
+                continue
+
+            # Retrieve the attribute, and ignore it if it is not callable.
+            attr = getattr(source_class, name)
+            if not callable(attr):
+                continue
+
+            # Add a wrapper method to this object.
+            fx = wrap(getattr(source_class, name))
+            setattr(cls, name, fx)
+
+        # Return the augmented class.
+        return cls
+
+    # Simply return the actual decorator; this is returned from this method
+    # and actually used to decorate the class.
+    return actual_decorator

pubsub/tests/unit/__init__.py → pubsub/google/cloud/pubsub_v1/publisher/__init__.py

@@ -1,4 +1,4 @@
-# Copyright 2016 Google Inc.
+# Copyright 2017, 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.
@@ -11,3 +11,12 @@
 # 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.
+
+from __future__ import absolute_import
+
+from google.cloud.pubsub_v1.publisher.client import Client
+
+
+__all__ = (
+    'Client',
+)

pubsub/google/cloud/pubsub_v1/publisher/batch/base.py

@@ -0,0 +1,145 @@
+# Copyright 2017, 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.
+
+from __future__ import absolute_import
+
+import abc
+
+import six
+
+
+@six.add_metaclass(abc.ABCMeta)
+class Batch(object):
+    """The base batching class for Pub/Sub publishing.
+
+    Although the :class:`~.pubsub_v1.publisher.batch.thread.Batch` class, based
+    on :class:`threading.Thread`, is fine for most cases, advanced
+    users may need to implement something based on a different concurrency
+    model.
+
+    This class defines the interface for the Batch implementation;
+    subclasses may be passed as the ``batch_class`` argument to
+    :class:`~.pubsub_v1.client.PublisherClient`.
+
+    The batching behavior works like this: When the
+    :class:`~.pubsub_v1.publisher.client.Client` is asked to publish a new
+    message, it requires a batch. The client will see if there is an
+    already-opened batch for the given topic; if there is, then the message
+    is sent to that batch. If there is not, then a new batch is created
+    and the message put there.
+
+    When a new batch is created, it automatically starts a timer counting
+    down to the maximum latency before the batch should commit.
+    Essentially, if enough time passes, the batch automatically commits
+    regardless of how much is in it. However, if either the message count or
+    size thresholds are encountered first, then the batch will commit early.
+    """
+    def __len__(self):
+        """Return the number of messages currently in the batch."""
+        return len(self.messages)
+
+    @property
+    @abc.abstractmethod
+    def messages(self):
+        """Return the messages currently in the batch.
+
+        Returns:
+            Sequence: The messages currently in the batch.
+        """
+        raise NotImplementedError
+
+    @property
+    @abc.abstractmethod
+    def size(self):
+        """Return the total size of all of the messages currently in the batch.
+
+        Returns:
+            int: The total size of all of the messages currently
+                 in the batch, in bytes.
+        """
+        raise NotImplementedError
+
+    @property
+    @abc.abstractmethod
+    def settings(self):
+        """Return the batch settings.
+
+        Returns:
+            ~.pubsub_v1.types.BatchSettings: The batch settings. These are
+                considered immutable once the batch has been opened.
+        """
+        raise NotImplementedError
+
+    @property
+    @abc.abstractmethod
+    def status(self):
+        """Return the status of this batch.
+
+        Returns:
+            str: The status of this batch. All statuses are human-readable,
+                all-lowercase strings. The ones represented in the
+                :class:`BaseBatch.Status` enum are special, but other statuses
+                are permitted.
+        """
+        raise NotImplementedError
+
+    def will_accept(self, message):
+        """Return True if the batch is able to accept the message.
+
+        Args:
+            message (~.pubsub_v1.types.PubsubMessage): The Pub/Sub message.
+
+        Returns:
+            bool: Whether this batch can accept the message.
+        """
+        # If this batch is not accepting messages generally, return False.
+        if self.status != self.Status.ACCEPTING_MESSAGES:
+            return False
+
+        # If this batch can not hold the message in question, return False.
+        if self.size + message.ByteSize() > self.settings.max_bytes:
+            return False
+
+        # Okay, everything is good.
+        return True
+
+    @abc.abstractmethod
+    def publish(self, message):
+        """Publish a single message.
+
+        Add the given message to this object; this will cause it to be
+        published once the batch either has enough messages or a sufficient
+        period of time has elapsed.
+
+        This method is called by :meth:`~.PublisherClient.publish`.
+
+        Args:
+            message (~.pubsub_v1.types.PubsubMessage): The Pub/Sub message.
+
+        Returns:
+            ~.pubsub_v1.publisher.batch.mp.Future: An object conforming to the
+                :class:`concurrent.futures.Future` interface.
+        """
+        raise NotImplementedError
+
+    class Status(object):
+        """An enum class representing valid statuses for a batch.
+
+        It is acceptable for a class to use a status that is not on this
+        class; this represents the list of statuses where the existing
+        library hooks in functionality.
+        """
+        ACCEPTING_MESSAGES = 'accepting messages'
+        ERROR = 'error'
+        SUCCESS = 'success'