Before you start

Please familiarize yourself with how Amazon Kinesis works by looking through its online documentations, in particular its Key Concepts and Limitations.

Getting Started

Download and install the library from Nuget here.

This library enables you to create a client application which consumes and processes records that have been pushed to an Amazon Kinesis stream by taking care of most of the plumbing involved.

To process incoming records, you need to provide an implementation for the IRecordProcessor interface which has the following methods:

To start, you can create a client application by calling the static method ReactoKinesixApp.CreateNew which returns a running instance of IReactoKinesixApp that will start processing records from the stream straight away.

You will notice that the ReactoKinesixApp.CreateNew requires an instance of IRecordProcessorFactory rather than an instance of IRecordProcessor. The rationale for this decision is that it enables

Tracking the state of your client application

To enable us to track the state of your client application (e.g. what shards are we processing and where in the stream of records did we get to (a checkpoint so that we can easily return to where we stopped at at a later time), the client application uses a Amazon DynamoDB table (such as the following) to store the necessary state information for each client application.

Assigning Worker IDs

Each node (e.g. EC2 instance running the client application) that is processing records from a stream should be given a unique worker ID to identify itself. If you're running your client application within Amazon EC2, then Instance ID is a perfect choice to act as a meaningful worker ID.

F# Example

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19: 
20: 
21: 
22: 
23: 
24: 
25: 
26: 
27: 
28: 
29: 
30: 
31: 
32: 
33: 
34: 
35: 
36: 
37: 
38: 
39: 
40: 
41: 
42: 
43: 
44: 
45: 
46: 
47: 

#r "AWSSDK.dll"
#r "ReactoKinesix.dll"

open Amazon
open ReactoKinesix
open ReactoKinesix.Model

let awsKey      = "AKIAI5Y767DTOFBUSYAA"
let awsSecret   = "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"
let region      = RegionEndpoint.USEast1
let appName     = "TestApp"
let streamName  = "TestStream"
let workerId    = "LocalHost"

let act (record : Record) =
    let msg = System.Text.Encoding.UTF8.GetString(record.Data)
    printfn "You got message [%s] : %s" record.SequenceNumber msg  

let processBatch (records : Record[]) = 
    records |> Array.iter act
    { Status = Success; Checkpoint = true }

let maxRetryExceeded (records : Record[]) =
    printfn "Batch [count:%d] failed." records.Length
    
let gen () = 
    { new IRecordProcessor with 
        member this.Process(shardId, records) = processBatch records
        member this.OnMaxRetryExceeded(records, _) = maxRetryExceeded records
        member this.Dispose() = printfn "Processor disposed" }

let factory = 
    { new IRecordProcessorFactory with
        member this.CreateNew _ = gen() }

printfn "Starting client application..."
let app = 
    ReactoKinesixApp.CreateNew
        (awsKey, awsSecret, region, appName, streamName, workerId, factory)

app.OnInitialized.Add(fun _ -> printfn "Client application started")
app.OnBatchProcessed.Add(fun _ -> printfn "Another batch processed...")

printfn "Press any key to quit..."
System.Console.ReadKey() |> ignore

app.Dispose()

• ReactoKinesix
• Home page
• Get Library via NuGet
• Source Code on GitHub
• License
• Release Notes
• Tutorials
• C# example
• F# example
• Documentation
• Getting Started
• Features
• Error Handling
• Distributed Processing
• API Reference