diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -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.
diff --git a/aws-kinesis-client.cabal b/aws-kinesis-client.cabal
--- a/aws-kinesis-client.cabal
+++ b/aws-kinesis-client.cabal
@@ -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
diff --git a/cli/CLI.hs b/cli/CLI.hs
--- a/cli/CLI.hs
+++ b/cli/CLI.hs
@@ -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
diff --git a/cli/CLI/Options.hs b/cli/CLI/Options.hs
--- a/cli/CLI/Options.hs
+++ b/cli/CLI/Options.hs
@@ -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
diff --git a/src/Aws/Kinesis/Client/Common.hs b/src/Aws/Kinesis/Client/Common.hs
--- a/src/Aws/Kinesis/Client/Common.hs
+++ b/src/Aws/Kinesis/Client/Common.hs
@@ -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
 
diff --git a/src/Aws/Kinesis/Client/Consumer.hs b/src/Aws/Kinesis/Client/Consumer.hs
--- a/src/Aws/Kinesis/Client/Consumer.hs
+++ b/src/Aws/Kinesis/Client/Consumer.hs
@@ -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,)
diff --git a/src/Aws/Kinesis/Client/Consumer/Internal.hs b/src/Aws/Kinesis/Client/Consumer/Internal.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Consumer/Internal.hs
@@ -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
diff --git a/src/Aws/Kinesis/Client/Consumer/Internal/Kit.hs b/src/Aws/Kinesis/Client/Consumer/Internal/Kit.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Consumer/Internal/Kit.hs
@@ -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 }
+
diff --git a/src/Aws/Kinesis/Client/Consumer/Internal/SavedStreamState.hs b/src/Aws/Kinesis/Client/Consumer/Internal/SavedStreamState.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Consumer/Internal/SavedStreamState.hs
@@ -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
+
+
diff --git a/src/Aws/Kinesis/Client/Consumer/Internal/ShardState.hs b/src/Aws/Kinesis/Client/Consumer/Internal/ShardState.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Consumer/Internal/ShardState.hs
@@ -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
+
diff --git a/src/Aws/Kinesis/Client/Consumer/Kit.hs b/src/Aws/Kinesis/Client/Consumer/Kit.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Consumer/Kit.hs
@@ -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
diff --git a/src/Aws/Kinesis/Client/Internal/Queue.hs b/src/Aws/Kinesis/Client/Internal/Queue.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Internal/Queue.hs
@@ -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 []
+
diff --git a/src/Aws/Kinesis/Client/Internal/Queue/Chunk.hs b/src/Aws/Kinesis/Client/Internal/Queue/Chunk.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Internal/Queue/Chunk.hs
@@ -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
+
diff --git a/src/Aws/Kinesis/Client/Producer.hs b/src/Aws/Kinesis/Client/Producer.hs
--- a/src/Aws/Kinesis/Client/Producer.hs
+++ b/src/Aws/Kinesis/Client/Producer.hs
@@ -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] " ⊕)
diff --git a/src/Aws/Kinesis/Client/Producer/Internal.hs b/src/Aws/Kinesis/Client/Producer/Internal.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Producer/Internal.hs
@@ -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 }
diff --git a/src/Aws/Kinesis/Client/Producer/Kit.hs b/src/Aws/Kinesis/Client/Producer/Kit.hs
new file mode 100644
--- /dev/null
+++ b/src/Aws/Kinesis/Client/Producer/Kit.hs
@@ -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
+