Add shutdown requested implemention in python client.

amazon_kclpy/dispatch.py

@@ -25,7 +25,8 @@ class MalformedAction(Exception):
     "processRecords": messages.ProcessRecordsInput,
     "shutdown": messages.ShutdownInput,
     "checkpoint": messages.CheckpointInput,
-    "record": messages.Record
+    "record": messages.Record,
+    "shutdownRequested": messages.ShutdownRequestedInput
 }
 
 
@@ -56,4 +57,4 @@ def message_decode(json_dict):
         raise MalformedAction("Received an action which couldn't be understood. Action was '{action}' -- Allowed {keys}"
                               .format(action=action, keys=_format_serializer_names()))
 
-    return serializer(json_dict)
+    return serializer(json_dict)

amazon_kclpy/kcl.py

@@ -220,6 +220,17 @@ def shutdown(self, checkpointer, reason):
         '''
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def shutdownRequested(self, checkpointer):
+        '''
+                Called by a KCLProcess instance to indicate that this record processor is being shutdown.
+                And it gives an opportunity for record processor to checkpoint before shutdown.
+
+                :type checkpointer: amazon_kclpy.kcl.Checkpointer
+                :param checkpointer: A checkpointer which accepts a sequence number or no parameters.
+                '''
+        raise NotImplementedError
+
     version = 1
 
 

amazon_kclpy/messages.py

@@ -210,6 +210,38 @@ def dispatch(self, checkpointer, record_processor):
         self._checkpointer = checkpointer
         record_processor.shutdown(self)
 
+class ShutdownRequestedInput(MessageDispatcher):
+    """
+    Used to tell the record processor it will be shutdown.
+    """
+    def __init__(self, json_dict):
+        self._checkpointer = None
+        self._action = json_dict['action']
+
+    @property
+    def checkpointer(self):
+        """
+        The checkpointer that can be used to checkpoint before actual shutdown.
+
+        :return: the checkpointer
+        :rtype: amazon_kclpy.kcl.Checkpointer
+        """
+        return self._checkpointer
+
+    @property
+    def action(self):
+        """
+        The action that spawned this message
+
+        :return: the original action value
+        :rtype: str
+        """
+        return self._action
+
+    def dispatch(self, checkpointer, record_processor):
+        self._checkpointer = checkpointer
+        record_processor.shutdown_requested(self)
+
 
 class CheckpointInput(object):
     """