We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
PubSub: Message publish_time return type unexpected
I am working on a PubSub project and was a bit confused about the Message `publish_time`. I expected to get `datetime` per [the docstring](https://github.com/GoogleCloudPlatform/google-cloud-python/blob/1acc8c22664229b6681ff91654932998e611e1c2/pubsub/google/cloud/pubsub_v1/subscriber/message.py#L152) for the method. Instead I got a `Timestamp` that apparently comes from protobuf types. I found there is even a [unit test](https://github.com/GoogleCloudPlatform/google-cloud-python/blob/1acc8c22664229b6681ff91654932998e611e1c2/pubsub/tests/unit/pubsub_v1/subscriber/test_message.py#L49) that verifies that `publish_time` returns a `Timestamp`.

Can we return `datetime`? Or should the docs just be updated to explain what is actually being returned? I'd prefer the former but wanted to ask before submitting a PR, given the inconsistency.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `pubsub/google/cloud/pubsub_v1/subscriber/message.py`
Content:
```
1 # Copyright 2017, Google LLC All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from __future__ import absolute_import
16 
17 import json
18 import math
19 import time
20 
21 from google.cloud.pubsub_v1.subscriber._protocol import requests
22 
23 
24 _MESSAGE_REPR = """\
25 Message {{
26   data: {!r}
27   attributes: {}
28 }}"""
29 
30 
31 def _indent(lines, prefix='  '):
32     """Indent some text.
33 
34     Note that this is present as ``textwrap.indent``, but not in Python 2.
35 
36     Args:
37         lines (str): The newline delimited string to be indented.
38         prefix (Optional[str]): The prefix to indent each line with. Default
39             to two spaces.
40 
41     Returns:
42         str: The newly indented content.
43     """
44     indented = []
45     for line in lines.split('\n'):
46         indented.append(prefix + line)
47     return '\n'.join(indented)
48 
49 
50 class Message(object):
51     """A representation of a single Pub/Sub message.
52 
53     The common way to interact with
54     :class:`~.pubsub_v1.subscriber.message.Message` objects is to receive
55     them in callbacks on subscriptions; most users should never have a need
56     to instantiate them by hand. (The exception to this is if you are
57     implementing a custom subclass to
58     :class:`~.pubsub_v1.subscriber._consumer.Consumer`.)
59 
60     Attributes:
61         message_id (str): The message ID. In general, you should not need
62             to use this directly.
63         data (bytes): The data in the message. Note that this will be a
64             :class:`bytes`, not a text string.
65         attributes (.ScalarMapContainer): The attributes sent along with the
66             message. See :attr:`attributes` for more information on this type.
67         publish_time (datetime): The time that this message was originally
68             published.
69     """
70 
71     def __init__(self, message, ack_id, request_queue):
72         """Construct the Message.
73 
74         .. note::
75 
76             This class should not be constructed directly; it is the
77             responsibility of :class:`BasePolicy` subclasses to do so.
78 
79         Args:
80             message (~.pubsub_v1.types.PubsubMessage): The message received
81                 from Pub/Sub.
82             ack_id (str): The ack_id received from Pub/Sub.
83             request_queue (queue.Queue): A queue provided by the policy that
84                 can accept requests; the policy is responsible for handling
85                 those requests.
86         """
87         self._message = message
88         self._ack_id = ack_id
89         self._request_queue = request_queue
90         self.message_id = message.message_id
91 
92         # The instantiation time is the time that this message
93         # was received. Tracking this provides us a way to be smart about
94         # the default lease deadline.
95         self._received_timestamp = time.time()
96 
97         # The policy should lease this message, telling PubSub that it has
98         # it until it is acked or otherwise dropped.
99         self.lease()
100 
101     def __repr__(self):
102         # Get an abbreviated version of the data.
103         abbv_data = self._message.data
104         if len(abbv_data) > 50:
105             abbv_data = abbv_data[:50] + b'...'
106 
107         pretty_attrs = json.dumps(
108             dict(self.attributes),
109             indent=2,
110             separators=(',', ': '),
111             sort_keys=True,
112         )
113         pretty_attrs = _indent(pretty_attrs)
114         # We don't actually want the first line indented.
115         pretty_attrs = pretty_attrs.lstrip()
116         return _MESSAGE_REPR.format(abbv_data, pretty_attrs)
117 
118     @property
119     def attributes(self):
120         """Return the attributes of the underlying Pub/Sub Message.
121 
122         .. warning::
123 
124             A ``ScalarMapContainer`` behaves slightly differently than a
125             ``dict``. For a Pub / Sub message this is a ``string->string`` map.
126             When trying to access a value via ``map['key']``, if the key is
127             not in the map, then the default value for the string type will
128             be returned, which is an empty string. It may be more intuitive
129             to just cast the map to a ``dict`` or to one use ``map.get``.
130 
131         Returns:
132             .ScalarMapContainer: The message's attributes. This is a
133             ``dict``-like object provided by ``google.protobuf``.
134         """
135         return self._message.attributes
136 
137     @property
138     def data(self):
139         """Return the data for the underlying Pub/Sub Message.
140 
141         Returns:
142             bytes: The message data. This is always a bytestring; if you
143                 want a text string, call :meth:`bytes.decode`.
144         """
145         return self._message.data
146 
147     @property
148     def publish_time(self):
149         """Return the time that the message was originally published.
150 
151         Returns:
152             datetime: The date and time that the message was published.
153         """
154         return self._message.publish_time
155 
156     @property
157     def size(self):
158         """Return the size of the underlying message, in bytes."""
159         return self._message.ByteSize()
160 
161     def ack(self):
162         """Acknowledge the given message.
163 
164         Acknowledging a message in Pub/Sub means that you are done
165         with it, and it will not be delivered to this subscription again.
166         You should avoid acknowledging messages until you have
167         *finished* processing them, so that in the event of a failure,
168         you receive the message again.
169 
170         .. warning::
171             Acks in Pub/Sub are best effort. You should always
172             ensure that your processing code is idempotent, as you may
173             receive any given message more than once.
174         """
175         time_to_ack = math.ceil(time.time() - self._received_timestamp)
176         self._request_queue.put(
177             requests.AckRequest(
178                 ack_id=self._ack_id,
179                 byte_size=self.size,
180                 time_to_ack=time_to_ack
181             )
182         )
183 
184     def drop(self):
185         """Release the message from lease management.
186 
187         This informs the policy to no longer hold on to the lease for this
188         message. Pub/Sub will re-deliver the message if it is not acknowledged
189         before the existing lease expires.
190 
191         .. warning::
192             For most use cases, the only reason to drop a message from
193             lease management is on :meth:`ack` or :meth:`nack`; these methods
194             both call this one. You probably do not want to call this method
195             directly.
196         """
197         self._request_queue.put(
198             requests.DropRequest(
199                 ack_id=self._ack_id,
200                 byte_size=self.size
201             )
202         )
203 
204     def lease(self):
205         """Inform the policy to lease this message continually.
206 
207         .. note::
208             This method is called by the constructor, and you should never
209             need to call it manually.
210         """
211         self._request_queue.put(
212             requests.LeaseRequest(
213                 ack_id=self._ack_id,
214                 byte_size=self.size
215             )
216         )
217 
218     def modify_ack_deadline(self, seconds):
219         """Resets the deadline for acknowledgement.
220 
221         New deadline will be the given value of seconds from now.
222 
223         The default implementation handles this for you; you should not need
224         to manually deal with setting ack deadlines. The exception case is
225         if you are implementing your own custom subclass of
226         :class:`~.pubsub_v1.subcriber._consumer.Consumer`.
227 
228         Args:
229             seconds (int): The number of seconds to set the lease deadline
230                 to. This should be between 0 and 600. Due to network latency,
231                 values below 10 are advised against.
232         """
233         self._request_queue.put(
234             requests.ModAckRequest(
235                 ack_id=self._ack_id,
236                 seconds=seconds
237             )
238         )
239 
240     def nack(self):
241         """Decline to acknowldge the given message.
242 
243         This will cause the message to be re-delivered to the subscription.
244         """
245         self._request_queue.put(
246             requests.NackRequest(
247                 ack_id=self._ack_id,
248                 byte_size=self.size
249             )
250         )
251
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/pubsub/google/cloud/pubsub_v1/subscriber/message.py b/pubsub/google/cloud/pubsub_v1/subscriber/message.py
--- a/pubsub/google/cloud/pubsub_v1/subscriber/message.py
+++ b/pubsub/google/cloud/pubsub_v1/subscriber/message.py
@@ -14,10 +14,12 @@
 
 from __future__ import absolute_import
 
+import datetime
 import json
 import math
 import time
 
+from google.api_core import datetime_helpers
 from google.cloud.pubsub_v1.subscriber._protocol import requests
 
 
@@ -151,7 +153,11 @@
         Returns:
             datetime: The date and time that the message was published.
         """
-        return self._message.publish_time
+        timestamp = self._message.publish_time
+        delta = datetime.timedelta(
+            seconds=timestamp.seconds,
+            microseconds=timestamp.nanos // 1000)
+        return datetime_helpers._UTC_EPOCH + delta
 
     @property
     def size(self):

{"golden_diff": "diff --git a/pubsub/google/cloud/pubsub_v1/subscriber/message.py b/pubsub/google/cloud/pubsub_v1/subscriber/message.py\n--- a/pubsub/google/cloud/pubsub_v1/subscriber/message.py\n+++ b/pubsub/google/cloud/pubsub_v1/subscriber/message.py\n@@ -14,10 +14,12 @@\n \n from __future__ import absolute_import\n \n+import datetime\n import json\n import math\n import time\n \n+from google.api_core import datetime_helpers\n from google.cloud.pubsub_v1.subscriber._protocol import requests\n \n \n@@ -151,7 +153,11 @@\n         Returns:\n             datetime: The date and time that the message was published.\n         \"\"\"\n-        return self._message.publish_time\n+        timestamp = self._message.publish_time\n+        delta = datetime.timedelta(\n+            seconds=timestamp.seconds,\n+            microseconds=timestamp.nanos // 1000)\n+        return datetime_helpers._UTC_EPOCH + delta\n \n     @property\n     def size(self):\n", "issue": "PubSub: Message publish_time return type unexpected\nI am working on a PubSub project and was a bit confused about the Message `publish_time`. I expected to get `datetime` per [the docstring](https://github.com/GoogleCloudPlatform/google-cloud-python/blob/1acc8c22664229b6681ff91654932998e611e1c2/pubsub/google/cloud/pubsub_v1/subscriber/message.py#L152) for the method. Instead I got a `Timestamp` that apparently comes from protobuf types. I found there is even a [unit test](https://github.com/GoogleCloudPlatform/google-cloud-python/blob/1acc8c22664229b6681ff91654932998e611e1c2/pubsub/tests/unit/pubsub_v1/subscriber/test_message.py#L49) that verifies that `publish_time` returns a `Timestamp`.\r\n\r\nCan we return `datetime`? Or should the docs just be updated to explain what is actually being returned? I'd prefer the former but wanted to ask before submitting a PR, given the inconsistency.\n", "before_files": [{"content": "# Copyright 2017, Google LLC All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\n\nimport json\nimport math\nimport time\n\nfrom google.cloud.pubsub_v1.subscriber._protocol import requests\n\n\n_MESSAGE_REPR = \"\"\"\\\nMessage {{\n  data: {!r}\n  attributes: {}\n}}\"\"\"\n\n\ndef _indent(lines, prefix='  '):\n    \"\"\"Indent some text.\n\n    Note that this is present as ``textwrap.indent``, but not in Python 2.\n\n    Args:\n        lines (str): The newline delimited string to be indented.\n        prefix (Optional[str]): The prefix to indent each line with. Default\n            to two spaces.\n\n    Returns:\n        str: The newly indented content.\n    \"\"\"\n    indented = []\n    for line in lines.split('\\n'):\n        indented.append(prefix + line)\n    return '\\n'.join(indented)\n\n\nclass Message(object):\n    \"\"\"A representation of a single Pub/Sub message.\n\n    The common way to interact with\n    :class:`~.pubsub_v1.subscriber.message.Message` objects is to receive\n    them in callbacks on subscriptions; most users should never have a need\n    to instantiate them by hand. (The exception to this is if you are\n    implementing a custom subclass to\n    :class:`~.pubsub_v1.subscriber._consumer.Consumer`.)\n\n    Attributes:\n        message_id (str): The message ID. In general, you should not need\n            to use this directly.\n        data (bytes): The data in the message. Note that this will be a\n            :class:`bytes`, not a text string.\n        attributes (.ScalarMapContainer): The attributes sent along with the\n            message. See :attr:`attributes` for more information on this type.\n        publish_time (datetime): The time that this message was originally\n            published.\n    \"\"\"\n\n    def __init__(self, message, ack_id, request_queue):\n        \"\"\"Construct the Message.\n\n        .. note::\n\n            This class should not be constructed directly; it is the\n            responsibility of :class:`BasePolicy` subclasses to do so.\n\n        Args:\n            message (~.pubsub_v1.types.PubsubMessage): The message received\n                from Pub/Sub.\n            ack_id (str): The ack_id received from Pub/Sub.\n            request_queue (queue.Queue): A queue provided by the policy that\n                can accept requests; the policy is responsible for handling\n                those requests.\n        \"\"\"\n        self._message = message\n        self._ack_id = ack_id\n        self._request_queue = request_queue\n        self.message_id = message.message_id\n\n        # The instantiation time is the time that this message\n        # was received. Tracking this provides us a way to be smart about\n        # the default lease deadline.\n        self._received_timestamp = time.time()\n\n        # The policy should lease this message, telling PubSub that it has\n        # it until it is acked or otherwise dropped.\n        self.lease()\n\n    def __repr__(self):\n        # Get an abbreviated version of the data.\n        abbv_data = self._message.data\n        if len(abbv_data) > 50:\n            abbv_data = abbv_data[:50] + b'...'\n\n        pretty_attrs = json.dumps(\n            dict(self.attributes),\n            indent=2,\n            separators=(',', ': '),\n            sort_keys=True,\n        )\n        pretty_attrs = _indent(pretty_attrs)\n        # We don't actually want the first line indented.\n        pretty_attrs = pretty_attrs.lstrip()\n        return _MESSAGE_REPR.format(abbv_data, pretty_attrs)\n\n    @property\n    def attributes(self):\n        \"\"\"Return the attributes of the underlying Pub/Sub Message.\n\n        .. warning::\n\n            A ``ScalarMapContainer`` behaves slightly differently than a\n            ``dict``. For a Pub / Sub message this is a ``string->string`` map.\n            When trying to access a value via ``map['key']``, if the key is\n            not in the map, then the default value for the string type will\n            be returned, which is an empty string. It may be more intuitive\n            to just cast the map to a ``dict`` or to one use ``map.get``.\n\n        Returns:\n            .ScalarMapContainer: The message's attributes. This is a\n            ``dict``-like object provided by ``google.protobuf``.\n        \"\"\"\n        return self._message.attributes\n\n    @property\n    def data(self):\n        \"\"\"Return the data for the underlying Pub/Sub Message.\n\n        Returns:\n            bytes: The message data. This is always a bytestring; if you\n                want a text string, call :meth:`bytes.decode`.\n        \"\"\"\n        return self._message.data\n\n    @property\n    def publish_time(self):\n        \"\"\"Return the time that the message was originally published.\n\n        Returns:\n            datetime: The date and time that the message was published.\n        \"\"\"\n        return self._message.publish_time\n\n    @property\n    def size(self):\n        \"\"\"Return the size of the underlying message, in bytes.\"\"\"\n        return self._message.ByteSize()\n\n    def ack(self):\n        \"\"\"Acknowledge the given message.\n\n        Acknowledging a message in Pub/Sub means that you are done\n        with it, and it will not be delivered to this subscription again.\n        You should avoid acknowledging messages until you have\n        *finished* processing them, so that in the event of a failure,\n        you receive the message again.\n\n        .. warning::\n            Acks in Pub/Sub are best effort. You should always\n            ensure that your processing code is idempotent, as you may\n            receive any given message more than once.\n        \"\"\"\n        time_to_ack = math.ceil(time.time() - self._received_timestamp)\n        self._request_queue.put(\n            requests.AckRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size,\n                time_to_ack=time_to_ack\n            )\n        )\n\n    def drop(self):\n        \"\"\"Release the message from lease management.\n\n        This informs the policy to no longer hold on to the lease for this\n        message. Pub/Sub will re-deliver the message if it is not acknowledged\n        before the existing lease expires.\n\n        .. warning::\n            For most use cases, the only reason to drop a message from\n            lease management is on :meth:`ack` or :meth:`nack`; these methods\n            both call this one. You probably do not want to call this method\n            directly.\n        \"\"\"\n        self._request_queue.put(\n            requests.DropRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size\n            )\n        )\n\n    def lease(self):\n        \"\"\"Inform the policy to lease this message continually.\n\n        .. note::\n            This method is called by the constructor, and you should never\n            need to call it manually.\n        \"\"\"\n        self._request_queue.put(\n            requests.LeaseRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size\n            )\n        )\n\n    def modify_ack_deadline(self, seconds):\n        \"\"\"Resets the deadline for acknowledgement.\n\n        New deadline will be the given value of seconds from now.\n\n        The default implementation handles this for you; you should not need\n        to manually deal with setting ack deadlines. The exception case is\n        if you are implementing your own custom subclass of\n        :class:`~.pubsub_v1.subcriber._consumer.Consumer`.\n\n        Args:\n            seconds (int): The number of seconds to set the lease deadline\n                to. This should be between 0 and 600. Due to network latency,\n                values below 10 are advised against.\n        \"\"\"\n        self._request_queue.put(\n            requests.ModAckRequest(\n                ack_id=self._ack_id,\n                seconds=seconds\n            )\n        )\n\n    def nack(self):\n        \"\"\"Decline to acknowldge the given message.\n\n        This will cause the message to be re-delivered to the subscription.\n        \"\"\"\n        self._request_queue.put(\n            requests.NackRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size\n            )\n        )\n", "path": "pubsub/google/cloud/pubsub_v1/subscriber/message.py"}], "after_files": [{"content": "# Copyright 2017, Google LLC All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\n\nimport datetime\nimport json\nimport math\nimport time\n\nfrom google.api_core import datetime_helpers\nfrom google.cloud.pubsub_v1.subscriber._protocol import requests\n\n\n_MESSAGE_REPR = \"\"\"\\\nMessage {{\n  data: {!r}\n  attributes: {}\n}}\"\"\"\n\n\ndef _indent(lines, prefix='  '):\n    \"\"\"Indent some text.\n\n    Note that this is present as ``textwrap.indent``, but not in Python 2.\n\n    Args:\n        lines (str): The newline delimited string to be indented.\n        prefix (Optional[str]): The prefix to indent each line with. Default\n            to two spaces.\n\n    Returns:\n        str: The newly indented content.\n    \"\"\"\n    indented = []\n    for line in lines.split('\\n'):\n        indented.append(prefix + line)\n    return '\\n'.join(indented)\n\n\nclass Message(object):\n    \"\"\"A representation of a single Pub/Sub message.\n\n    The common way to interact with\n    :class:`~.pubsub_v1.subscriber.message.Message` objects is to receive\n    them in callbacks on subscriptions; most users should never have a need\n    to instantiate them by hand. (The exception to this is if you are\n    implementing a custom subclass to\n    :class:`~.pubsub_v1.subscriber._consumer.Consumer`.)\n\n    Attributes:\n        message_id (str): The message ID. In general, you should not need\n            to use this directly.\n        data (bytes): The data in the message. Note that this will be a\n            :class:`bytes`, not a text string.\n        attributes (.ScalarMapContainer): The attributes sent along with the\n            message. See :attr:`attributes` for more information on this type.\n        publish_time (datetime): The time that this message was originally\n            published.\n    \"\"\"\n\n    def __init__(self, message, ack_id, request_queue):\n        \"\"\"Construct the Message.\n\n        .. note::\n\n            This class should not be constructed directly; it is the\n            responsibility of :class:`BasePolicy` subclasses to do so.\n\n        Args:\n            message (~.pubsub_v1.types.PubsubMessage): The message received\n                from Pub/Sub.\n            ack_id (str): The ack_id received from Pub/Sub.\n            request_queue (queue.Queue): A queue provided by the policy that\n                can accept requests; the policy is responsible for handling\n                those requests.\n        \"\"\"\n        self._message = message\n        self._ack_id = ack_id\n        self._request_queue = request_queue\n        self.message_id = message.message_id\n\n        # The instantiation time is the time that this message\n        # was received. Tracking this provides us a way to be smart about\n        # the default lease deadline.\n        self._received_timestamp = time.time()\n\n        # The policy should lease this message, telling PubSub that it has\n        # it until it is acked or otherwise dropped.\n        self.lease()\n\n    def __repr__(self):\n        # Get an abbreviated version of the data.\n        abbv_data = self._message.data\n        if len(abbv_data) > 50:\n            abbv_data = abbv_data[:50] + b'...'\n\n        pretty_attrs = json.dumps(\n            dict(self.attributes),\n            indent=2,\n            separators=(',', ': '),\n            sort_keys=True,\n        )\n        pretty_attrs = _indent(pretty_attrs)\n        # We don't actually want the first line indented.\n        pretty_attrs = pretty_attrs.lstrip()\n        return _MESSAGE_REPR.format(abbv_data, pretty_attrs)\n\n    @property\n    def attributes(self):\n        \"\"\"Return the attributes of the underlying Pub/Sub Message.\n\n        .. warning::\n\n            A ``ScalarMapContainer`` behaves slightly differently than a\n            ``dict``. For a Pub / Sub message this is a ``string->string`` map.\n            When trying to access a value via ``map['key']``, if the key is\n            not in the map, then the default value for the string type will\n            be returned, which is an empty string. It may be more intuitive\n            to just cast the map to a ``dict`` or to one use ``map.get``.\n\n        Returns:\n            .ScalarMapContainer: The message's attributes. This is a\n            ``dict``-like object provided by ``google.protobuf``.\n        \"\"\"\n        return self._message.attributes\n\n    @property\n    def data(self):\n        \"\"\"Return the data for the underlying Pub/Sub Message.\n\n        Returns:\n            bytes: The message data. This is always a bytestring; if you\n                want a text string, call :meth:`bytes.decode`.\n        \"\"\"\n        return self._message.data\n\n    @property\n    def publish_time(self):\n        \"\"\"Return the time that the message was originally published.\n\n        Returns:\n            datetime: The date and time that the message was published.\n        \"\"\"\n        timestamp = self._message.publish_time\n        delta = datetime.timedelta(\n            seconds=timestamp.seconds,\n            microseconds=timestamp.nanos // 1000)\n        return datetime_helpers._UTC_EPOCH + delta\n\n    @property\n    def size(self):\n        \"\"\"Return the size of the underlying message, in bytes.\"\"\"\n        return self._message.ByteSize()\n\n    def ack(self):\n        \"\"\"Acknowledge the given message.\n\n        Acknowledging a message in Pub/Sub means that you are done\n        with it, and it will not be delivered to this subscription again.\n        You should avoid acknowledging messages until you have\n        *finished* processing them, so that in the event of a failure,\n        you receive the message again.\n\n        .. warning::\n            Acks in Pub/Sub are best effort. You should always\n            ensure that your processing code is idempotent, as you may\n            receive any given message more than once.\n        \"\"\"\n        time_to_ack = math.ceil(time.time() - self._received_timestamp)\n        self._request_queue.put(\n            requests.AckRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size,\n                time_to_ack=time_to_ack\n            )\n        )\n\n    def drop(self):\n        \"\"\"Release the message from lease management.\n\n        This informs the policy to no longer hold on to the lease for this\n        message. Pub/Sub will re-deliver the message if it is not acknowledged\n        before the existing lease expires.\n\n        .. warning::\n            For most use cases, the only reason to drop a message from\n            lease management is on :meth:`ack` or :meth:`nack`; these methods\n            both call this one. You probably do not want to call this method\n            directly.\n        \"\"\"\n        self._request_queue.put(\n            requests.DropRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size\n            )\n        )\n\n    def lease(self):\n        \"\"\"Inform the policy to lease this message continually.\n\n        .. note::\n            This method is called by the constructor, and you should never\n            need to call it manually.\n        \"\"\"\n        self._request_queue.put(\n            requests.LeaseRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size\n            )\n        )\n\n    def modify_ack_deadline(self, seconds):\n        \"\"\"Resets the deadline for acknowledgement.\n\n        New deadline will be the given value of seconds from now.\n\n        The default implementation handles this for you; you should not need\n        to manually deal with setting ack deadlines. The exception case is\n        if you are implementing your own custom subclass of\n        :class:`~.pubsub_v1.subcriber._consumer.Consumer`.\n\n        Args:\n            seconds (int): The number of seconds to set the lease deadline\n                to. This should be between 0 and 600. Due to network latency,\n                values below 10 are advised against.\n        \"\"\"\n        self._request_queue.put(\n            requests.ModAckRequest(\n                ack_id=self._ack_id,\n                seconds=seconds\n            )\n        )\n\n    def nack(self):\n        \"\"\"Decline to acknowldge the given message.\n\n        This will cause the message to be re-delivered to the subscription.\n        \"\"\"\n        self._request_queue.put(\n            requests.NackRequest(\n                ack_id=self._ack_id,\n                byte_size=self.size\n            )\n        )\n", "path": "pubsub/google/cloud/pubsub_v1/subscriber/message.py"}]}