anchor.diff:      [<CLIEvent>] member this.OnRecordProcessed          = recordProcessedEvent.Publish
       [<CLIEvent>] member this.OnProcessErrored           = processErroredEvent.Publish
       [<CLIEvent>] member this.OnBatchProcessed           = batchProcessedEvent.Publish
  +    [<CLIEvent>] member this.OnCheckpoint               = checkpointEvent.Publish
       [<CLIEvent>] member this.OnHeartbeat                = heartbeatEvent.Publish
       [<CLIEvent>] member this.OnConditionalCheckFailed   = conditionalCheckFailedEvent.Publish
   
       interface IDisposable with
  -        member this.Dispose () =
  -            heartbeatSub.Dispose()
  -            fetchSub.Dispose()
  -            receivedSub.Dispose()
  -            initSub.Dispose()
  +        member this.Dispose () = 
  +            if System.Threading.Interlocked.CompareExchange(disposeInvoked, 1, 0) = 0 then
  +                cts.Cancel()
  +                heartbeatSub.Dispose()
  +                fetchSub.Dispose()
  +                receivedSub.Dispose()
  +                initSub.Dispose()
   
   type ReactoKinesixApp (awsKey     : string, 
                          awsSecret  : string, 
  
positive.code: open System
  open System.Reactive
  open System.Reactive.Linq
  
  open Amazon
  open Amazon.DynamoDBv2
  
negative.code:     
      // until we need to stop processing, keep fetching new records when we've finished processing the previous batch
      let fetchSub        = batchProcessedEvent.Publish
                              .Zip(checkpointEvent.Publish, fun batchProcessedRes _ -> batchProcessedRes)
                              .TakeUntil(stopProcessing)
                              .Subscribe(fun (_, iterator) -> fetchNextRecords iterator |> Async.StartImmediate)
  
  