aws-kinesis-client 0.3.0.2 → 0.4.0.0

raw patch · 16 files changed

+1393/−851 lines, 16 filesdep +enclosed-exceptionsdep +natsdep −eitherdep −errorsdep −hoist-errordep ~aws-kinesis-clientdep ~basedep ~monad-control

Dependencies added: enclosed-exceptions, nats

Dependencies removed: either, errors, hoist-error, stm-conduit

Dependency ranges changed: aws-kinesis-client, base, monad-control

Files

• CHANGELOG.md +37/−0
• aws-kinesis-client.cabal +33/−13
• cli/CLI.hs +41/−29
• cli/CLI/Options.hs +2/−2
• src/Aws/Kinesis/Client/Common.hs +25/−74
• src/Aws/Kinesis/Client/Consumer.hs +67/−261
• src/Aws/Kinesis/Client/Consumer/Internal.hs +171/−0
• src/Aws/Kinesis/Client/Consumer/Internal/Kit.hs +104/−0
• src/Aws/Kinesis/Client/Consumer/Internal/SavedStreamState.hs +67/−0
• src/Aws/Kinesis/Client/Consumer/Internal/ShardState.hs +82/−0
• src/Aws/Kinesis/Client/Consumer/Kit.hs +39/−0
• src/Aws/Kinesis/Client/Internal/Queue.hs +203/−0
• src/Aws/Kinesis/Client/Internal/Queue/Chunk.hs +69/−0
• src/Aws/Kinesis/Client/Producer.hs +144/−472
• src/Aws/Kinesis/Client/Producer/Internal.hs +245/−0
• src/Aws/Kinesis/Client/Producer/Kit.hs +64/−0

@@ -1,3 +1,40 @@+### v0.4.0.0++This release contains breaking changes.++- Replaced the internal use of `MonadError` with proper open-world exceptions.+  The exposed operations which may throw exceptions have been adjusted to+  return `m (Either e a)`; resolves [issue+  \#17](https://github.com/alephcloud/hs-aws-kinesis-client/issues/17).++- [Producer] Implemented a cleaner shutdown routine policy: previously, the+  producer would stay alive indefinitely until the thread on which it was+  created is killed. Now, the producer is shut down immediately once the scope of+  `withKinesisProducer` exits, at which point the queue will be flushed according+  to a configurable (and optional) timeout. Resolves [issue+  \#19](https://github.com/alephcloud/hs-aws-kinesis-client/issues/19).++- [Producer] Restructured the monolithic error sum types into many smaller+  types to better support the standard mode of use for error handling.++- [Consumer] Removed unused `ConsumerError` type.++- [Consumer] Changed batch size to use `Natural` type rather than `Int`.++- [Producer] Parameterized the implementation by an arbitrary queue structure;+  by default, the producer's message queue is now implemented using `TBMChan`+  to avoid a bug in `TQueue` which causes live-locking under very heavy load.++Other non-breaking changes:++- [Producer] A better, more reliable implementation of consuming the input+  queue in chunks.++- Removed dependencies on `either`, `errors`, `hoist-error`, `stm-conduit`;+  added dependency on `enclosed-exceptions`.++- Increased lower bound on `monad-control` to 1.0 (for hygienic purposes).+ ### v0.3.0.2  - Upgrade aws-general, aws-kinesis lower bounds.

@@ -1,12 +1,17 @@ name:                aws-kinesis-client-version:             0.3.0.2+version:             0.4.0.0 synopsis:            A producer & consumer client library for AWS Kinesis--- description:+description:+    This package provides a Producer client for bulk-writing messages to a+    Kinesis stream through a bounded queue, and a Consumer client for reading+    messages round-robin from all shards of a Kiensis stream.  Additionally, a+    rudimentary CLI interface to the Consumer is provided.+ license:             Apache-2.0 license-file:        LICENSE author:              Jon Sterling maintainer:          jon@jonmsterling.com-copyright:           Copyright (c) 2013-2014 PivotCloud, Inc.+copyright:           Copyright (c) 2013-2015 PivotCloud, Inc. category:            Cloud build-type:          Simple cabal-version:       >=1.10@@ -16,13 +21,26 @@   type:     git   location: https://github.com/alephcloud/hs-aws-kinesis-client.git +flag debug+  description: Enable debug logging+  default:     False+ library-  exposed-modules:     Aws.Kinesis.Client.Common,+  exposed-modules:     Aws.Kinesis.Client.Common                        Aws.Kinesis.Client.Consumer+                       Aws.Kinesis.Client.Consumer.Kit+                       Aws.Kinesis.Client.Consumer.Internal+                       Aws.Kinesis.Client.Consumer.Internal.ShardState+                       Aws.Kinesis.Client.Consumer.Internal.SavedStreamState+                       Aws.Kinesis.Client.Consumer.Internal.Kit                        Aws.Kinesis.Client.Producer+                       Aws.Kinesis.Client.Producer.Kit+                       Aws.Kinesis.Client.Producer.Internal+                       Aws.Kinesis.Client.Internal.Queue+                       Aws.Kinesis.Client.Internal.Queue.Chunk   -- other-modules:   -- other-extensions:-  build-depends:       base >=4.7 && <4.8,+  build-depends:       base >=4.7 && <4.9,                        base-unicode-symbols,                        aeson >=0.8,                        aws >=0.10.5,@@ -30,21 +48,20 @@                        conduit >=1.2.3.1,                        containers >=0.5.5.1,                        data-carousel >=0.1.0.0,-                       either >=4.3.2.1,-                       errors >=1.4.7,+                       enclosed-exceptions >=1.0.1,                        http-conduit >=2.1.5,                        kan-extensions >=4.2,                        lens >=4.7,                        lens-action >=0.1.0.1,                        lifted-async >=0.3.0,                        lifted-base >=0.2.3.3,-                       monad-control >=0.3.3.0,+                       monad-control >=1.0,                        mtl >=2.2.1,+                       nats >=1,                        random >=1.1,                        resourcet >=1.1.3.3,                        stm >=2.4.4,                        stm-chans >=3.0.0.2,-                       stm-conduit >=2.5.3,                        stm-queue-extras >=0.2.0.0,                        text >=1.2.0.3,                        transformers >=0.3.0.0,@@ -53,6 +70,9 @@   default-language:    Haskell2010   ghc-options:         -Wall +  if flag(debug)+    cpp-options:       -DDEBUG+ executable kinesis-cli     default-language: Haskell2010     hs-source-dirs: cli@@ -64,19 +84,19 @@                        aws >=0.10.5,                        aws-general >=0.2.1,                        aws-kinesis >=0.1.4,-                       aws-kinesis-client,+                       aws-kinesis-client >=0.4.0.0,                        conduit >=1.2.3.1,-                       either >=4.3.2.1,-                       hoist-error >=0.1.0.2,                        http-conduit >=2.1.5,                        kan-extensions >=4.2,                        lens >=4.7,                        lifted-async >=0.3.0,                        lifted-base >=0.2.3.3,-                       monad-control >=0.3.3.0,+                       monad-control >=1.0,                        mtl >=2.2.1,                        text >=1.2.0.3,                        transformers >=0.3.0.0,                        bytestring >=0.10.4,                        optparse-applicative >=0.11   ghc-options:         -Wall+  if flag(debug)+    cpp-options:       -DDEBUG

@@ -1,4 +1,4 @@--- Copyright (c) 2013-2014 PivotCloud, Inc.+-- Copyright (c) 2013-2015 PivotCloud, Inc. -- -- Aws.Kinesis.Client.Consumer --@@ -32,7 +32,7 @@  -- | -- Module: Main--- Copyright: Copyright © 2013-2014 PivotCloud, Inc.+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc. -- License: Apache-2.0 -- Maintainer: Jon Sterling <jsterling@alephcloud.com> -- Stability: experimental@@ -55,12 +55,9 @@ import Control.Monad import Control.Monad.Trans import Control.Monad.Trans.Control-import Control.Monad.Trans.Either import Control.Monad.Codensity import Control.Monad.Reader.Class import Control.Monad.Trans.Reader (ReaderT(..))-import Control.Monad.Error.Class-import Control.Monad.Error.Hoist import Control.Monad.Unicode  import qualified Data.Aeson as A@@ -89,7 +86,6 @@   = ( MonadReader CLIOptions m     , MonadIO m     , MonadBaseControl IO m-    , MonadError ConsumerError m     )  fetchCredentials@@ -99,7 +95,7 @@   view clioUseInstanceMetadata ≫= \case     True →       loadCredentialsFromInstanceMetadata-        <!?> KinesisError (toException NoInstanceMetadataCredentials)+        ≫= maybe (throwIO NoInstanceMetadataCredentials) return     False →       view clioAccessKeys ≫= \case         Just (CredentialsFromAccessKeys aks) →@@ -108,44 +104,60 @@             (aks ^. akSecretAccessKey)         Just (CredentialsFromFile path) →           loadCredentialsFromFile path credentialsDefaultKey-            <!?> KinesisError (toException MissingCredentials)+            ≫= maybe (throwIO MissingCredentials) return         Nothing →-          throwError $ KinesisError (toException MissingCredentials)+          throwIO MissingCredentials -app+managedKinesisKit   ∷ MonadCLI m-  ⇒ Codensity m ExitCode-app = do+  ⇒ Codensity m KinesisKit+managedKinesisKit = do   CLIOptions{..} ← ask   manager ← managedHttpManager   credentials ← lift fetchCredentials+  return KinesisKit+    { _kkManager = manager+    , _kkConfiguration = Configuration+         { timeInfo = Timestamp+         , credentials = credentials+         , logger = defaultLog Warning+         }+    , _kkKinesisConfiguration = defaultKinesisConfiguration _clioRegion+    }++cliConsumerKit+  ∷ MonadCLI m+  ⇒ KinesisKit+  → m ConsumerKit+cliConsumerKit kinesisKit = do+  CLIOptions{..} ← ask   savedStreamState ←     case _clioStateIn of       Just path → liftIO $ do         code ← catch (Just <$> BL8.readFile path) $ \case           exn             | isDoesNotExistError exn → return Nothing-            | otherwise → throw exn+            | otherwise → throwIO exn         traverse (either (fail ∘ ("Invalid saved state: " ⊕)) return) $           A.eitherDecode <$> code       Nothing → return Nothing+  return $+    makeConsumerKit kinesisKit _clioStreamName+      & ckBatchSize .~ 100+      & ckIteratorType .~ _clioIteratorType+      & ckSavedStreamState .~ savedStreamState -  consumer ← managedKinesisConsumer ConsumerKit-    { _ckKinesisKit = KinesisKit-        { _kkManager = manager-        , _kkConfiguration = Configuration-             { timeInfo = Timestamp-             , credentials = credentials-             , logger = defaultLog Warning-             }-        , _kkKinesisConfiguration = defaultKinesisConfiguration _clioRegion-        }-    , _ckStreamName = _clioStreamName-    , _ckBatchSize = 100-    , _ckIteratorType = _clioIteratorType-    , _ckSavedStreamState = savedStreamState-    }+app+  ∷ MonadCLI m+  ⇒ Codensity m ExitCode+app = do+  CLIOptions{..} ← ask +  consumer ←+    managedKinesisKit+      ≫= lift ∘ cliConsumerKit+      ≫= managedKinesisConsumer+   let     source = consumerSource consumer     step n r = succ n <$ liftIO (B8.putStrLn $ recordData r)@@ -180,7 +192,7 @@  main ∷ IO () main = do-  exitCode ← eitherT (fail ∘ show) return $+  exitCode ←     liftIO (execParser parserInfo)       ≫= runReaderT (lowerCodensity app)   exitWith exitCode

@@ -1,4 +1,4 @@--- Copyright (c) 2013-2014 PivotCloud, Inc.+-- Copyright (c) 2013-2015 PivotCloud, Inc. -- -- Aws.Kinesis.Client.Consumer --@@ -23,7 +23,7 @@  -- | -- Module: CLI.Options--- Copyright: Copyright © 2013-2014 PivotCloud, Inc.+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc. -- License: Apache-2.0 -- Maintainer: Jon Sterling <jsterling@alephcloud.com> -- Stability: experimental

@@ -1,4 +1,4 @@--- Copyright (c) 2013-2014 PivotCloud, Inc.+-- Copyright (c) 2013-2015 PivotCloud, Inc. -- -- Aws.Kinesis.Client.Common --@@ -25,7 +25,7 @@  -- | -- Module: Aws.Kinesis.Client.Common--- Copyright: Copyright © 2013-2014 PivotCloud, Inc.+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc. -- License: Apache-2.0 -- Maintainer: Jon Sterling <jsterling@alephcloud.com> -- Stability: experimental@@ -35,33 +35,21 @@ , kkConfiguration , kkKinesisConfiguration , kkManager-, MonadKinesis , runKinesis    -- * Fetching Shards , streamShardSource , streamOpenShardSource , shardIsOpen--  -- * Miscellaneous monad stuff-, mapError-, handleError-, mapEnvironment ) where  import qualified Aws import qualified Aws.Core as Aws import qualified Aws.Kinesis as Kin-import Control.Exception import Control.Lens-import Control.Error import Control.Monad-import Control.Monad.Reader.Class-import Control.Monad.Error.Class import Control.Monad.Trans-import Control.Monad.Trans.Reader import Control.Monad.Trans.Resource-import Control.Monad.Unicode import Data.Conduit import qualified Data.Conduit.List as CL import qualified Network.HTTP.Conduit as HC@@ -91,43 +79,34 @@ kkManager ∷ Lens' KinesisKit HC.Manager kkManager = lens _kkManager $ \kk mgr → kk { _kkManager = mgr } --- | The minimal effect modality for running Kinesis commands.----type MonadKinesis m-  = ( MonadIO m-    , MonadReader KinesisKit m-    , MonadError SomeException m-    )---- | Run a Kinesis request inside 'MonadKinesis'.+-- | Run a Kinesis request. -- runKinesis-  ∷ ( MonadKinesis m-    , Aws.ServiceConfiguration req ~ Kin.KinesisConfiguration+  ∷ ( Aws.ServiceConfiguration req ~ Kin.KinesisConfiguration     , Aws.Transaction req resp     )-  ⇒ req-  → m resp-runKinesis req = do-  KinesisKit{..} ← view id-  eitherT throwError return ∘ syncIO ∘ runResourceT $+  ⇒ KinesisKit+  → req+  → IO resp+runKinesis KinesisKit{..} =+  runResourceT ∘     Aws.pureAws       _kkConfiguration       _kkKinesisConfiguration       _kkManager-      req  shardIsOpen   ∷ Kin.Shard   → Bool shardIsOpen Kin.Shard{..} =-  isNothing $ shardSequenceNumberRange ^. _2+  has _Nothing $ shardSequenceNumberRange ^. _2  fetchShardsConduit-  ∷ MonadKinesis m-  ⇒ Kin.StreamName+  ∷ MonadIO m+  ⇒ KinesisKit+  → Kin.StreamName   → Conduit (Maybe Kin.ShardId) m Kin.Shard-fetchShardsConduit streamName =+fetchShardsConduit kit streamName =   awaitForever $ \mshardId → do     let req = Kin.DescribeStream           { Kin.describeStreamExclusiveStartShardId = mshardId@@ -135,7 +114,7 @@           , Kin.describeStreamStreamName = streamName           }     resp@(Kin.DescribeStreamResponse Kin.StreamDescription{..}) ←-      lift $ runKinesis req+      liftIO $ runKinesis kit req     yield `mapM_` streamDescriptionShards     void ∘ traverse (leftover ∘ Just) $       Kin.describeStreamExclusiveStartShardId =<<@@ -146,51 +125,23 @@ -- | A 'Source' of shards for a stream. -- streamShardSource-  ∷ MonadKinesis m-  ⇒ Kin.StreamName+  ∷ MonadIO m+  ⇒ KinesisKit+  → Kin.StreamName   → Source m Kin.Shard-streamShardSource streamName =-  CL.sourceList [Nothing] $= fetchShardsConduit streamName+streamShardSource kit streamName =+  CL.sourceList [Nothing] $= fetchShardsConduit kit streamName  -- | A 'Source' of open shards for a stream. -- streamOpenShardSource-  ∷ MonadKinesis m-  ⇒ Kin.StreamName+  ∷ MonadIO m+  ⇒ KinesisKit+  → Kin.StreamName   → Source m Kin.Shard-streamOpenShardSource streamName =-  flip mapOutputMaybe (streamShardSource streamName) $ \shard →+streamOpenShardSource kit streamName =+  flip mapOutputMaybe (streamShardSource kit streamName) $ \shard →     if shardIsOpen shard       then Just shard       else Nothing---- | This function may be used to transform a computation with errors in one--- type to one with errors in another type. Whilst the argument is fixed at--- 'EitherT', this may be used where the argument is in an /arbitrary/ monad,--- since it will simply get instantiated at 'EitherT'.----mapError-  ∷ MonadError e' m-  ⇒ (e → e')-  → EitherT e m a-  → m a-mapError e = eitherT (throwError ∘ e) return---- | This is a verion of 'catchError' with its arguments flipped.----handleError-  ∷ MonadError e m-  ⇒ (e → m α)-  → m α-  → m α-handleError = flip catchError---- | Analogous to 'withReader', but supports a result in 'MonadReader'.----mapEnvironment-  ∷ MonadReader r' m-  ⇒ Getter r' r-  → ReaderT r m a-  → m a-mapEnvironment l m = view l ≫= runReaderT m 

@@ -1,4 +1,4 @@--- Copyright (c) 2013-2014 PivotCloud, Inc.+-- Copyright (c) 2013-2015 PivotCloud, Inc. -- -- Aws.Kinesis.Client.Consumer --@@ -16,20 +16,14 @@ -- License for the specific language governing permissions and limitations -- under the License. -{-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE LambdaCase #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-}-{-# LANGUAGE TypeFamilies #-} {-# LANGUAGE UnicodeSyntax #-}  -- | -- Module: Aws.Kinesis.Client.Consumer--- Copyright: Copyright © 2013-2014 PivotCloud, Inc.+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc. -- License: Apache-2.0 -- Maintainer: Jon Sterling <jsterling@alephcloud.com> -- Stability: experimental@@ -48,164 +42,47 @@    -- * Consumer Environment , ConsumerKit(..)+, makeConsumerKit+, SavedStreamState++  -- ** Lenses , ckKinesisKit , ckStreamName , ckBatchSize-, ConsumerError(..)-, MonadConsumer-, SavedStreamState+, ckIteratorType+, ckSavedStreamState ) where -import qualified Aws.Kinesis as Kin-import Aws.Kinesis.Client.Common+import Aws.Kinesis+import Aws.Kinesis.Client.Consumer.Internal -import Control.Applicative-import Control.Applicative.Unicode import Control.Concurrent.Async.Lifted import Control.Concurrent.Lifted hiding (yield) import Control.Concurrent.STM-import Control.Concurrent.STM.Queue-import Control.Exception+import Control.Exception.Lifted import Control.Lens import Control.Lens.Action import Control.Monad.Codensity-import Control.Monad.Error.Class import Control.Monad.Reader import Control.Monad.Trans.Control import Control.Monad.Unicode-import qualified Data.Aeson as Æ import qualified Data.Carousel as CR import qualified Data.Map as M-import qualified Data.HashMap.Strict as HM import Data.Traversable (for) import Data.Conduit-import qualified Data.Conduit.List as CL import Prelude.Unicode --- | The internal representation for shards used by the consumer.----data ShardState-  = ShardState-  { _ssIterator ∷ !(TVar (Maybe Kin.ShardIterator))-  , _ssShardId ∷ !Kin.ShardId-  , _ssLastSequenceNumber ∷ !(TVar (Maybe Kin.SequenceNumber))-  }---- | A getter for '_ssIterator'.----ssIterator ∷ Getter ShardState (TVar (Maybe Kin.ShardIterator))-ssIterator = to _ssIterator---- | A lens for '_ssShardId'.----ssShardId ∷ Lens' ShardState Kin.ShardId-ssShardId = lens _ssShardId $ \ss sid → ss { _ssShardId = sid }---- | A getter for '_ssLastSequenceNumber'.----ssLastSequenceNumber ∷ Getter ShardState (TVar (Maybe Kin.SequenceNumber))-ssLastSequenceNumber = to _ssLastSequenceNumber---- | 'ShardState' is quotiented by shard ID.----instance Eq ShardState where-  ss == ss' = ss ^. ssShardId ≡ ss' ^. ssShardId--data ConsumerError-  = NoShards-  -- ^ A stream must always have at least one open shard.--  | KinesisError !SomeException-  -- ^ An error which derives from a request made to Kinesis.--  deriving Show---- | The 'ConsumerKit' contains what is needed to initialize a 'KinesisConsumer'.-data ConsumerKit-  = ConsumerKit-  { _ckKinesisKit ∷ !KinesisKit-  -- ^ The credentials and configuration for making requests to AWS Kinesis.--  , _ckStreamName ∷ !Kin.StreamName-  -- ^ The name of the stream to consume from.--  , _ckBatchSize ∷ {-# UNPACK #-} !Int-  -- ^ The number of records to fetch at once from the stream.--  , _ckIteratorType ∷ !Kin.ShardIteratorType-  -- ^ The type of iterator to consume.--  , _ckSavedStreamState ∷ !(Maybe SavedStreamState)-  -- ^ Optionally, an initial stream state. The iterator type in-  -- '_ckIteratorType' will be used for any shards not present in the saved-  -- stream state; otherwise, 'Kin.AfterSequenceNumber' will be used.-  }---- | A lens for '_ckKinesisKit'.----ckKinesisKit ∷ Lens' ConsumerKit KinesisKit-ckKinesisKit = lens _ckKinesisKit $ \ck kk → ck { _ckKinesisKit = kk }---- | A lens for '_ckStreamName'.----ckStreamName ∷ Lens' ConsumerKit Kin.StreamName-ckStreamName = lens _ckStreamName $ \ck sn → ck { _ckStreamName = sn }---- | A lens for '_ckBatchSize'.----ckBatchSize ∷ Lens' ConsumerKit Int-ckBatchSize = lens _ckBatchSize $ \ck bs → ck { _ckBatchSize = bs }---- | A lens for '_ckIteratorType'.----ckIteratorType ∷ Lens' ConsumerKit Kin.ShardIteratorType-ckIteratorType = lens _ckIteratorType $ \ck it → ck { _ckIteratorType = it }---- | A lens for '_ckSavedStreamState'.----ckSavedStreamState ∷ Lens' ConsumerKit (Maybe SavedStreamState)-ckSavedStreamState = lens _ckSavedStreamState $ \ck ss → ck { _ckSavedStreamState = ss }---type MessageQueueItem = (ShardState, Kin.Record)-type StreamState = CR.Carousel ShardState--newtype SavedStreamState-  = SavedStreamState-  { _savedStreamState ∷ M.Map Kin.ShardId Kin.SequenceNumber-  }---- | An iso for 'SavedStreamState'.----_SavedStreamState ∷ Iso' SavedStreamState (M.Map Kin.ShardId Kin.SequenceNumber)-_SavedStreamState = iso _savedStreamState SavedStreamState--instance Æ.ToJSON SavedStreamState where-  toJSON (SavedStreamState m) =-    Æ.Object ∘ HM.fromList ∘ flip fmap (M.toList m) $ \(sid, sn) →-      let Æ.String sid' = Æ.toJSON sid-      in sid' Æ..= sn--instance Æ.FromJSON SavedStreamState where-  parseJSON =-    Æ.withObject "SavedStreamState" $ \xs → do-      fmap (SavedStreamState ∘ M.fromList) ∘ for (HM.toList xs) $ \(sid, sn) → do-        pure (,)-          ⊛ Æ.parseJSON (Æ.String sid)-          ⊛ Æ.parseJSON sn-- -- | The 'KinesisConsumer' maintains state about which shards to pull from. -- data KinesisConsumer   = KinesisConsumer-  { _kcMessageQueue ∷ !(TBQueue MessageQueueItem)+  { _kcMessageQueue ∷ !MessageQueue   , _kcStreamState ∷ !(TVar StreamState)   }  -- | A getter for '_kcMessageQueue'. ---kcMessageQueue ∷ Getter KinesisConsumer (TBQueue MessageQueueItem)+kcMessageQueue ∷ Getter KinesisConsumer MessageQueue kcMessageQueue = to _kcMessageQueue  -- | A getter for '_kcStreamState'.@@ -213,25 +90,14 @@ kcStreamState ∷ Getter KinesisConsumer (TVar StreamState) kcStreamState = to _kcStreamState --- | The basic effect modality required for operating the consumer.----type MonadConsumer m-  = ( MonadIO m-    , MonadBaseControl IO m-    , MonadError ConsumerError m-    )--type MonadConsumerInternal m-  = ( MonadConsumer m-    , MonadReader ConsumerKit m-    )- -- | This constructs a 'KinesisConsumer' and closes it when you have done with -- it; this is equivalent to 'withKinesisConsumer', except that the -- continuation is replaced with returning the consumer in 'Codensity'. -- managedKinesisConsumer-  ∷ MonadConsumer m+  ∷ ( MonadIO m+    , MonadBaseControl IO m+    )   ⇒ ConsumerKit   → Codensity m KinesisConsumer managedKinesisConsumer kit =@@ -241,145 +107,80 @@ -- it. -- withKinesisConsumer-  ∷ MonadConsumer m+  ∷ ( MonadIO m+    , MonadBaseControl IO m+    )   ⇒ ConsumerKit   → (KinesisConsumer → m α)   → m α-withKinesisConsumer kit inner =-  flip runReaderT kit $ do-    batchSize ← view ckBatchSize-    messageQueue ← liftIO ∘ newTBQueueIO $ batchSize * 10+withKinesisConsumer kit inner = do+  let batchSize = kit ^. ckBatchSize+  messageQueue ← liftIO ∘ newTBQueueIO $ fromIntegral batchSize * 10 -    state ← updateStreamState CR.empty ≫= liftIO ∘ newTVarIO-    let reshardingLoop = forever $-          handleError (\_ → liftIO $ threadDelay 3000000) $ do-            liftIO (readTVarIO state)-              ≫= updateStreamState-              ≫= liftIO ∘ atomically ∘ writeTVar state-            liftIO $ threadDelay 10000000+  state ← liftIO $ updateStreamState kit CR.empty ≫= newTVarIO -        producerLoop = forever $-          handleError (\_ → liftIO $ threadDelay 2000000) $ do-            recordsCount ← replenishMessages messageQueue state-            liftIO ∘ threadDelay $-              case recordsCount of-                0 → 5000000-                _ → 70000+  let+    -- The "magic" constants used in the loops below are derived from weeks of+    -- optimizing the Consumer not to cause rate-limiting errors, whilst still+    -- supporting prompt retrieval of records. It is likely that further+    -- optimization is possible. +    reshardingLoop =+      forever ∘ handle (\(SomeException _) → threadDelay 3000000) $ do+        readTVarIO state+          ≫= updateStreamState kit+          ≫= atomically ∘ writeTVar state+        threadDelay 10000000 -    withAsync reshardingLoop $ \reshardingHandle → do-      link reshardingHandle-      withAsync producerLoop $ \producerHandle → do-        link producerHandle-        res ← lift ∘ inner $ KinesisConsumer messageQueue state-        return res+    producerLoop =+      forever ∘ handle (\(SomeException _) → threadDelay 2000000) $ do+        recordsCount ← replenishMessages kit messageQueue state --- | This requests new information from Kinesis and reconciles that with an--- existing carousel of shard states.----updateStreamState-  ∷ MonadConsumerInternal m-  ⇒ StreamState-  → m StreamState-updateStreamState state = do-  streamName ← view ckStreamName-  defaultIteratorType ← view ckIteratorType-  savedState ← view ckSavedStreamState+        threadDelay $+          case recordsCount of+            0 → 5000000+            _ → 70000 -  mapError KinesisError ∘ mapEnvironment ckKinesisKit $ do-    let existingShardIds = state ^. CR.clList <&> view ssShardId-        shardSource = flip mapOutputMaybe (streamOpenShardSource streamName) $ \sh@Kin.Shard{..} →-          if shardShardId `elem` existingShardIds-            then Nothing-            else Just sh+  withAsync (liftIO reshardingLoop) $ \reshardingHandle → do+    link reshardingHandle+    withAsync (liftIO producerLoop) $ \producerHandle → do+      link producerHandle+      res ← inner $ KinesisConsumer messageQueue state+      return res -    newShards ← shardSource $$ CL.consume-    shardStates ← forM newShards $ \Kin.Shard{..} → do-      let startingSequenceNumber =-            savedState ^? _Just ∘ _SavedStreamState ∘ ix shardShardId-          iteratorType =-            maybe-              defaultIteratorType-              (const Kin.AfterSequenceNumber)-              startingSequenceNumber -      Kin.GetShardIteratorResponse it ← runKinesis Kin.GetShardIterator-        { Kin.getShardIteratorShardId = shardShardId-        , Kin.getShardIteratorShardIteratorType = iteratorType-        , Kin.getShardIteratorStartingSequenceNumber = startingSequenceNumber-        , Kin.getShardIteratorStreamName = streamName-        }-      liftIO $ do-        iteratorVar ← newTVarIO $ Just it-        sequenceNumberVar ← newTVarIO Nothing-        return ShardState-          { _ssIterator = iteratorVar-          , _ssShardId = shardShardId-          , _ssLastSequenceNumber = sequenceNumberVar-          }-    return ∘ CR.nub $ CR.append shardStates state---- | Waits for a message queue to be emptied and fills it up again.----replenishMessages-  ∷ MonadConsumerInternal m-  ⇒ TBQueue MessageQueueItem-  → TVar StreamState-  → m Int-replenishMessages messageQueue shardsVar = do-  bufferSize ← view ckBatchSize-  liftIO ∘ atomically ∘ awaitQueueEmpty $ messageQueue-  (shard, iterator) ← liftIO ∘ atomically $ do-    mshard ← shardsVar ^!? act readTVar ∘ CR.cursor-    shard ← maybe retry return mshard-    miterator ← shard ^! ssIterator ∘ act readTVar-    iterator ← maybe retry return miterator-    return (shard, iterator)--  Kin.GetRecordsResponse{..} ← mapError KinesisError ∘ mapEnvironment ckKinesisKit $ runKinesis Kin.GetRecords-    { getRecordsLimit = Just bufferSize-    , getRecordsShardIterator = iterator-    }--  liftIO ∘ atomically $ do-    writeTVar (shard ^. ssIterator) getRecordsResNextShardIterator-    forM_ getRecordsResRecords $ writeTBQueue messageQueue ∘ (shard ,)-    modifyTVar shardsVar CR.moveRight--  return $ length getRecordsResRecords- -- | Await and read a single record from the consumer. -- readConsumer-  ∷ MonadConsumer m+  ∷ MonadIO m   ⇒ KinesisConsumer-  → m Kin.Record+  → m Record readConsumer consumer =   liftIO ∘ atomically $ do     (ss, rec) ← consumer ^! kcMessageQueue ∘ act readTBQueue-    writeTVar (ss ^. ssLastSequenceNumber) ∘ Just $ Kin.recordSequenceNumber rec+    writeTVar (ss ^. ssLastSequenceNumber) ∘ Just $ recordSequenceNumber rec     return rec  -- | Try to read a single record from the consumer; if there is non queued up, -- then 'Nothing' will be returned. -- tryReadConsumer-  ∷ MonadConsumer m+  ∷ MonadIO m   ⇒ KinesisConsumer-  → m (Maybe Kin.Record)+  → m (Maybe Record) tryReadConsumer consumer =   liftIO ∘ atomically $ do     mitem ← consumer ^! kcMessageQueue ∘ act tryReadTBQueue     for mitem $ \(ss, rec) → do-      writeTVar (ss ^. ssLastSequenceNumber) ∘ Just $ Kin.recordSequenceNumber rec+      writeTVar (ss ^. ssLastSequenceNumber) ∘ Just $ recordSequenceNumber rec       return rec  -- | A conduit for getting records. -- consumerSource-  ∷ MonadConsumer m+  ∷ MonadIO m   ⇒ KinesisConsumer-  → Source m Kin.Record+  → Source m Record consumerSource consumer =   forever $     lift (readConsumer consumer)@@ -388,14 +189,19 @@ -- | Get the last read sequence number at each shard. -- consumerStreamState-  ∷ MonadConsumer m+  ∷ MonadIO m   ⇒ KinesisConsumer   → m SavedStreamState consumerStreamState consumer =   liftIO ∘ atomically $ do-    shards ← consumer ^! kcStreamState ∘ act readTVar ∘ CR.clList-    pairs ← for shards $ \ss →-      (ss ^. ssShardId,) <$>-        ss ^! ssLastSequenceNumber ∘ act readTVar-    return ∘ SavedStreamState ∘ M.fromList $ pairs ≫= \(sid, msn) →-      msn ^.. _Just ∘ to (sid,)+    shards ← consumer+      ^! kcStreamState+       ∘ act readTVar+       ∘ CR.clList+    pairs ← for shards $ \state → state+      ^! ssLastSequenceNumber+       ∘ act readTVar+       ∘ to (state ^. ssShardId,)+    return ∘ review _SavedStreamState ∘ M.fromList $+      pairs ≫= \(sid, msn) →+        msn ^.. _Just ∘ to (sid,)

@@ -0,0 +1,171 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Consumer.Internal+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE CPP #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Consumer.Internal+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Consumer.Internal+( -- * Types+  MessageQueueItem+, MessageQueue+, StreamState++  -- * Operations+, updateStreamState+, replenishMessages++  -- * Re-exports+, module Aws.Kinesis.Client.Consumer.Internal.Kit+, module Aws.Kinesis.Client.Consumer.Internal.ShardState+, module Aws.Kinesis.Client.Consumer.Internal.SavedStreamState+) where++import Aws.Kinesis+import Aws.Kinesis.Client.Common+import Aws.Kinesis.Client.Consumer.Internal.Kit+import Aws.Kinesis.Client.Consumer.Internal.ShardState+import Aws.Kinesis.Client.Consumer.Internal.SavedStreamState++import Control.Lens+import Control.Lens.Action+import Control.Concurrent.STM+import Control.Concurrent.STM.Queue+import Control.Monad+import Control.Monad.Trans+import qualified Data.Carousel as CR+import Data.Conduit+import qualified Data.Conduit.List as CondL+import Prelude.Unicode++#ifdef DEBUG+import Data.Monoid.Unicode+import System.IO+#else+#endif++type MessageQueueItem = (ShardState, Record)+type MessageQueue = TBQueue MessageQueueItem+type StreamState = CR.Carousel ShardState++-- | This requests new information from Kinesis and reconciles that with an+-- existing carousel of shard states.+--+updateStreamState+  ∷ ConsumerKit+  → StreamState+  → IO StreamState+updateStreamState ConsumerKit{..} state = do+  let+    existingShardIds = state ^. CR.clList <&> view ssShardId+    shardSource =+      flip mapOutputMaybe (streamOpenShardSource _ckKinesisKit _ckStreamName) $ \sh@Shard{..} →+        if shardShardId ∈ existingShardIds+          then Nothing+          else Just sh++  newShards ← shardSource $$ CondL.consume+  shardStates ← forM newShards $ \Shard{..} → do+    let+      startingSequenceNumber =+        _ckSavedStreamState ^? _Just ∘ _SavedStreamState ∘ ix shardShardId+      iteratorType =+        maybe+          _ckIteratorType+          (const AfterSequenceNumber)+          startingSequenceNumber++#ifdef DEBUG+    debugPrint stdout $ "Getting " ⊕ show iteratorType ⊕ " iterator for shard " ⊕ show shardShardId+#else+    return ()+#endif++    GetShardIteratorResponse it ← runKinesis _ckKinesisKit GetShardIterator+      { getShardIteratorShardId = shardShardId+      , getShardIteratorShardIteratorType = iteratorType+      , getShardIteratorStartingSequenceNumber = startingSequenceNumber+      , getShardIteratorStreamName = _ckStreamName+      }++    liftIO ∘ atomically $ do+      iteratorVar ← newTVar $ Just it+      sequenceNumberVar ← newTVar Nothing+      return $ makeShardState shardShardId iteratorVar sequenceNumberVar++  return ∘ CR.nub $ CR.append shardStates state++-- | Waits for a message queue to be emptied and fills it up again.+--+replenishMessages+  ∷ ConsumerKit+  → MessageQueue+  → TVar StreamState+  → IO Int+replenishMessages ConsumerKit{..} messageQueue shardsVar = do+  liftIO ∘ atomically ∘ awaitQueueEmpty $ messageQueue+  (shard, iterator) ← liftIO ∘ atomically $ do+    mshard ← shardsVar ^!? act readTVar ∘ CR.cursor+    shard ← maybe retry return mshard+    miterator ← shard ^! ssIterator ∘ act readTVar+    iterator ← maybe retry return miterator+    return (shard, iterator)++  GetRecordsResponse{..} ← runKinesis _ckKinesisKit GetRecords+    { getRecordsLimit = Just $ fromIntegral _ckBatchSize+    , getRecordsShardIterator = iterator+    }++#ifdef DEBUG+  debugPrint stdout $+    "Replenished shard "+      ⊕ show (shard ^. ssShardId)+      ⊕ " with "+      ⊕ show (length getRecordsResRecords)+      ⊕ " records"+#else+  return ()+#endif++  liftIO ∘ atomically $ do+    writeTVar (shard ^. ssIterator) getRecordsResNextShardIterator+    forM_ getRecordsResRecords $ writeTBQueue messageQueue ∘ (shard ,)+    modifyTVar shardsVar CR.moveRight++  return $ length getRecordsResRecords++#ifdef DEBUG+debugPrint+  ∷ MonadIO m+  ⇒ Handle+  → String+  → m ()+debugPrint h =+  liftIO+    ∘ hPutStrLn h+    ∘ ("[Kinesis Consumer] " ⊕)+#else+#endif

@@ -0,0 +1,104 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Consumer.Internal.Kit+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Consumer.Internal.Kit+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Consumer.Internal.Kit+( ConsumerKit(..)+, makeConsumerKit+, ckKinesisKit+, ckStreamName+, ckBatchSize+, ckIteratorType+, ckSavedStreamState+) where++import Aws.Kinesis+import Aws.Kinesis.Client.Common+import Aws.Kinesis.Client.Consumer.Internal.SavedStreamState+import Control.Lens+import Numeric.Natural++-- | The 'ConsumerKit' contains what is needed to initialize a 'KinesisConsumer'.+data ConsumerKit+  = ConsumerKit+  { _ckKinesisKit ∷ !KinesisKit+  -- ^ The credentials and configuration for making requests to AWS Kinesis.++  , _ckStreamName ∷ !StreamName+  -- ^ The name of the stream to consume from.++  , _ckBatchSize ∷ !Natural+  -- ^ The number of records to fetch at once from the stream.++  , _ckIteratorType ∷ !ShardIteratorType+  -- ^ The type of iterator to consume.++  , _ckSavedStreamState ∷ !(Maybe SavedStreamState)+  -- ^ Optionally, an initial stream state. The iterator type in+  -- '_ckIteratorType' will be used for any shards not present in the saved+  -- stream state; otherwise, 'AfterSequenceNumber' will be used.+  }++-- | Create a 'ConsumerKit' with default settings (using iterator type+-- 'Latest' and a batch size of @200@).+--+makeConsumerKit+  ∷ KinesisKit+  → StreamName+  → ConsumerKit+makeConsumerKit kinesisKit stream = ConsumerKit+  { _ckKinesisKit = kinesisKit+  , _ckStreamName = stream+  , _ckBatchSize = 200+  , _ckIteratorType = Latest+  , _ckSavedStreamState = Nothing+  }++-- | A lens for '_ckKinesisKit'.+--+ckKinesisKit ∷ Lens' ConsumerKit KinesisKit+ckKinesisKit = lens _ckKinesisKit $ \ck kk → ck { _ckKinesisKit = kk }++-- | A lens for '_ckStreamName'.+--+ckStreamName ∷ Lens' ConsumerKit StreamName+ckStreamName = lens _ckStreamName $ \ck sn → ck { _ckStreamName = sn }++-- | A lens for '_ckBatchSize'.+--+ckBatchSize ∷ Lens' ConsumerKit Natural+ckBatchSize = lens _ckBatchSize $ \ck bs → ck { _ckBatchSize = bs }++-- | A lens for '_ckIteratorType'.+--+ckIteratorType ∷ Lens' ConsumerKit ShardIteratorType+ckIteratorType = lens _ckIteratorType $ \ck it → ck { _ckIteratorType = it }++-- | A lens for '_ckSavedStreamState'.+--+ckSavedStreamState ∷ Lens' ConsumerKit (Maybe SavedStreamState)+ckSavedStreamState = lens _ckSavedStreamState $ \ck ss → ck { _ckSavedStreamState = ss }+

@@ -0,0 +1,67 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Consumer.Internal.SavedStreamState+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Consumer.Internal.SavedStreamState+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Consumer.Internal.SavedStreamState+( SavedStreamState+, _SavedStreamState+) where++import Aws.Kinesis+import Control.Applicative (pure)+import Control.Applicative.Unicode+import Control.Lens hiding ((.=))+import Data.Aeson+import qualified Data.Map as M+import qualified Data.HashMap.Strict as HM+import Data.Traversable+import Prelude.Unicode++newtype SavedStreamState+  = SavedStreamState+  { _savedStreamState ∷ M.Map ShardId SequenceNumber+  }++-- | An iso for 'SavedStreamState'.+--+_SavedStreamState ∷ Iso' SavedStreamState (M.Map ShardId SequenceNumber)+_SavedStreamState = iso _savedStreamState SavedStreamState++instance ToJSON SavedStreamState where+  toJSON (SavedStreamState m) =+    Object ∘ HM.fromList ∘ flip fmap (M.toList m) $ \(sid, sn) →+      let String sid' = toJSON sid+      in sid' .= sn++instance FromJSON SavedStreamState where+  parseJSON =+    withObject "SavedStreamState" $ \xs → do+      fmap (SavedStreamState ∘ M.fromList) ∘ for (HM.toList xs) $ \(sid, sn) → do+        pure (,)+          ⊛ parseJSON (String sid)+          ⊛ parseJSON sn++

@@ -0,0 +1,82 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Consumer.Internal.ShardState+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Consumer.Internal.ShardState+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Consumer.Internal.ShardState+( ShardState+, makeShardState+, ssIterator+, ssShardId+, ssLastSequenceNumber+) where++import Aws.Kinesis++import Control.Concurrent.STM.TVar+import Control.Lens+import Prelude.Unicode++-- | The internal representation for shards used by the consumer.+--+data ShardState+  = ShardState+  { _ssIterator ∷ !(TVar (Maybe ShardIterator))+  , _ssShardId ∷ !ShardId+  , _ssLastSequenceNumber ∷ !(TVar (Maybe SequenceNumber))+  }++makeShardState+  ∷ ShardId+  → TVar (Maybe ShardIterator)+  → TVar (Maybe SequenceNumber)+  → ShardState+makeShardState sid vit vlsn = do+  ShardState+    { _ssIterator = vit+    , _ssShardId = sid+    , _ssLastSequenceNumber = vlsn+    }++-- | A getter for '_ssIterator'.+--+ssIterator ∷ Getter ShardState (TVar (Maybe ShardIterator))+ssIterator = to _ssIterator++-- | A lens for '_ssShardId'.+--+ssShardId ∷ Lens' ShardState ShardId+ssShardId = lens _ssShardId $ \ss sid → ss { _ssShardId = sid }++-- | A getter for '_ssLastSequenceNumber'.+--+ssLastSequenceNumber ∷ Getter ShardState (TVar (Maybe SequenceNumber))+ssLastSequenceNumber = to _ssLastSequenceNumber++-- | 'ShardState' is quotiented by shard ID.+--+instance Eq ShardState where+  ss == ss' = ss ^. ssShardId ≡ ss' ^. ssShardId+

@@ -0,0 +1,39 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Consumer.Kit+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++-- |+-- Module: Aws.Kinesis.Client.Consumer.Kit+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Consumer.Kit+( ConsumerKit+, makeConsumerKit+, ckKinesisKit+, ckStreamName+, ckBatchSize+, ckIteratorType+, ckSavedStreamState+, SavedStreamState+, _SavedStreamState+) where++import Aws.Kinesis.Client.Consumer.Internal.Kit+import Aws.Kinesis.Client.Consumer.Internal.SavedStreamState

@@ -0,0 +1,203 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Internal.Queue+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Internal.Queue+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Internal.Queue+( BoundedCloseableQueue(..)+) where++import Control.Applicative+import Control.Concurrent.STM+import Control.Concurrent.STM.TBMQueue+import Control.Concurrent.STM.TBMChan+import Control.Monad.Unicode+import Numeric.Natural+import Prelude.Unicode++-- | A signature for bounded, closeable queues.+--+class BoundedCloseableQueue q α | q → α where+  -- Create a queue with size @n@.+  newQueue+    ∷ Natural -- ^ the size @n@ of the queue+    → IO q++  -- Close the queue.+  closeQueue+    ∷ q+    → IO ()++  -- | Returns 'False' if and only if the queue is closed. If the queue is full+  -- this function shall block.+  writeQueue+    ∷ q+    → α+    → IO Bool++  -- | Non-blocking version of 'writeQueue'. Returns 'Nothing' if the queue was+  -- full. Otherwise it returns 'Just True' if the value was successfully+  -- written and 'Just False' if the queue was closed.+  tryWriteQueue+    ∷ q+    → α+    → IO (Maybe Bool)++  -- | Returns 'Nothing' if and only if the queue is closed. If this queue is+  -- empty this function blocks.+  readQueue+    ∷ q+    → IO (Maybe α)++  -- | Take up to @n@ items from the queue with a timeout of @t@.+  takeQueueTimeout+    ∷ q+    → Natural -- ^ the number of items @n@ to take+    → Natural -- ^ the timeout @t@ in microseconds+    → IO [α]++  -- | Whether the queue is empty.+  isEmptyQueue+    ∷ q+    → IO Bool++  -- | Whether the queue is closed.+  isClosedQueue+    ∷ q+    → IO Bool++  -- | Whether the queue is empty and closed. The trivial default+  -- implementation may be overridden with one which provides transactional+  -- guarantees.+  isClosedAndEmptyQueue+    ∷ q+    → IO Bool+  isClosedAndEmptyQueue q =+    (&&) <$> isEmptyQueue q <*> isClosedQueue q++instance BoundedCloseableQueue (TBMQueue a) a where+  newQueue =+    newTBMQueueIO ∘ fromIntegral++  closeQueue =+    atomically ∘ closeTBMQueue++  writeQueue q a =+    atomically $ isClosedTBMQueue q ≫= \case+      True → return False+      False → True <$ writeTBMQueue q a++  tryWriteQueue q a =+    atomically $ tryWriteTBMQueue q a ≫= \case+      Nothing → return $ Just False+      Just False → return Nothing+      Just True → return $ Just True++  readQueue =+    atomically ∘ readTBMQueue++  takeQueueTimeout q n timeoutDelay = do+    timedOutVar ← registerDelay $ fromIntegral timeoutDelay++    let+      timeout =+        readTVar timedOutVar ≫= check++      readItems xs = do+        atomically (Left <$> timeout <|> Right <$> readTBMQueue q) ≫= \case+          Left _ → return xs+          Right Nothing → return xs+          Right (Just x) → go (x:xs)++      go xs+        | length xs ≥ fromIntegral n = return xs+        | otherwise = readItems xs++    go []++  isClosedQueue =+    atomically ∘ isClosedTBMQueue++  isEmptyQueue =+    atomically ∘ isEmptyTBMQueue++  isClosedAndEmptyQueue q =+    atomically $+      (&&) <$> isClosedTBMQueue q <*> isEmptyTBMQueue q++instance BoundedCloseableQueue (TBMChan a) a where+  newQueue =+    newTBMChanIO ∘ fromIntegral++  closeQueue =+    atomically ∘ closeTBMChan++  writeQueue q a =+    atomically $ isClosedTBMChan q ≫= \case+      True → return False+      False → True <$ writeTBMChan q a++  tryWriteQueue q a =+    atomically $ tryWriteTBMChan q a ≫= \case+      Nothing → return $ Just False+      Just False → return Nothing+      Just True → return $ Just True++  readQueue =+    atomically ∘ readTBMChan++  isClosedQueue =+    atomically ∘ isClosedTBMChan++  isEmptyQueue =+    atomically ∘ isEmptyTBMChan++  isClosedAndEmptyQueue q =+    atomically $+      (&&) <$> isClosedTBMChan q <*> isEmptyTBMChan q++  takeQueueTimeout q n timeoutDelay = do+    timedOutVar ← registerDelay $ fromIntegral timeoutDelay++    let+      timeout =+        readTVar timedOutVar ≫= check++      readItems xs = do+        atomically (Left <$> timeout <|> Right <$> readTBMChan q) ≫= \case+          Left _ → return xs+          Right Nothing → return xs+          Right (Just x) → go (x:xs)++      go xs+        | length xs ≥ fromIntegral n = return xs+        | otherwise = readItems xs++    go []+

@@ -0,0 +1,69 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Internal.Queue.Chunk+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Internal.Queue.Chunk+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Internal.Queue.Chunk+( ChunkingPolicy(..)+, chunkedSourceFromQueue+) where++import Aws.Kinesis.Client.Internal.Queue++import Control.Monad+import Control.Monad.Trans+import Data.Conduit+import Numeric.Natural++-- | A policy for chunking the contents of the message queue.+--+data ChunkingPolicy+  = ChunkingPolicy+  { _cpMaxChunkSize ∷ !Natural+  -- ^ The largest chunk size that is permitted.++  , _cpMinChunkingInterval ∷ !Natural+  -- ^ The time in microseconds after which a chunk should be committed, even+  -- if the maximum chunk size has not yet been reached.+  }++-- | A 'Source' that reads chunks off a bounded STM queue according some+-- 'ChunkingPolicy'.+--+chunkedSourceFromQueue+  ∷ BoundedCloseableQueue q α+  ⇒ ChunkingPolicy+  → q+  → Source IO [α]+chunkedSourceFromQueue cp@ChunkingPolicy{..} q = do+  shouldTerminate ← lift $ isClosedAndEmptyQueue q+  unless shouldTerminate $ do+    items ← lift $ takeQueueTimeout q _cpMaxChunkSize _cpMinChunkingInterval+    unless (null items) $ do+      yield items++    chunkedSourceFromQueue cp q+

@@ -1,4 +1,4 @@--- Copyright (c) 2013-2014 PivotCloud, Inc.+-- Copyright (c) 2013-2015 PivotCloud, Inc. -- -- Aws.Kinesis.Client.Producer --@@ -17,26 +17,17 @@ -- under the License.  {-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE GADTs #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiWayIf #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE TupleSections #-} {-# LANGUAGE UnicodeSyntax #-}  -- | -- Module: Aws.Kinesis.Client.Producer--- Copyright: Copyright © 2013-2014 PivotCloud, Inc.+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc. -- License: Apache-2.0 -- Maintainer: Jon Sterling <jsterling@alephcloud.com> -- Stability: experimental@@ -51,297 +42,81 @@ , writeProducer , Message -  -- * Producer Environment-, MonadProducer-, ProducerKit(..)-, pkKinesisKit-, pkStreamName-, pkBatchPolicy-, pkRetryPolicy-, pkMessageQueueBounds-, pkMaxConcurrency--, ProducerError(..)-, _KinesisError-, _MessageNotEnqueued-, _InvalidConcurrentConsumerCount-, _MessageTooLarge--, pattern MaxMessageSize--, BatchPolicy-, defaultBatchPolicy-, bpBatchSize-, bpEndpoint--, RetryPolicy-, defaultRetryPolicy-, rpRetryCount+, module Aws.Kinesis.Client.Producer.Kit -, RecordEndpoint(..)+-- * Exceptions+, WriteProducerException(..)+, ProducerCleanupTimedOut(..)+, ProducerWorkerDied(..) ) where  import qualified Aws.Kinesis as Kin-import qualified Aws.Kinesis.Commands.PutRecords as Kin import Aws.Kinesis.Client.Common+import Aws.Kinesis.Client.Producer.Kit+import Aws.Kinesis.Client.Producer.Internal+import Aws.Kinesis.Client.Internal.Queue+import Aws.Kinesis.Client.Internal.Queue.Chunk  import Control.Applicative import Control.Concurrent.Async.Lifted import Control.Concurrent.Lifted hiding (yield)-import Control.Concurrent.STM-import Control.Concurrent.STM.TBMQueue+import Control.Exception.Enclosed import Control.Exception.Lifted import Control.Lens import Control.Monad import Control.Monad.Codensity-import Control.Monad.Error.Class import Control.Monad.Reader import Control.Monad.Trans.Control-import Control.Monad.Trans.Either+import Control.Monad.Trans.Except import Data.Conduit-import Data.Conduit.TQueue-import qualified Data.Conduit.List as CL import Data.Maybe+import Data.Monoid.Unicode import qualified Data.Text as T import qualified Data.Text.Encoding as T import Data.Traversable import Data.Typeable+import Numeric.Natural import Prelude.Unicode import qualified System.Random as R import System.IO --- | There are two endpoints which may be used to send records to Kinesis.----data RecordEndpoint-  = PutRecordEndpoint-  -- ^ Use the @PutRecord@ endpoint, which sends records one at a time. -  | PutRecordsEndpoint-  -- ^ Use the @PutRecords@ endpoint, which sends records in batches.--  deriving (Eq, Show)---- | The maximum size in bytes of a message.----pattern MaxMessageSize = 51000---- | The producer batches records according to a user-specified policy.----data BatchPolicy-  = BatchPolicy-  { _bpBatchSize ∷ {-# UNPACK #-} !Int-  , _bpEndpoint ∷ !RecordEndpoint-  } deriving (Eq, Show)---- | The number of records to send in a single request. This is only used--- when the endpoint is set to 'PutRecordsEndpoint'.----bpBatchSize ∷ Lens' BatchPolicy Int-bpBatchSize = lens _bpBatchSize $ \bp bs → bp { _bpBatchSize = bs }---- | The endpoint to use when sending records to Kinesis.----bpEndpoint ∷ Lens' BatchPolicy RecordEndpoint-bpEndpoint = lens _bpEndpoint $ \bp ep → bp { _bpEndpoint = ep }---- | The default batching policy sends @200@ records per 'PutRecordsEndpoint'--- request.----defaultBatchPolicy ∷ BatchPolicy-defaultBatchPolicy = BatchPolicy-  { _bpBatchSize = 200-  , _bpEndpoint = PutRecordsEndpoint-  }---- | The producer will attempt to re-send records which failed according to a--- user-specified policy. This policy applies to failures which occur in the--- process of sending a message to Kinesis, not those which occur in the course--- of enqueuing a message.-data RetryPolicy-  = RetryPolicy-  { _rpRetryCount ∷ {-# UNPACK #-} !Int-  } deriving (Eq, Show)---- | The number of times to retry sending a message after it has first failed.----rpRetryCount ∷ Lens' RetryPolicy Int-rpRetryCount = lens _rpRetryCount $ \rp n → rp { _rpRetryCount = n }---- | The default retry policy will attempt @5@ retries for a message.----defaultRetryPolicy ∷ RetryPolicy-defaultRetryPolicy = RetryPolicy-  { _rpRetryCount = 5-  }--type Message = T.Text--data MessageQueueItem-  = MessageQueueItem-  { _mqiMessage ∷ !Message-  -- ^ The contents of the message--  , _mqiPartitionKey ∷ !Kin.PartitionKey-  -- ^ The partition key the message is destined for--  , _mqiRemainingAttempts ∷ !Int-  -- ^ The number of times remaining to try and publish this message-  } deriving (Eq, Show)--mqiMessage ∷ Lens' MessageQueueItem Message-mqiMessage = lens _mqiMessage $ \i m → i { _mqiMessage = m }--mqiPartitionKey ∷ Lens' MessageQueueItem Kin.PartitionKey-mqiPartitionKey = lens _mqiPartitionKey $ \i s → i { _mqiPartitionKey = s }--mqiRemainingAttempts ∷ Lens' MessageQueueItem Int-mqiRemainingAttempts = lens _mqiRemainingAttempts $ \i n → i { _mqiRemainingAttempts = n }--messageQueueItemIsEligible-  ∷ MessageQueueItem-  → Bool-messageQueueItemIsEligible =-  (≥ 1) ∘ _mqiRemainingAttempts---- | The basic input required to construct a Kinesis producer.----data ProducerKit-  = ProducerKit-  { _pkKinesisKit ∷ !KinesisKit-  -- ^ The basic information required to send requests to AWS Kinesis.--  , _pkStreamName ∷ !Kin.StreamName-  -- ^ The name of the stream to send records to.--  , _pkBatchPolicy ∷ !BatchPolicy-  -- ^ The record batching policy for the producer.--  , _pkRetryPolicy ∷ !RetryPolicy-  -- ^ The retry policy for the producer.--  , _pkMessageQueueBounds ∷ {-# UNPACK #-} !Int-  -- ^ The maximum number of records that may be enqueued at one time.--  , _pkMaxConcurrency ∷ {-# UNPACK #-} !Int-  -- ^ The number of requests to run concurrently (minimum: 1).-  }---- | A lens for '_pkKinesisKit'.----pkKinesisKit ∷ Lens' ProducerKit KinesisKit-pkKinesisKit = lens _pkKinesisKit $ \pk kk → pk { _pkKinesisKit = kk }---- | A lens for '_pkStreamName'.----pkStreamName ∷ Lens' ProducerKit Kin.StreamName-pkStreamName = lens _pkStreamName $ \pk sn → pk { _pkStreamName = sn }---- | A lens for '_pkBatchPolicy'.----pkBatchPolicy ∷ Lens' ProducerKit BatchPolicy-pkBatchPolicy = lens _pkBatchPolicy $ \pk bp → pk { _pkBatchPolicy = bp }---- | A lens for '_pkRetryPolicy'.----pkRetryPolicy ∷ Lens' ProducerKit RetryPolicy-pkRetryPolicy = lens _pkRetryPolicy $ \pk rp → pk { _pkRetryPolicy = rp }---- | A lens for '_pkMessageQueueBounds'.----pkMessageQueueBounds ∷ Lens' ProducerKit Int-pkMessageQueueBounds = lens _pkMessageQueueBounds $ \pk qb → pk { _pkMessageQueueBounds = qb }---- | A lens for '_pkMaxConcurrency'.----pkMaxConcurrency ∷ Lens' ProducerKit Int-pkMaxConcurrency = lens _pkMaxConcurrency $ \pk n → pk { _pkMaxConcurrency = n }- -- | The (abstract) Kinesis producer client. -- data KinesisProducer-  = KinesisProducer-  { _kpMessageQueue ∷ !(TBMQueue MessageQueueItem)+  = ∀ q. BoundedCloseableQueue q MessageQueueItem+  ⇒ KinesisProducer+  { _kpMessageQueue ∷ !q   , _kpRetryPolicy ∷ !RetryPolicy   } -kpMessageQueue ∷ Getter KinesisProducer (TBMQueue MessageQueueItem)-kpMessageQueue = to _kpMessageQueue--kpRetryPolicy ∷ Getter KinesisProducer RetryPolicy-kpRetryPolicy = to _kpRetryPolicy--data ProducerError-  = KinesisError !SomeException-  -- ^ Represents an error which occured as a result of a request to Kinesis.--  | MessageNotEnqueued Message-  -- ^ Thrown when a message could not be enqueued since the queue was full.-  -- This error must be handled at the call-site.-+data WriteProducerException+  = ProducerQueueClosed+    -- ^ Thrown when a message could not be enqueued since the queue was closed.+  | ProducerQueueFull+    -- ^ Thrown when a message could not be enqueued since the queue was full.   | MessageTooLarge-  -- ^ Thrown when the message was larger than the maximum message size ('MaxMessageSize').--  | InvalidConcurrentConsumerCount-  -- ^ Thrown when 'pkMaxConcurrency' is set with an invalid value.--  deriving (Typeable, Show)--instance Exception ProducerError---- | A prism for 'KinesisError'.----_KinesisError ∷ Prism' ProducerError SomeException-_KinesisError =-  prism KinesisError $ \case-    KinesisError e → Right e-    e → Left e---- | A prism for 'MessageNotEnqueued'.----_MessageNotEnqueued ∷ Prism' ProducerError Message-_MessageNotEnqueued =-  prism MessageNotEnqueued $ \case-    MessageNotEnqueued m → Right m-    e → Left e+    -- ^ Thrown when the message was larger than the maximum message size+    -- ('MaxMessageSize')+  deriving (Typeable, Show, Eq) --- | A prism for 'MessageTooLarge'.----_MessageTooLarge ∷ Prism' ProducerError ()-_MessageTooLarge =-  prism (const MessageTooLarge) $ \case-    MessageTooLarge → Right ()-    e → Left e+instance Exception WriteProducerException --- | A prism for 'InvalidConcurrentConsumerCount'.----_InvalidConcurrentConsumerCount ∷ Prism' ProducerError ()-_InvalidConcurrentConsumerCount =-  prism (const InvalidConcurrentConsumerCount) $ \case-    InvalidConcurrentConsumerCount → Right ()-    e → Left e+-- | Thrown when the producer's cleanup routine takes longer than the+-- configured timeout.+data ProducerCleanupTimedOut+  = ProducerCleanupTimedOut+  deriving (Typeable, Show, Eq) --- | The basic effect modality required to use the Kinesis producer.----type MonadProducer m-  = ( MonadIO m-    , MonadBaseControl IO m-    , MonadError ProducerError m-    )+instance Exception ProducerCleanupTimedOut -type MonadProducerInternal m-  = ( MonadProducer m-    , MonadReader ProducerKit m-    )+-- | Thrown when the producer's worker dies unexpectedly (this is fatal, and+-- should never happen).+data ProducerWorkerDied+  = ProducerWorkerDied (Maybe SomeException)+  deriving (Typeable, Show) --- | Lifts something in 'MonadKinesis' to 'MonadProducer'.----liftKinesis-  ∷ MonadProducerInternal m-  ⇒ EitherT SomeException (ReaderT KinesisKit m) α-  → m α-liftKinesis =-  mapEnvironment pkKinesisKit-    ∘ mapError KinesisError+instance Exception ProducerWorkerDied  -- | Generates a valid 'Kin.PartitionKey'. --@@ -354,138 +129,32 @@   Kin.partitionKey (T.pack name)     & either (error ∘ T.unpack) id --- | This will take up to @n@ items from a 'TBMQueue'.----takeTBMQueue-  ∷ Int-  → TBMQueue α-  → STM [α]-takeTBMQueue n q-  | n <= 0 = return []-  | otherwise = do-      res ← tryReadTBMQueue q-      case res of-        Just (Just x) → (x:) <$> takeTBMQueue (n - 1) q-        _ → return [] --- | A policy for chunking the contents of the message queue.----data ChunkingPolicy-  = ChunkingPolicy-  { _cpMaxChunkSize ∷ Int-  -- ^ The largest chunk size that is permitted.--  , _cpThrottlingDelay ∷ Int-  -- ^ The time after which a chunk should be committed, even if the maximum-  -- chunk size has not yet been reached.-  }---- | A 'Source' that reads chunks off a bounded STM queue according some--- 'ChunkingPolicy'.----chunkedSourceTBMQueue-  ∷ ( MonadIO m-    , MonadBaseControl IO m-    )-  ⇒ ChunkingPolicy-  → TBMQueue α-  → Source m [α]-chunkedSourceTBMQueue bp@ChunkingPolicy{..} q = do-  terminateNow ← liftIO ∘ atomically $ isClosedTBMQueue q-  unless terminateNow $ do-    items ← liftIO ∘ atomically $ takeTBMQueue _cpMaxChunkSize q-    unless (null items) $ do-      yield items--    when (length items < _cpMaxChunkSize) $-      threadDelay _cpThrottlingDelay--    chunkedSourceTBMQueue bp q---- | Transform a 'Source' into a chunked 'Source' according to some--- 'ChunkingPolicy'.----chunkSource-  ∷ ( MonadIO m-    , MonadBaseControl IO m-    )-  ⇒ ChunkingPolicy-  → Source m α-  → Source m [α]-chunkSource cp src = do-  queue ← liftIO $ newTBMQueueIO $ _cpMaxChunkSize cp-  worker ← lift ∘ async $ src $$+ sinkTBMQueue queue True-  addCleanup (\_ → cancel worker) $ do-    lift $ link worker-    chunkedSourceTBMQueue cp queue-  return ()---- | A conduit for concurently sending multiple records to Kinesis using the--- @PutRecord@ endpoint.----concurrentPutRecordSink-  ∷ MonadProducerInternal m-  ⇒ Sink [MessageQueueItem] m ()-concurrentPutRecordSink = do-  maxWorkerCount ← view pkMaxConcurrency-  awaitForever $ \messages → do-    lift ∘ flip (mapConcurrentlyN maxWorkerCount 100) messages $ \m → do-      CL.sourceList [m] $$ putRecordSink----- | A conduit for sending a record to Kinesis using the @PutRecord@ endpoint;--- this is a conduit in order to restore failed messages as leftovers.----putRecordSink-  ∷ MonadProducerInternal m-  ⇒ Sink MessageQueueItem m ()-putRecordSink = do-  streamName ← view pkStreamName--  awaitForever $ \item → do-    let handler e = do-          liftIO $ do-            hPutStrLn stderr $ "Kinesis producer client error (will wait 5s): " ++ show e-            threadDelay 5000000-          leftover $ item & mqiRemainingAttempts -~ 1--    when (messageQueueItemIsEligible item) $-      handleError handler $ do-        let partitionKey = item ^. mqiPartitionKey-        void ∘ lift ∘ liftKinesis $ runKinesis Kin.PutRecord-          { Kin.putRecordData = item ^. mqiMessage ∘ to T.encodeUtf8-          , Kin.putRecordExplicitHashKey = Nothing-          , Kin.putRecordPartitionKey = partitionKey-          , Kin.putRecordSequenceNumberForOrdering = Nothing-          , Kin.putRecordStreamName = streamName-          }- splitEvery-  ∷ Int+  ∷ Natural   → [α]   → [[α]] splitEvery _ [] = [] splitEvery n list = first : splitEvery n rest   where-    (first,rest) = splitAt n list+    (first,rest) = splitAt (fromIntegral n) list  -- | A conduit for sending records to Kinesis using the @PutRecords@ endpoint. -- This is a conduit in order to restore failed messages as leftovers. -- putRecordsSink-  ∷ MonadProducerInternal m-  ⇒ Sink [MessageQueueItem] m ()-putRecordsSink = do-  streamName ← view pkStreamName-  batchSize ← view $ pkBatchPolicy ∘ bpBatchSize-  maxWorkerCount ← view pkMaxConcurrency+  ∷ ProducerKit+  → Sink [MessageQueueItem] IO ()+putRecordsSink ProducerKit{..} = do+  let batchSize = _pkBatchPolicy ^. bpBatchSize+   awaitForever $ \messages → do     let batches = splitEvery batchSize messages-    leftovers ← lift ∘ flip (mapConcurrentlyN maxWorkerCount 100) batches $ \items → do+    leftovers ← lift ∘ flip (mapConcurrentlyN _pkMaxConcurrency 100) batches $ \items → do       case filter messageQueueItemIsEligible items of         [] → return []         eligibleItems → do-          handleError (\e → eligibleItems <$ liftIO (hPutStrLn stderr $ show e)) $ do+          handleAny (\(SomeException e) → eligibleItems <$ debugPrint stderr (show e)) $ do             requestEntries ← for eligibleItems $ \m → do               let partitionKey = m ^. mqiPartitionKey               return Kin.PutRecordsRequestEntry@@ -494,9 +163,15 @@                 , Kin.putRecordsRequestEntryPartitionKey = partitionKey                 } -            Kin.PutRecordsResponse{..} ←  liftKinesis $ runKinesis Kin.PutRecords+#ifdef DEBUG+            debugPrint stdout $ "will put " ⊕ show (length requestEntries) ⊕ " records"+#else+            return ()+#endif++            Kin.PutRecordsResponse{..} ← runKinesis _pkKinesisKit Kin.PutRecords               { Kin.putRecordsRecords = requestEntries-              , Kin.putRecordsStreamName = streamName+              , Kin.putRecordsStreamName = _pkStreamName               }             let               processResult m m'@@ -510,107 +185,106 @@           <&> mqiRemainingAttempts -~ 1            & filter messageQueueItemIsEligible -sendMessagesSink-  ∷ MonadProducerInternal m-  ⇒ Sink [MessageQueueItem] m ()-sendMessagesSink = do-  batchPolicy ← view pkBatchPolicy-  case batchPolicy ^. bpEndpoint of-    PutRecordsEndpoint → putRecordsSink-    PutRecordEndpoint → concurrentPutRecordSink- -- | Enqueues a message to Kinesis on the next shard. If a message cannot be--- enqueued (because the client has exceeded its queue size), the--- 'MessageNotEnqueued' exception will be thrown.+-- enqueued, an error of type 'WriteProducerException' will be returned. -- writeProducer-  ∷ MonadProducer m+  ∷ MonadIO m   ⇒ KinesisProducer   → Message-  → m ()-writeProducer producer !msg = do-  when (T.length msg > MaxMessageSize) $-    throwError MessageTooLarge--  gen ← liftIO R.newStdGen-  result ← liftIO ∘ atomically $ do-    tryWriteTBMQueue (producer ^. kpMessageQueue) MessageQueueItem-      { _mqiMessage = msg-      , _mqiPartitionKey = generatePartitionKey gen-      , _mqiRemainingAttempts = producer ^. kpRetryPolicy . rpRetryCount . to succ-      }-  case result of-    Just True → return ()-    _ → throwError $ MessageNotEnqueued msg+  → m (Either WriteProducerException ())+writeProducer KinesisProducer{..} !msg =+  runExceptT $ do+    when (T.length msg > MaxMessageSize) $+      throwE MessageTooLarge --- | This is a 'Source' that returns all the items presently in a queue: it--- terminates when the queue is empty.----exhaustTBMQueue-  ∷ TBMQueue α-  → Source STM α-exhaustTBMQueue q = do-  mx ← lift $ tryReadTBMQueue q-  case mx of-    Just (Just x) → do-      yield x-      exhaustTBMQueue q-    _ → return ()+    gen ← liftIO R.newStdGen+    result ← liftIO $+      tryWriteQueue _kpMessageQueue MessageQueueItem+        { _mqiMessage = msg+        , _mqiPartitionKey = generatePartitionKey gen+        , _mqiRemainingAttempts = _kpRetryPolicy ^. rpRetryCount ∘ to succ+        }+    case result of+      Just written → unless written $ throwE ProducerQueueFull+      Nothing → throwE $ ProducerQueueClosed  -- | This constructs a 'KinesisProducer' and closes it when you have done with -- it. This is equivalent to 'withKinesisProducer', but replaces the -- continuation with a return in 'Codensity'. -- managedKinesisProducer-  ∷ ∀ m-  . ( MonadIO m+  ∷ ( MonadIO m     , MonadBaseControl IO m-    , MonadError ProducerError m     )   ⇒ ProducerKit   → Codensity m KinesisProducer-managedKinesisProducer kit = do-  when (kit ^. pkMaxConcurrency < 1) ∘ lift $-    throwError InvalidConcurrentConsumerCount+managedKinesisProducer kit@ProducerKit{_pkQueueImplementation = QueueImplementation (_ ∷ proxy q)} = do+  messageQueue ← liftIO ∘ newQueue ∘ fromIntegral $ kit ^. pkMessageQueueBounds -  messageQueue ← liftIO ∘ newTBMQueueIO $ kit ^. pkMessageQueueBounds+  let+    producer = KinesisProducer+      { _kpMessageQueue = (messageQueue ∷ q)+      , _kpRetryPolicy = kit ^. pkRetryPolicy+      } -  let chunkingPolicy = ChunkingPolicy ((kit ^. pkBatchPolicy ∘ bpBatchSize) * (kit ^. pkMaxConcurrency)) 5000000-      -- TODO: this 'forever' is only here to restart if we get killed.-      -- Replace with proper error handling.-      consumerLoop ∷ m () = flip runReaderT kit ∘ forever $-        chunkSource chunkingPolicy (sourceTBMQueue messageQueue)-          $$ sendMessagesSink+    chunkingPolicy = ChunkingPolicy+      { _cpMaxChunkSize = (kit ^. pkBatchPolicy ∘ bpBatchSize) * (kit ^. pkMaxConcurrency)+      , _cpMinChunkingInterval = 5000000+      } -  let cleanupConsumer consumerHandle = do-        liftIO ∘ atomically $ closeTBMQueue messageQueue-        flip runReaderT kit $ do-          leftovers ← liftIO ∘ atomically $-            exhaustTBMQueue messageQueue-              $$ CL.consume-          chunkSource chunkingPolicy (CL.sourceList leftovers)-            $$ sendMessagesSink-        cancel consumerHandle+    processQueue =+      chunkedSourceFromQueue chunkingPolicy messageQueue+        $$ putRecordsSink kit -  consumerHandle ← managedBracket (async consumerLoop) cleanupConsumer+    -- TODO: figure out better error handling here (such as a limit to respawns)+    workerLoop ∷ IO () = do+      result ← tryAny processQueue+      case result of+        Left exn → do+          debugPrint stderr $ "Respawning worker loop after exception: " ⊕ show exn+          workerLoop+        Right () → return () -  Codensity $ \inner → do-    link consumerHandle-    res ← inner KinesisProducer-      { _kpMessageQueue = messageQueue-      , _kpRetryPolicy = kit ^. pkRetryPolicy-      }-    () ← wait consumerHandle-    return res+    cleanupWorker _ = do+      closeQueue messageQueue+#ifdef DEBUG+      debugPrint stdout "Closing queues, will clean up"+#else+      return ()+#endif +      withAsync processQueue $ \cleanupHandle → do+        case _pkCleanupTimeout kit of+          Just timeout →+            withAsync (threadDelay $ fromIntegral timeout) $ \timeoutHandle → do+              result ← waitEitherCatchCancel timeoutHandle cleanupHandle+              case result of+                Left _timeoutResult →+                  throwIO ProducerCleanupTimedOut+                Right workerResult →+                  throwIO ∘ ProducerWorkerDied $ workerResult ^? _Left+          Nothing →+            wait cleanupHandle ++  workerHandle ← Codensity $ bracket (async (liftIO workerLoop)) (liftIO ∘ cleanupWorker)++  Codensity $ \inner → do+    withAsync (inner producer) $ \innerHandle → do+      result ← waitEitherCatchCancel innerHandle workerHandle+      case result of+        Left innerResult →+          either throwIO return innerResult+        Right (workerResult ∷ Either SomeException ()) →+          throwIO ∘ ProducerWorkerDied $ workerResult ^? _Left+ -- | This constructs a 'KinesisProducer' and closes it when you have done with -- it. -- withKinesisProducer   ∷ ( MonadIO m     , MonadBaseControl IO m-    , MonadError ProducerError m     )   ⇒ ProducerKit   → (KinesisProducer → m α)@@ -618,33 +292,31 @@ withKinesisProducer =   runCodensity ∘ managedKinesisProducer -managedBracket-  ∷ MonadBaseControl IO m-  ⇒ m α-  → (α → m β)-  → Codensity m α-managedBracket action cleanup =-  Codensity $ bracket action cleanup- -- | map at most n actions concurrently -- mapConcurrentlyN-  ∷ ( MonadIO m-    , MonadBaseControl IO m-    , Traversable t-    )-  ⇒ Int -- ^ number of concurrent actions-  → Int -- ^ startup delay between actions in milliseconds-  → (a → m b)+  ∷ Traversable t+  ⇒ Natural -- ^ number of concurrent actions+  → Natural -- ^ startup delay between actions in milliseconds+  → (a → IO b)   → t a-  → m (t b)+  → IO (t b) mapConcurrentlyN n delay f t = do-  sem ← liftIO $ newQSem n+  sem ← liftIO ∘ newQSem $ fromIntegral n   mapConcurrently (run sem) t_     where       (_, t_) = mapAccumL (\i v → (succ i, (i,v))) 0 t       run sem (i,a) =         liftBaseOp_ (bracket_ (waitQSem sem) (signalQSem sem)) $ do-          liftIO ∘ threadDelay $ 1000 * delay * i+          liftIO ∘ threadDelay ∘ fromIntegral $ 1000 * delay * i           f a +debugPrint+  ∷ MonadIO m+  ⇒ Handle+  → String+  → m ()+debugPrint h =+  liftIO+    ∘ hPutStrLn h+    ∘ ("[Kinesis Producer] " ⊕)

@@ -0,0 +1,245 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Producer.Internal+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Producer.Internal+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Producer.Internal+( ProducerKit(..)+, makeProducerKit+, pkKinesisKit+, pkStreamName+, pkBatchPolicy+, pkRetryPolicy+, pkMessageQueueBounds+, pkMaxConcurrency+, pkCleanupTimeout+, pkQueueImplementation++  -- * Queue Implementations+, QueueImplementation(..)+, defaultQueueImplementation++  -- * Policies+, BatchPolicy+, defaultBatchPolicy+, bpBatchSize++, RetryPolicy+, defaultRetryPolicy+, rpRetryCount++  -- * Other Types+, Message+, MessageQueueItem(..)+, mqiMessage+, mqiRemainingAttempts+, mqiPartitionKey+, messageQueueItemIsEligible+) where++import Aws.Kinesis+import Aws.Kinesis.Client.Common+import Aws.Kinesis.Client.Internal.Queue++import Control.Concurrent.STM.TBMChan+import Control.Lens+import Data.Proxy+import qualified Data.Text as T+import Numeric.Natural+import Prelude.Unicode++type Message = T.Text++data MessageQueueItem+  = MessageQueueItem+  { _mqiMessage ∷ !Message+  -- ^ The contents of the message++  , _mqiPartitionKey ∷ !PartitionKey+  -- ^ The partition key the message is destined for++  , _mqiRemainingAttempts ∷ !Natural+  -- ^ The number of times remaining to try and publish this message+  } deriving (Eq, Show)++mqiMessage ∷ Lens' MessageQueueItem Message+mqiMessage = lens _mqiMessage $ \i m → i { _mqiMessage = m }++mqiPartitionKey ∷ Lens' MessageQueueItem PartitionKey+mqiPartitionKey = lens _mqiPartitionKey $ \i s → i { _mqiPartitionKey = s }++mqiRemainingAttempts ∷ Lens' MessageQueueItem Natural+mqiRemainingAttempts = lens _mqiRemainingAttempts $ \i n → i { _mqiRemainingAttempts = n }++messageQueueItemIsEligible+  ∷ MessageQueueItem+  → Bool+messageQueueItemIsEligible =+  (≥ 1) ∘ _mqiRemainingAttempts++-- | The producer batches records according to a user-specified policy.+--+data BatchPolicy+  = BatchPolicy+  { _bpBatchSize ∷ !Natural+  } deriving (Eq, Show)++-- | The number of records to send in a single request. This is only used+-- when the endpoint is set to 'PutRecordsEndpoint'.+--+bpBatchSize ∷ Lens' BatchPolicy Natural+bpBatchSize = lens _bpBatchSize $ \bp bs → bp { _bpBatchSize = bs }++-- | The default batching policy sends @200@ records per 'PutRecordsEndpoint'+-- request.+--+defaultBatchPolicy ∷ BatchPolicy+defaultBatchPolicy = BatchPolicy+  { _bpBatchSize = 200+  }++-- | The producer will attempt to re-send records which failed according to a+-- user-specified policy. This policy applies to failures which occur in the+-- process of sending a message to Kinesis, not those which occur in the course+-- of enqueuing a message.+data RetryPolicy+  = RetryPolicy+  { _rpRetryCount ∷ !Natural+  } deriving (Eq, Show)++-- | The number of times to retry sending a message after it has first failed.+--+rpRetryCount ∷ Lens' RetryPolicy Natural+rpRetryCount = lens _rpRetryCount $ \rp n → rp { _rpRetryCount = n }++-- | The default retry policy will attempt @5@ retries for a message.+--+defaultRetryPolicy ∷ RetryPolicy+defaultRetryPolicy = RetryPolicy+  { _rpRetryCount = 5+  }++-- | A proxy object for specifying a concrete queue implementation. You may+-- provide your own, or use 'defaultQueueImplementation'.+--+data QueueImplementation+  = ∀ proxy q+  . BoundedCloseableQueue q MessageQueueItem+  ⇒ QueueImplementation (proxy q)++-- | The default 'QueueImplementation' is based on 'TBMChan'.+--+defaultQueueImplementation ∷ QueueImplementation+defaultQueueImplementation = QueueImplementation (Proxy ∷ ∀ α. Proxy (TBMChan α))++-- | The basic input required to construct a Kinesis producer.+--+data ProducerKit+  = ProducerKit+  { _pkKinesisKit ∷ !KinesisKit+  -- ^ The basic information required to send requests to AWS Kinesis.++  , _pkStreamName ∷ !StreamName+  -- ^ The name of the stream to send records to.++  , _pkBatchPolicy ∷ !BatchPolicy+  -- ^ The record batching policy for the producer.++  , _pkRetryPolicy ∷ !RetryPolicy+  -- ^ The retry policy for the producer.++  , _pkMessageQueueBounds ∷ !Natural+  -- ^ The maximum number of records that may be enqueued at one time.++  , _pkMaxConcurrency ∷ !Natural+  -- ^ The number of requests to run concurrently (minimum: 1).++  , _pkCleanupTimeout ∷ !(Maybe Natural)+  -- ^ The timeout in microseconds, after which the producer's cleanup routine+  -- will terminate, finished or not, throwing 'ProducerCleanupTimedOut'.++  , _pkQueueImplementation ∷ QueueImplementation+  -- ^ The Kinesis Producer is parameterized over a concrete queue implementation.+  }++-- | Create a 'ProducerKit' with default settings.+--+makeProducerKit+  ∷ KinesisKit+  → StreamName+  → ProducerKit+makeProducerKit kinesisKit stream = ProducerKit+  { _pkKinesisKit = kinesisKit+  , _pkStreamName = stream+  , _pkBatchPolicy = defaultBatchPolicy+  , _pkRetryPolicy = defaultRetryPolicy+  , _pkMessageQueueBounds = 10000+  , _pkMaxConcurrency = 3+  , _pkCleanupTimeout = Nothing+  , _pkQueueImplementation = defaultQueueImplementation+  }++-- | A lens for '_pkKinesisKit'.+--+pkKinesisKit ∷ Lens' ProducerKit KinesisKit+pkKinesisKit = lens _pkKinesisKit $ \pk kk → pk { _pkKinesisKit = kk }++-- | A lens for '_pkStreamName'.+--+pkStreamName ∷ Lens' ProducerKit StreamName+pkStreamName = lens _pkStreamName $ \pk sn → pk { _pkStreamName = sn }++-- | A lens for '_pkBatchPolicy'.+--+pkBatchPolicy ∷ Lens' ProducerKit BatchPolicy+pkBatchPolicy = lens _pkBatchPolicy $ \pk bp → pk { _pkBatchPolicy = bp }++-- | A lens for '_pkRetryPolicy'.+--+pkRetryPolicy ∷ Lens' ProducerKit RetryPolicy+pkRetryPolicy = lens _pkRetryPolicy $ \pk rp → pk { _pkRetryPolicy = rp }++-- | A lens for '_pkMessageQueueBounds'.+--+pkMessageQueueBounds ∷ Lens' ProducerKit Natural+pkMessageQueueBounds = lens _pkMessageQueueBounds $ \pk qb → pk { _pkMessageQueueBounds = qb }++-- | A lens for '_pkMaxConcurrency'.+--+pkMaxConcurrency ∷ Lens' ProducerKit Natural+pkMaxConcurrency = lens _pkMaxConcurrency $ \pk n → pk { _pkMaxConcurrency = n }++-- | A lens for '_pkCleanupTimeout'.+--+pkCleanupTimeout ∷ Lens' ProducerKit (Maybe Natural)+pkCleanupTimeout = lens _pkCleanupTimeout $ \pk n → pk { _pkCleanupTimeout = n }++-- | A lens for '_pkQueueImplementation'.+--+pkQueueImplementation ∷ Setter' ProducerKit QueueImplementation+pkQueueImplementation = lens _pkQueueImplementation $ \pk q → pk { _pkQueueImplementation = q }

@@ -0,0 +1,64 @@+-- Copyright (c) 2013-2015 PivotCloud, Inc.+--+-- Aws.Kinesis.Client.Producer.Kit+--+-- Please feel free to contact us at licensing@pivotmail.com with any+-- contributions, additions, or other feedback; we would love to hear from+-- you.+--+-- Licensed under the Apache License, Version 2.0 (the "License"); you may+-- not use this file except in compliance with the License. You may obtain a+-- copy of the License at http://www.apache.org/licenses/LICENSE-2.0+--+-- Unless required by applicable law or agreed to in writing, software+-- distributed under the License is distributed on an "AS IS" BASIS, WITHOUT+-- WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the+-- License for the specific language governing permissions and limitations+-- under the License.++{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE UnicodeSyntax #-}++-- |+-- Module: Aws.Kinesis.Client.Producer.Kit+-- Copyright: Copyright © 2013-2015 PivotCloud, Inc.+-- License: Apache-2.0+-- Maintainer: Jon Sterling <jsterling@alephcloud.com>+-- Stability: experimental+--+module Aws.Kinesis.Client.Producer.Kit+( -- * Producer Kit+  ProducerKit+, makeProducerKit+, pkKinesisKit+, pkStreamName+, pkBatchPolicy+, pkRetryPolicy+, pkMessageQueueBounds+, pkMaxConcurrency+, pkCleanupTimeout+, pkQueueImplementation++  -- * Queue Implementations+, QueueImplementation(..)+, defaultQueueImplementation++  -- * Policies+, BatchPolicy+, defaultBatchPolicy+, bpBatchSize++, RetryPolicy+, defaultRetryPolicy+, rpRetryCount++  -- ** Constants+, pattern MaxMessageSize+) where++import Aws.Kinesis.Client.Producer.Internal++-- | The maximum size in bytes of a message.+--+pattern MaxMessageSize = 51000+