Introducing Enhanced Fan-Out support (#31)

* Introducing support for KCL 2.x

  Changes for EFO support

    * Introducing input objects for ruby
    * Introducing kclrb v2 module which supports interfaces from KCL 2.x

  The sample no longer uses an external file, but instead emits records to stderror.

  Updating all KCL dependencies to 2.1.1

Author: sahilpalvia

VERSION

@@ -1 +1 @@
-1.0.1
+2.0.0

aws-kclrb.gemspec

@@ -8,7 +8,7 @@ Gem::Specification.new do |spec|
   spec.files           = Dir['lib/**/*.rb'] 
   spec.files          += Dir['spec/**/*.rb']
   spec.files          += ['README.md', 'LICENSE.txt', 'VERSION', 'NOTICE.txt', '.yardopts', '.rspec']
-  spec.license         = 'Amazon Software License'
+  spec.licenses         = ['Amazon Software License']
   spec.platform        = Gem::Platform::RUBY
   spec.homepage        = 'http://github.com/aws/amazon-kinesis-client-ruby'
   spec.require_paths   = ['lib']

lib/aws/kclrb.rb

@@ -13,4 +13,5 @@
 #  permissions and limitations under the License.
 
 require 'aws/kclrb/record_processor'
+require 'aws/kclrb/messages'
 require 'aws/kclrb/kcl_process'

lib/aws/kclrb/kcl_process.rb

@@ -14,13 +14,16 @@
 
 require 'aws/kclrb/io_proxy'
 require 'aws/kclrb/checkpointer'
+require 'aws/kclrb/messages'
+require 'aws/kclrb/record_processor'
 
 module Aws
   module KCLrb
     # Error raised if the {KCLProcess} received an input action that it
     # could not parse or it could not handle.
-    class MalformedAction < RuntimeError; end
-  
+    class MalformedAction < RuntimeError;
+    end
+
     # Entry point for a KCL application in Ruby.
     #
     # Implementers of KCL applications in Ruby should instantiate this
@@ -31,12 +34,16 @@ class KCLProcess
       # @param input [IO] An `IO`-like object to read input lines from.
       # @param output [IO] An `IO`-like object to write output lines to.
       # @param error [IO] An `IO`-like object to write error lines to.
-      def initialize(processor, input=$stdin, output=$stdout, error=$stderr)
-        @processor = processor
+      def initialize(processor, input = $stdin, output = $stdout, error = $stderr)
+        if processor.version == 1
+          @processor = Aws::KCLrb::V2::V2ToV1Adapter.new(processor)
+        else
+          @processor = processor
+        end
         @io_proxy = IOProxy.new(input, output, error)
         @checkpointer = CheckpointerImpl.new(@io_proxy)
       end
-  
+
       # Starts this KCL processor's main loop.
       def run
         action = @io_proxy.read_action
@@ -45,9 +52,9 @@ def run
           action = @io_proxy.read_action
         end
       end
-  
+
       private
-      
+
       # @api private
       # Parses an input action and invokes the appropriate method of the
       # record processor.
@@ -63,21 +70,28 @@ def process_action(action)
         action_name = action.fetch('action')
         case action_name
         when 'initialize'
-          dispatch_to_processor(:init_processor, action.fetch('shardId'))
+          dispatch_to_processor(:init_processor,
+                                Aws::KCLrb::V2::InitializeInput.new(action.fetch('shardId'),
+                                                                    action.fetch('sequenceNumber')))
         when 'processRecords'
-          dispatch_to_processor(:process_records, action.fetch('records'), @checkpointer)
-        when 'shutdown'
-          dispatch_to_processor(:shutdown, @checkpointer, action.fetch('reason'))
+          dispatch_to_processor(:process_records,
+                                Aws::KCLrb::V2::ProcessRecordsInput.new(action.fetch('records'),
+                                                                        action.fetch('millisBehindLatest'),
+                                                                        @checkpointer))
+        when 'leaseLost'
+          dispatch_to_processor(:lease_lost, Aws::KCLrb::V2::LeaseLostInput.new)
+        when 'shardEnded'
+          dispatch_to_processor(:shard_ended, Aws::KCLrb::V2::ShardEndedInput.new(@checkpointer))
         when 'shutdownRequested'
-          dispatch_to_processor(:shutdown_requested, @checkpointer)
+          dispatch_to_processor(:shutdown_requested, Aws::KCLrb::V2::ShutdownRequestedInput.new(@checkpointer))
         else
           raise MalformedAction.new("Received an action which couldn't be understood. Action was '#{action}'")
         end
         @io_proxy.write_action('status', {'responseFor' => action_name})
       rescue KeyError => ke
         raise MalformedAction.new("Action '#{action}': #{ke.message}")
       end
-  
+
       # @api private
       # Calls the specified method on the record processor, and handles
       # any resulting exceptions by writing to the error stream.
@@ -91,7 +105,7 @@ def dispatch_to_processor(method, *args)
         #   of issue.
         @io_proxy.write_error(processor_error)
       end
-  
+
     end
   end
 end

lib/aws/kclrb/messages.rb

@@ -0,0 +1,53 @@
+module Aws
+  module KCLrb
+    module V2
+      # @abstract
+      # Input object for RecordProcessorBase#init_processor method.
+      class InitializeInput
+        attr_reader :shard_id, :sequence_number
+
+        def initialize(shard_id, sequence_number)
+          @shard_id = shard_id
+          @sequence_number = sequence_number
+        end
+      end
+
+      # @abstract
+      # Input object for RecordProcessorBase#process_records method.
+      class ProcessRecordsInput
+        attr_reader :records, :millis_behind_latest, :checkpointer
+
+        def initialize(records, millis_behind_latest, checkpointer = nil)
+          @records = records
+          @millis_behind_latest = millis_behind_latest
+          @checkpointer = checkpointer
+        end
+      end
+
+      # @abstract
+      # Input object for RecordProcessorBase#lease_lost method.
+      class LeaseLostInput
+      end
+
+      # @abstract
+      # Input object forRecordProcessorBase#shard_ended method.
+      class ShardEndedInput
+        attr_reader :checkpointer
+
+        def initialize(checkpointer = nil)
+          @checkpointer = checkpointer
+        end
+      end
+
+      # @abstract
+      # Input object for RecordProcessorBase#shutdown_requested method.
+      class ShutdownRequestedInput
+        attr_reader :checkpointer
+
+        def initialize(checkpointer = nil)
+          @checkpointer = checkpointer
+        end
+      end
+    end
+  end
+end

lib/aws/kclrb/record_processor.rb

@@ -1,5 +1,5 @@
 #
-#  Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # 
 #  Licensed under the Amazon Software License (the "License").
 #  You may not use this file except in compliance with the License.
@@ -32,7 +32,7 @@ class RecordProcessorBase
       def init_processor(shard_id)
         fail NotImplementedError.new
       end
-  
+
       # @abstract
       # Called by a KCLProcess with a list of records to be processed and a checkpointer
       # which accepts sequence numbers from the records to indicate where in the stream
@@ -52,7 +52,7 @@ def init_processor(shard_id)
       def process_records(records, checkpointer)
         fail NotImplementedError.new
       end
-  
+
       # @abstract
       # Called by a KCLProcess instance to indicate that this record processor
       # should shutdown. After this is called, there will be no more calls to
@@ -81,6 +81,109 @@ def shutdown(checkpointer, reason)
       #   number or no parameters.
       def shutdown_requested(checkpointer)
       end
+
+      def version
+        1
+      end
+    end
+
+    module V2
+      # @abstract
+      # Base class for implementing a record processor.
+      #
+      # A `RecordProcessor` processes a shard in a stream. See {https://github.com/awslabs/amazon-kinesis-client/blob/master/src/main/java/com/amazonaws/services/kinesis/clientlibrary/interfaces/IRecordProcessor.java the corresponding KCL interface}.
+      # Its methods will be called as follows:
+      #
+      # 1. {#init_processor} will be called once
+      # 2. {#process_records} will be called zero or more times
+      # 3. {#lease_lost} will be called zero to one time
+      # 4. {#shard_ended} will be called zero or more times
+      # 5. {#shutdown_requested} will be called zero to one time
+      class RecordProcessorBase
+        # @abstract
+        # Called once by a KCLProcess before any calls to process_records.
+        #
+        # @param initialize_input [InitializeInput] Initialize processor input
+        #   object
+        def init_processor(initialize_input)
+          fail NotImplementedError.new
+        end
+
+        # @abstract
+        # Called by a KCLProcess with a list of records to be processed and a
+        # checkpointer which accepts sequence numbers from the records to
+        # indicate where in the stream to checkpoint.
+        #
+        # @param record_processor_input [RecordProcessorInput] Process records
+        #   input object
+        def process_records(process_records_input)
+          fail NotImplementedError.new
+        end
+
+        # @abstract
+        # Called by a KCLProcess instance to indicate that this record processor
+        # should shutdown. After this is called, there will be no more calls to
+        # any other methods of this record processor.
+        #
+        # @param lease_lost_input [LeaseLostInput] Lease lost input object
+        #
+        #    - Clients should not checkpoint because there is possibly another
+        #      record processor which has acquired the lease for this shard.
+        def lease_lost(lease_lost_input)
+          fail NotImplementedError.new
+        end
+
+        # @abstract
+        # Called by a KCLProcess instance to indicate that this record processor
+        # should shutdown. After this is called, there will be no more calls to
+        # any other methods of this record processor.
+        #
+        # @param shard_ended_input [ShardEndedInput] Shard ended input object
+        #
+        #   - Clients need to checkpoint at this time.
+        def shard_ended(shard_ended_input)
+          fail NotImplementedError.new
+        end
+
+        # @abstract
+        # Called by a KCLProcess instance to indicate that this record processor
+        # is requesting a shutdown. This method should be overriden if required.
+        #
+        # @param shutdown_requested_input [ShutdownRequestedInput]
+        def shutdown_requested(shutdown_requested_input)
+        end
+
+        def version
+          2
+        end
+      end
+
+      class V2ToV1Adapter < Aws::KCLrb::V2::RecordProcessorBase
+        def initialize(processor)
+          @processor = processor
+        end
+
+        def init_processor(initialize_input)
+          @processor.init_processor(initialize_input.shard_id)
+        end
+
+        def process_records(process_records_input)
+          @processor.process_records(process_records_input.records,
+                                     process_records_input.checkpointer)
+        end
+
+        def lease_lost(lease_lost_input)
+          @processor.shutdown(nil, 'ZOMBIE')
+        end
+
+        def shard_ended(shard_ended_input)
+          @processor.shutdown(shard_ended_input.checkpointer, 'TERMINATE')
+        end
+
+        def shutdown_requested(shutdown_requested_input)
+          @processor.shutdown_requested(shutdown_requested_input.checkpointer)
+        end
+      end
     end
   end
 end