aws-kinesis-client 0.2.0.2 → 0.2.0.3
raw patch · 5 files changed
+56/−13 lines, 5 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- CHANGELOG.md +12/−0
- aws-kinesis-client.cabal +2/−1
- cli/CLI.hs +26/−8
- cli/CLI/Options.hs +15/−4
- src/Aws/Kinesis/Client/Producer.hs +1/−0
CHANGELOG.md view
@@ -1,3 +1,15 @@+### v0.2.0.3++All changes were in the Consumer CLI.++- Add a --timeout option, which will cause the consumer CLI terminate after a+ specified number of seconds.++- Change the behavior of --save-state to only save the state when a run of the+ consumer CLI was successful (i.e. either no limit was specified, or the+ precise number of items requested was indeed retrieved within the alotted time).++ ### v0.2.0.2 - Upgrade to newer versions of `aws-general` and `aws-kinesis` which support the
aws-kinesis-client.cabal view
@@ -1,5 +1,5 @@ name: aws-kinesis-client-version: 0.2.0.2+version: 0.2.0.3 synopsis: A producer & consumer client library for AWS Kinesis -- description: license: Apache-2.0@@ -71,6 +71,7 @@ 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, mtl >=2.2.1,
cli/CLI.hs view
@@ -26,6 +26,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TupleSections #-} {-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE UnicodeSyntax #-} @@ -47,6 +48,8 @@ import CLI.Options +import Control.Concurrent.Lifted+import Control.Concurrent.Async.Lifted import Control.Exception.Lifted import Control.Lens import Control.Monad@@ -126,7 +129,7 @@ A.eitherDecode <$> code Nothing → return Nothing - consumer ← managedKinesisConsumer $ ConsumerKit+ consumer ← managedKinesisConsumer ConsumerKit { _ckKinesisKit = KinesisKit { _kkManager = manager , _kkConfiguration = Configuration@@ -144,15 +147,30 @@ let source = consumerSource consumer- presink = CL.mapM_ $ liftIO ∘ B8.putStrLn ∘ recordData+ step n r = succ n <$ liftIO (B8.putStrLn $ recordData r)+ countingSink = CL.foldM step (1 ∷ Int) sink = case _clioLimit of- Just limit → CL.isolate limit =$ presink- Nothing → presink+ Just limit → CL.isolate limit =$ countingSink+ Nothing → countingSink - lift ∘ finally (source $$ sink) $- void ∘ for _clioStateOut $ \outPath → do- state ← consumerStreamState consumer- liftIO ∘ BL8.writeFile outPath $ A.encode state+ runConsumer = do+ n ← catch (source $$ sink) $ \SomeException{} → return 0+ return $ maybe True (n ≥) _clioLimit+++ result ← lift $+ case _clioTimeout of+ Just seconds → race (threadDelay $ seconds * 1000000) runConsumer+ Nothing → Right <$> runConsumer++ let+ successful = case result of+ Left () → isn't _Just _clioLimit+ Right b → b++ lift ∘ when successful ∘ void ∘ for _clioStateOut $ \outPath → do+ state ← consumerStreamState consumer+ liftIO ∘ BL8.writeFile outPath $ A.encode state main ∷ IO () main =
cli/CLI/Options.hs view
@@ -32,6 +32,7 @@ ( CLIOptions(..) , clioStreamName , clioLimit+, clioTimeout , clioIteratorType , clioAccessKeys , clioStateIn@@ -75,6 +76,7 @@ { _clioStreamName ∷ !StreamName , _clioRegion ∷ !Region , _clioLimit ∷ !(Maybe Int)+ , _clioTimeout ∷ !(Maybe Int) , _clioIteratorType ∷ !ShardIteratorType , _clioAccessKeys ∷ !(Maybe CredentialsInput) , _clioUseInstanceMetadata ∷ !Bool@@ -136,8 +138,16 @@ long "limit" ⊕ short 'l' ⊕ metavar "L"- ⊕ help "Fetch `L` records"+ ⊕ help "Fetch `L` records. If a limit is provided, then the run will only be considered successful if it results in the CLI fetching `L` records; otherwise, a run is always considered successful." +timeoutParser ∷ Parser Int+timeoutParser =+ option auto $+ long "timeout"+ ⊕ short 't'+ ⊕ metavar "T"+ ⊕ help "Terminate the consumer after `T` seconds. Even if a limit has been provided, the consumer will terminate after at most `T` seconds."+ iteratorTypeParser ∷ Parser ShardIteratorType iteratorTypeParser = option auto $@@ -152,7 +162,7 @@ stateOutParser = strOption $ long "save-state"- ⊕ help "Write the last known state of each shard to a file"+ ⊕ help "Write the last known state of each shard to a file; this will only occur when the run has completed in a \"successful\" state." ⊕ metavar "FILE" stateInParser ∷ Parser FilePath@@ -194,6 +204,7 @@ ⊛ streamNameParser ⊛ regionParser ⊛ optional limitParser+ ⊛ optional timeoutParser ⊛ iteratorTypeParser ⊛ optional credentialsInputParser ⊛ useInstanceMetadataParser@@ -204,6 +215,6 @@ parserInfo = info (helper ⊛ optionsParser) $ fullDesc- ⊕ progDesc "Fetch a given number of records from a Kinesis stream; unlike the standard command line utilities, this interface is suitable for use with a sharded stream. If you both specify a saved stream state to be restored and an iterator type, the latter will be used on any shards which are not contained in the saved state. Minimally, you must specify your AWS credentials, a stream name, and an optional limit."- ⊕ header "The Kinesis Consumer CLI v0.2.0.1"+ ⊕ progDesc "Fetch a given number of records from a Kinesis stream; unlike the standard command line utilities, this interface is suitable for use with a sharded stream. If you both specify a saved stream state to be restored and an iterator type, the latter will be used on any shards which are not contained in the saved state. Minimally, you must specify your AWS credentials, a stream name, and an optional limit & timeout."+ ⊕ header "The Kinesis Consumer CLI v0.2.0.3"
src/Aws/Kinesis/Client/Producer.hs view
@@ -72,6 +72,7 @@ ) where import qualified Aws.Kinesis as Kin+import qualified Aws.Kinesis.Commands.PutRecords as Kin import Aws.Kinesis.Client.Common import Control.Applicative