aws-kinesis-client 0.2.0.3 → 0.3.0.0

raw patch · 3 files changed

+130/−38 lines, 3 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +9/−0
• aws-kinesis-client.cabal +1/−1
• src/Aws/Kinesis/Client/Producer.hs +120/−37

@@ -1,3 +1,12 @@+### v0.3.0.0++All changes were in the Producer client.++- Do not retry messages indefinitely; add support for a configurable retry+  policy.++- Reject messages synchronously which exceed the Kinesis record size limit.+ ### v0.2.0.3  All changes were in the Consumer CLI.

@@ -1,5 +1,5 @@ name:                aws-kinesis-client-version:             0.2.0.3+version:             0.3.0.0 synopsis:            A producer & consumer client library for AWS Kinesis -- description: license:             Apache-2.0

@@ -28,6 +28,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TupleSections #-}@@ -56,6 +57,7 @@ , pkKinesisKit , pkStreamName , pkBatchPolicy+, pkRetryPolicy , pkMessageQueueBounds , pkMaxConcurrency @@ -63,11 +65,19 @@ , _KinesisError , _MessageNotEnqueued , _InvalidConcurrentConsumerCount+, _MessageTooLarge +, pattern MaxMessageSize+ , BatchPolicy , defaultBatchPolicy , bpBatchSize , bpEndpoint++, RetryPolicy+, defaultRetryPolicy+, rpRetryCount+ , RecordEndpoint(..) ) where @@ -110,6 +120,10 @@    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@@ -129,7 +143,7 @@ bpEndpoint ∷ Lens' BatchPolicy RecordEndpoint bpEndpoint = lens _bpEndpoint $ \bp ep → bp { _bpEndpoint = ep } --- | The default batching policy sends '200' records per 'PutRecordsEndpoint'+-- | The default batching policy sends @200@ records per 'PutRecordsEndpoint' -- request. -- defaultBatchPolicy ∷ BatchPolicy@@ -138,13 +152,39 @@   , _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@@ -153,6 +193,15 @@ 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@@ -166,6 +215,9 @@   , _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. @@ -188,6 +240,11 @@ 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@@ -200,14 +257,18 @@  -- | The (abstract) Kinesis producer client. ---newtype KinesisProducer+data KinesisProducer   = KinesisProducer-  { _kpMessageQueue ∷ TBMQueue MessageQueueItem+  { _kpMessageQueue ∷ !(TBMQueue MessageQueueItem)+  , _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.@@ -216,6 +277,9 @@   -- ^ Thrown when a message could not be enqueued since the queue was full.   -- This error must be handled at the call-site. +  | MessageTooLarge+  -- ^ Thrown when the message was larger than the maximum message size ('MaxMessageSize').+   | InvalidConcurrentConsumerCount   -- ^ Thrown when 'pkMaxConcurrency' is set with an invalid value. @@ -239,6 +303,14 @@     MessageNotEnqueued m → Right m     e → Left e +-- | A prism for 'MessageTooLarge'.+--+_MessageTooLarge ∷ Prism' ProducerError ()+_MessageTooLarge =+  prism (const MessageTooLarge) $ \case+    MessageTooLarge → Right ()+    e → Left e+ -- | A prism for 'InvalidConcurrentConsumerCount'. -- _InvalidConcurrentConsumerCount ∷ Prism' ProducerError ()@@ -375,17 +447,18 @@             putStrLn $ "Error: " ++ show e             putStrLn "Will wait 5s"             threadDelay 5000000-          leftover item+          leftover $ item & mqiRemainingAttempts -~ 1 -    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-        }+    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@@ -408,31 +481,34 @@   maxWorkerCount ← view pkMaxConcurrency   awaitForever $ \messages → do     let batches = splitEvery batchSize messages-    leftovers ← lift ∘ flip (mapConcurrentlyN maxWorkerCount 100) batches $ \ms → do-      let handler e = do-            liftIO $ print e-            return ms--      handleError handler $ do-        items ← for ms $ \m → do-          let partitionKey = m ^. mqiPartitionKey-          return Kin.PutRecordsRequestEntry-            { Kin.putRecordsRequestEntryData = m ^. mqiMessage ∘ to T.encodeUtf8-            , Kin.putRecordsRequestEntryExplicitHashKey = Nothing-            , Kin.putRecordsRequestEntryPartitionKey = partitionKey-            }+    leftovers ← lift ∘ flip (mapConcurrentlyN maxWorkerCount 100) batches $ \items → do+      case filter messageQueueItemIsEligible items of+        [] → return []+        eligibleItems → do+          handleError (\e → eligibleItems <$ liftIO (print e)) $ do+            requestEntries ← for eligibleItems $ \m → do+              let partitionKey = m ^. mqiPartitionKey+              return Kin.PutRecordsRequestEntry+                { Kin.putRecordsRequestEntryData = m ^. mqiMessage ∘ to T.encodeUtf8+                , Kin.putRecordsRequestEntryExplicitHashKey = Nothing+                , Kin.putRecordsRequestEntryPartitionKey = partitionKey+                } -        Kin.PutRecordsResponse{..} ←  liftKinesis $ runKinesis Kin.PutRecords-          { Kin.putRecordsRecords = items-          , Kin.putRecordsStreamName = streamName-          }-        let processResult m m'-              | isJust (Kin.putRecordsResponseRecordErrorCode m') = Just m-              | otherwise = Nothing-        return ∘ catMaybes $ zipWith processResult ms putRecordsResponseRecords+            Kin.PutRecordsResponse{..} ←  liftKinesis $ runKinesis Kin.PutRecords+              { Kin.putRecordsRecords = requestEntries+              , Kin.putRecordsStreamName = streamName+              }+            let+              processResult m m'+                | isJust (Kin.putRecordsResponseRecordErrorCode m') = Just m+                | otherwise = Nothing+            return ∘ catMaybes $ zipWith processResult eligibleItems putRecordsResponseRecords -    forM_ leftovers $ \mss →-      unless (null mss) $ leftover mss+    forM_ leftovers $ \items →+      unless (null items) $+        leftover $ items+          <&> mqiRemainingAttempts -~ 1+           & filter messageQueueItemIsEligible  sendMessagesSink   ∷ MonadProducerInternal m@@ -453,11 +529,15 @@   → 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 ()@@ -516,7 +596,10 @@    Codensity $ \inner → do     link consumerHandle-    res ← inner $ KinesisProducer messageQueue+    res ← inner KinesisProducer+      { _kpMessageQueue = messageQueue+      , _kpRetryPolicy = kit ^. pkRetryPolicy+      }     () ← wait consumerHandle     return res