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:                             streamName,
                              workerId,
                              shardId,
                              action     : Record -> unit) =
      let batchReceivedEvent          = new Event<Iterator * Record seq>()
      let recordProcessedEvent        = new Event<Record>()
      let processErroredEvent         = new Event<Record * Exception>()
      let batchProcessedEvent         = new Event<int * Iterator>()
  
      let heartbeatEvent              = new Event<unit>()
      let conditionalCheckFailedEvent = new Event<unit>()
  
      let catchCondCheck f = 
          try
              f()
          with
          | :? ConditionalCheckFailedException -> 
              conditionalCheckFailedEvent.Trigger()
              reraise()
  
      let updateHeartbeat _ = 
          (fun () -> DynamoDBUtils.updateHeartbeat dynamoDB tableName workerId shardId |> Async.RunSynchronously)
          |> catchCondCheck
  
      let updateCheckpoint seqNum =
          (fun () -> DynamoDBUtils.updateCheckpoint seqNum dynamoDB tableName workerId shardId |> Async.RunSynchronously)
          |> catchCondCheck
  
      let fetchNextRecords iterator = 
          async {
  
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)
  
  