aws-kinesis-client 0.2.0.0 → 0.2.0.1
raw patch · 4 files changed
+92/−30 lines, 4 files
Files
- CHANGELOG.md +15/−0
- aws-kinesis-client.cabal +1/−1
- cli/CLI.hs +28/−23
- cli/CLI/Options.hs +48/−6
CHANGELOG.md view
@@ -1,3 +1,18 @@+### v0.2.0.1++All changes were in the Consumer CLI.++- Add switch to get credentials from EC2 instance metadata,+ --use-instance-metadata++- Make --limit optional to consume indefinitely++- Allow custom region with --region++- If the CLI is terminated for any reason, before shutting down it will record+ its state if --save-state is passed++ ### v0.2.0.0 - [Consumer] Add the ability to save & restore stream state (i.e. last consumed
aws-kinesis-client.cabal view
@@ -1,5 +1,5 @@ name: aws-kinesis-client-version: 0.2.0.0+version: 0.2.0.1 synopsis: A producer & consumer client library for AWS Kinesis -- description: license: Apache-2.0
cli/CLI.hs view
@@ -41,14 +41,13 @@ ) where import Aws-import Aws.General import Aws.Kinesis hiding (Record) import Aws.Kinesis.Client.Common import Aws.Kinesis.Client.Consumer import CLI.Options -import Control.Exception+import Control.Exception.Lifted import Control.Lens import Control.Monad import Control.Monad.Trans@@ -75,6 +74,7 @@ data CLIError = MissingCredentials+ | NoInstanceMetadataCredentials deriving (Typeable, Show) instance Exception CLIError@@ -86,25 +86,25 @@ , MonadError ConsumerError m ) -limitConduit- ∷ MonadCLI m- ⇒ Conduit a m a-limitConduit =- lift (view clioLimit) ≫=- CL.isolate- fetchCredentials ∷ MonadCLI m ⇒ m Credentials fetchCredentials = do- view clioAccessKeys ≫= \case- Left aks →- makeCredentials- (aks ^. akAccessKeyId)- (aks ^. akSecretAccessKey)- Right path →- loadCredentialsFromFile path credentialsDefaultKey- <!?> KinesisError (toException MissingCredentials)+ view clioUseInstanceMetadata ≫= \case+ True →+ loadCredentialsFromInstanceMetadata+ <!?> KinesisError (toException NoInstanceMetadataCredentials)+ False →+ view clioAccessKeys ≫= \case+ Just (CredentialsFromAccessKeys aks) →+ makeCredentials+ (aks ^. akAccessKeyId)+ (aks ^. akSecretAccessKey)+ Just (CredentialsFromFile path) →+ loadCredentialsFromFile path credentialsDefaultKey+ <!?> KinesisError (toException MissingCredentials)+ Nothing →+ throwError $ KinesisError (toException MissingCredentials) app ∷ MonadCLI m@@ -128,7 +128,7 @@ , credentials = credentials , logger = defaultLog Warning }- , _kkKinesisConfiguration = KinesisConfiguration UsWest2+ , _kkKinesisConfiguration = KinesisConfiguration _clioRegion } , _ckStreamName = _clioStreamName , _ckBatchSize = 100@@ -136,12 +136,17 @@ , _ckSavedStreamState = savedStreamState } - lift $ consumerSource consumer $$- limitConduit =$ CL.mapM_ (liftIO ∘ B8.putStrLn ∘ recordData)+ let+ source = consumerSource consumer+ presink = CL.mapM_ $ liftIO ∘ B8.putStrLn ∘ recordData+ sink = case _clioLimit of+ Just limit → CL.isolate limit =$ presink+ Nothing → presink - void ∘ for _clioStateOut $ \outPath → do- state ← lift $ consumerStreamState consumer- liftIO ∘ BL8.writeFile outPath $ A.encode state+ lift ∘ finally (source $$ sink) $+ void ∘ for _clioStateOut $ \outPath → do+ state ← consumerStreamState consumer+ liftIO ∘ BL8.writeFile outPath $ A.encode state main ∷ IO () main =
cli/CLI/Options.hs view
@@ -36,6 +36,9 @@ , clioAccessKeys , clioStateIn , clioStateOut+, clioUseInstanceMetadata+, clioRegion+, CredentialsInput(..) , AccessKeys(..) , akAccessKeyId , akSecretAccessKey@@ -43,13 +46,17 @@ , parserInfo ) where +import Aws.General import Aws.Kinesis import qualified Data.ByteString.Char8 as B8 import Control.Applicative.Unicode import Control.Lens+import Control.Monad.Unicode import Data.Monoid.Unicode+import qualified Data.Text as T import qualified Data.Text.Lens as T import Options.Applicative+import Options.Applicative.Types import Prelude.Unicode data AccessKeys@@ -58,12 +65,19 @@ , _akSecretAccessKey ∷ !B8.ByteString } deriving Show +data CredentialsInput+ = CredentialsFromAccessKeys AccessKeys+ | CredentialsFromFile FilePath+ deriving Show+ data CLIOptions = CLIOptions { _clioStreamName ∷ !StreamName- , _clioLimit ∷ !Int+ , _clioRegion ∷ !Region+ , _clioLimit ∷ !(Maybe Int) , _clioIteratorType ∷ !ShardIteratorType- , _clioAccessKeys ∷ !(Either AccessKeys FilePath)+ , _clioAccessKeys ∷ !(Maybe CredentialsInput)+ , _clioUseInstanceMetadata ∷ !Bool , _clioStateIn ∷ !(Maybe FilePath) , _clioStateOut ∷ !(Maybe FilePath) } deriving Show@@ -148,13 +162,41 @@ ⊕ help "Read a saved stream state from a file. For any shards whose state is restored, the 'AFTER_SEQUENCE_NUMBER' iterator type will be used; other shards will use the iterator type you have specified. Some shards may have been merged or closed between when the state was saved and restored; at this point, no effort has been made to do anything here beyond the obvious (shards are identified by their shard-id)." ⊕ metavar "FILE" +credentialsInputParser ∷ Parser CredentialsInput+credentialsInputParser =+ foldr (<|>) empty $+ [ CredentialsFromAccessKeys <$> accessKeysParser+ , CredentialsFromFile <$> accessKeysPathParser+ ]++useInstanceMetadataParser ∷ Parser Bool+useInstanceMetadataParser =+ switch $+ long "use-instance-metadata"+ ⊕ help "Read the credentials from the instance metadata"++regionReader ∷ ReadM Region+regionReader = do+ fromText ∘ T.pack <$> readerAsk ≫=+ either readerError return+++regionParser ∷ Parser Region+regionParser =+ option regionReader $+ long "region"+ ⊕ value UsWest2+ ⊕ help "Choose an AWS Kinesis region (default: us-west-2)"+ optionsParser ∷ Parser CLIOptions optionsParser = pure CLIOptions ⊛ streamNameParser- ⊛ limitParser+ ⊛ regionParser+ ⊛ optional limitParser ⊛ iteratorTypeParser- ⊛ (Left <$> accessKeysParser <|> Right <$> accessKeysPathParser)+ ⊛ optional credentialsInputParser+ ⊛ useInstanceMetadataParser ⊛ optional stateInParser ⊛ optional stateOutParser @@ -162,6 +204,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 a limit."- ⊕ header "The Kinesis Consumer CLI"+ ⊕ 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"