Add QualifiedOffset object to store shard data (#38)

* Introduce explicit ack methods

* Add QualifiedOffset object to store shard data

* cr

* docstring fix

* minor

nuclio_sdk/__init__.py

@@ -18,3 +18,4 @@
 from nuclio_sdk.platform import Platform
 from nuclio_sdk.response import Response
 from nuclio_sdk.exceptions import ExceptionWithResponse
+from nuclio_sdk.qualified_offset import QualifiedOffset

nuclio_sdk/platform.py

@@ -36,14 +36,14 @@ def __init__(
 
         self._control_callback = on_control_callback
 
-    async def explicit_ack(self, event):
+    async def explicit_ack(self, qualified_offset):
         """
-        Notifying the processor to ack a stream message
+        Notifying the processor to ack on a qualified offset
 
-        :param event
-        :type event
+        :param qualified_offset: the qualified offset to ack
+        :type qualified_offset: QualifiedOffset
         """
-        message = event.compile_explicit_ack_message()
+        message = qualified_offset.compile_explicit_ack_message()
         if self._control_callback:
             await self._control_callback(message)
         else:

nuclio_sdk/qualified_offset.py

@@ -0,0 +1,31 @@
+# Copyright 2017 The Nuclio Authors.
+#
+# 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.
+
+
+class QualifiedOffset(object):
+    def __init__(self, topic, partition, offset):
+        self.topic = topic
+        self.partition = partition
+        self.offset = offset
+
+    @staticmethod
+    def from_event(event):
+        return QualifiedOffset(event.topic, event.partition, event.offset)
+
+    def compile_explicit_ack_message(self):
+        return {
+            "topic": self.topic,
+            "partition": self.partition,
+            "offset": self.offset,
+        }