diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,4 @@ +0.1 +=== + +First public version. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,20 @@ +Copyright (c) 2014 AlephCloud Systems, Inc. + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be +included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,68 @@ +[![Build Status](https://travis-ci.org/alephcloud/hs-aws-general.svg?branch=master)](https://travis-ci.org/alephcloud/hs-aws-kinesis) + + +Haskell Bindings for Amazon AWS Kinesis +======================================= + +*API Version 2013-12-02* + +[Amazon AWS Kinesis API Reference](http://docs.aws.amazon.com/kinesis/2013-12-02/APIReference) + +This package depends on the [aws-general](https://github.com/alephcloud/hs-aws-general) package +and the [aws](https://github.com/aristidb/aws) package. From the latter the it borrows the +machinery for [managing AWS credentials](https://hackage.haskell.org/package/aws-0.9/docs/Aws.html#g:17) +and [making requests](https://hackage.haskell.org/package/aws-0.9/docs/Aws.html). There +is also some documentation, including an usage example, in the +[README of the aws package](https://github.com/aristidb/aws/blob/master/README.org) + + +Installation +============ + +Assuming that the Haskell compiler *GHC* and the Haskell build tool *cabal* is already +installed run the following command from the shell: + +~~~{.sh} +cabal install --enable-tests +~~~ + +Running Tests +============= + +There are few test cases included in the package. These tests require an AWS account +and AWS credentials stored in the file `~/.aws-keys` in the format described in the +[documentation of the aws package](https://hackage.haskell.org/package/aws-0.9/docs/Aws.html#g:17). + +When running these tests some (low) costs may incur for usage of the AWS services. +Therefor the user must explicitly consent to the usage of the AWS credentials by +passing the commandline options `--run-with-aws-credentials` to the test application. + +~~~{.sh} +cabal test --test-option=--run-with-aws-credentials +~~~ + +Example Usage +============= + +Here is a very simple example for making a single request to AWS Kinesis. For more ellaborate +usage refer to the [documentation of the AWS package](https://hackage.haskell.org/package/aws). + +~~~{.haskell} +import Aws +import Aws.Core +import Aws.General +import Aws.Kinesis +import Data.IORef + +cfg <- Aws.baseConfiguration +creds <- Credentials "access-key-id" "secret-access-key" `fmap` newIORef [] +let kinesisCfg = KinesisConfiguration UsWest2 +simpleAws cfg kinesisCfg $ ListStreams Nothing Nothing +~~~ + +In order to run the example you must replace `"access-key-id"` and +`"secret-access-key"` with the respective values for your AWS account. + +You may also take a look at the test examples in +[tests/Main.hs](https://github.com/alephcloud/hs-aws-kinesis/blob/master/tests/Main.hs). + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/aws-kinesis.cabal b/aws-kinesis.cabal new file mode 100644 --- /dev/null +++ b/aws-kinesis.cabal @@ -0,0 +1,113 @@ +-- ------------------------------------------------------ -- +-- Copyright © 2014 AlephCloud Systems, Inc. +-- ------------------------------------------------------ -- + +Name: aws-kinesis +Version: 0.1 +Synopsis: Bindings for AWS Kinesis Version 2013-12-02 +description: + Bindings for AWS Kinesis + . + /API Version: 2013-12-02/ + . + + +Homepage: https://github.com/alephcloud/hs-aws-kinesis +Bug-reports: https://github.com/alephcloud/hs-aws-kinesis/issues +License: MIT +License-file: LICENSE +Author: Lars Kuhtz +Maintainer: Lars Kuhtz +Copyright: Copyright (c) 2014 AlephCloud, Inc. +Category: Network, Web, AWS, Cloud, Distributed Computing +Build-type: Simple + +cabal-version: >= 1.16 + +extra-doc-files: + README.md, + CHANGELOG.md + +extra-source-files: + constraints + +source-repository head + type: git + location: https://github.com/alephcloud/hs-aws-kinesis.git + +source-repository this + type: git + location: https://github.com/alephcloud/hs-aws-kinesis.git + tag: 0.1 + +Library + default-language: Haskell2010 + hs-source-dirs: src + + exposed-modules: + Aws.Kinesis + Aws.Kinesis.Core + Aws.Kinesis.Types + Aws.Kinesis.Commands.CreateStream + Aws.Kinesis.Commands.DeleteStream + Aws.Kinesis.Commands.PutRecord + Aws.Kinesis.Commands.GetRecords + Aws.Kinesis.Commands.GetShardIterator + Aws.Kinesis.Commands.ListStreams + Aws.Kinesis.Commands.DescribeStream + Aws.Kinesis.Commands.SplitShard + Aws.Kinesis.Commands.MergeShards + + build-depends: + QuickCheck >= 2.7, + aeson >= 0.7, + aws >= 0.9, + aws-general >= 0.1, + base == 4.*, + base64-bytestring >= 1.0, + blaze-builder >= 0.3, + bytestring >= 0.10, + conduit >= 1.1, + conduit-extra >= 1.1, + http-conduit >= 2.1, + http-types >= 0.8, + parsers >= 0.11, + quickcheck-instances >= 0.3, + resourcet >= 1.1, + text >= 1.1, + time >= 1.4, + transformers >= 0.3 + + ghc-options: -Wall + +test-suite kinesis-tests + type: exitcode-stdio-1.0 + default-language: Haskell2010 + hs-source-dirs: tests + main-is: Main.hs + + other-modules: + Utils + + build-depends: + QuickCheck >= 2.7, + aeson >= 0.7, + aws >= 0.9, + aws-general >= 0.1, + aws-kinesis, + base == 4.*, + bytestring >= 0.10, + errors >= 1.4.7, + mtl >= 2.0, + tagged >= 0.7, + tasty >= 0.8, + tasty-quickcheck >= 0.8, + text >= 1.1, + transformers >= 0.3 + + ghc-options: -Wall -threaded + + + + + diff --git a/constraints b/constraints new file mode 100644 --- /dev/null +++ b/constraints @@ -0,0 +1,102 @@ +constraints: QuickCheck ==2.7.5, + aeson ==0.7.0.6, + array ==0.5.0.0, + asn1-encoding ==0.8.1.3, + asn1-parse ==0.8.1, + asn1-types ==0.2.3, + attoparsec ==0.12.1.0, + attoparsec-conduit ==1.1.0, + aws ==0.9.2, + aws-general ==0.1.1, + aws-kinesis ==0.1, + base ==4.7.0.1, + base16-bytestring ==0.1.1.6, + base64-bytestring ==1.0.0.1, + blaze-builder ==0.3.3.2, + blaze-builder-conduit ==1.1.0, + blaze-html ==0.7.0.2, + blaze-markup ==0.6.1.0, + byteable ==0.1.1, + bytestring ==0.10.4.0, + case-insensitive ==1.2.0.0, + cereal ==0.4.0.1, + charset ==0.3.7, + cipher-aes ==0.2.8, + cipher-des ==0.0.6, + cipher-rc4 ==0.1.4, + conduit ==1.1.6, + conduit-extra ==1.1.1, + connection ==0.2.1, + containers ==0.5.5.1, + cookie ==0.4.1.1, + cprng-aes ==0.5.2, + crypto-cipher-types ==0.0.9, + crypto-numbers ==0.2.3, + crypto-pubkey ==0.2.4, + crypto-pubkey-types ==0.4.2.2, + crypto-random ==0.0.7, + cryptohash ==0.11.6, + data-default ==0.5.3, + data-default-class ==0.0.1, + data-default-instances-base ==0.0.1, + data-default-instances-containers ==0.0.1, + data-default-instances-dlist ==0.0.1, + data-default-instances-old-locale ==0.0.1, + deepseq ==1.3.0.2, + directory ==1.2.1.0, + dlist ==0.7.1, + exceptions ==0.6.1, + filepath ==1.3.0.2, + ghc-prim ==0.3.1.0, + hashable ==1.2.2.0, + http-client ==0.3.4, + http-client-tls ==0.2.1.2, + http-conduit ==2.1.2.3, + http-types ==0.8.5, + integer-gmp ==0.5.1.0, + lifted-base ==0.2.3.0, + mime-types ==0.1.0.4, + mmorph ==1.0.3, + monad-control ==0.3.3.0, + mtl ==2.1.3.1, + nats ==0.2, + network ==2.5.0.0, + old-locale ==1.0.0.6, + old-time ==1.1.0.2, + parsec ==3.1.5, + parsers ==0.12, + pem ==0.2.2, + pretty ==1.1.1.1, + primitive ==0.5.3.0, + process ==1.2.0.0, + publicsuffixlist ==0.1, + quickcheck-instances ==0.3.8, + random ==1.0.1.1, + resourcet ==1.1.2.2, + rts ==1.0, + scientific ==0.3.3.0, + securemem ==0.1.3, + semigroups ==0.15.1, + socks ==0.5.4, + streaming-commons ==0.1.3.1, + syb ==0.4.2, + system-filepath ==0.4.12, + template-haskell ==2.9.0.0, + text ==1.1.1.3, + tf-random ==0.5, + time ==1.4.2, + tls ==1.2.8, + transformers ==0.3.0.0, + transformers-base ==0.4.2, + unix ==2.7.0.1, + unordered-containers ==0.2.5.0, + utf8-string ==0.3.8, + vector ==0.10.11.0, + void ==0.6.1, + x509 ==1.4.11, + x509-store ==1.4.4, + x509-system ==1.4.5, + x509-validation ==1.5.0, + xml-conduit ==1.2.0.3, + xml-types ==0.3.4, + zlib ==0.5.4.1 diff --git a/src/Aws/Kinesis.hs b/src/Aws/Kinesis.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis.hs @@ -0,0 +1,58 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module: Aws.Kinesis +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- +-- +-- The types and functions of this package are supposed to be +-- use with the machinery from the +-- . +-- +-- Here is a very simple example for making a single request to AWS Kinesis. +-- +-- > import Aws +-- > import Aws.Core +-- > import Aws.General +-- > import Aws.Kinesis +-- > import Data.IORef +-- > +-- > cfg <- Aws.baseConfiguration +-- > creds <- Credentials "access-key-id" "secret-access-key" `fmap` newIORef [] +-- > let kinesisCfg = KinesisConfiguration UsWest2 +-- > simpleAws cfg kinesisCfg $ ListStreams Nothing Nothing +-- +-- In order to run the example you must replace @"access-key-id"@ and +-- @"secret-access-key"@ with the respective values for your AWS account. +-- +module Aws.Kinesis +( module Aws.Kinesis.Core +, module Aws.Kinesis.Types +, module Aws.Kinesis.Commands.CreateStream +, module Aws.Kinesis.Commands.DeleteStream +, module Aws.Kinesis.Commands.DescribeStream +, module Aws.Kinesis.Commands.GetRecords +, module Aws.Kinesis.Commands.GetShardIterator +, module Aws.Kinesis.Commands.ListStreams +, module Aws.Kinesis.Commands.MergeShards +, module Aws.Kinesis.Commands.PutRecord +, module Aws.Kinesis.Commands.SplitShard +) where + +import Aws.Kinesis.Core +import Aws.Kinesis.Types +import Aws.Kinesis.Commands.CreateStream +import Aws.Kinesis.Commands.DeleteStream +import Aws.Kinesis.Commands.DescribeStream +import Aws.Kinesis.Commands.GetRecords +import Aws.Kinesis.Commands.GetShardIterator +import Aws.Kinesis.Commands.ListStreams +import Aws.Kinesis.Commands.MergeShards +import Aws.Kinesis.Commands.PutRecord +import Aws.Kinesis.Commands.SplitShard diff --git a/src/Aws/Kinesis/Commands/CreateStream.hs b/src/Aws/Kinesis/Commands/CreateStream.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/CreateStream.hs @@ -0,0 +1,138 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE DeriveDataTypeable #-} + +-- | +-- Module: Aws.Kinesis.Commands.CreateStream +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- This operation adds a new Amazon Kinesis stream to your AWS account. A +-- stream captures and transports data records that are continuously emitted +-- from different data sources or producers. Scale-out within an Amazon Kinesis +-- stream is explicitly supported by means of shards, which are uniquely +-- identified groups of data records in an Amazon Kinesis stream. +-- +-- You specify and control the number of shards that a stream is composed of. +-- Each open shard can support up to 5 read transactions per second, up to a +-- maximum total of 2 MB of data read per second. Each shard can support up to +-- 1000 write transactions per second, up to a maximum total of 1 MB data +-- written per second. You can add shards to a stream if the amount of data +-- input increases and you can remove shards if the amount of data input +-- decreases. +-- +-- The stream name identifies the stream. The name is scoped to the AWS account +-- used by the application. It is also scoped by region. That is, two streams +-- in two different accounts can have the same name, and two streams in the +-- same account, but in two different regions, can have the same name. +-- +-- CreateStream is an asynchronous operation. Upon receiving a CreateStream +-- request, Amazon Kinesis immediately returns and sets the stream status to +-- CREATING. After the stream is created, Amazon Kinesis sets the stream status +-- to ACTIVE. You should perform read and write operations only on an ACTIVE +-- stream. +-- +-- You receive a LimitExceededException when making a CreateStream request if +-- you try to do one of the following: +-- +-- Have more than five streams in the CREATING state at any point in time. +-- Create more shards than are authorized for your account. Note: The default +-- limit for an AWS account is 10 shards per stream. If you need to create a +-- stream with more than 10 shards, contact AWS Support to increase the limit +-- on your account. +-- +-- You can use the DescribeStream operation to check the stream status, which +-- is returned in StreamStatus. +-- +-- CreateStream has a limit of 5 transactions per second per account. +-- +-- +-- +module Aws.Kinesis.Commands.CreateStream +( CreateStream(..) +, CreateStreamResponse(..) +, CreateStreamExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Core +import Aws.Kinesis.Types + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +createStreamAction :: KinesisAction +createStreamAction = KinesisCreateStream + +data CreateStream = CreateStream + { createStreamShardCound :: !Int + -- ^ The number of shards that the stream will use. The throughput of the + -- stream is a function of the number of shards; more shards are required + -- for greater provisioned throughput. + -- + -- Note: The default limit for an AWS account is 10 shards per stream. If + -- you need to create a stream with more than 10 shards, contact AWS + -- Support to increase the limit on your account. + -- + -- Note that this limit is not checked by the code. + + , createStreamStreamName :: !StreamName + -- ^ A name to identify the stream. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON CreateStream where + toJSON CreateStream{..} = object + [ "ShardCount" .= createStreamShardCound + , "StreamName" .= createStreamStreamName + ] + +data CreateStreamResponse = CreateStreamResponse + deriving (Show, Read, Eq, Ord, Typeable) + +instance ResponseConsumer r CreateStreamResponse where + type ResponseMetadata CreateStreamResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance FromJSON CreateStreamResponse where + parseJSON _ = return CreateStreamResponse + +instance SignQuery CreateStream where + type ServiceConfiguration CreateStream = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = createStreamAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction CreateStream CreateStreamResponse + +instance AsMemoryResponse CreateStreamResponse where + type MemoryResponse CreateStreamResponse = CreateStreamResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data CreateStreamExceptions + = CreateStreamInvalidArgumentException + -- ^ /Code 400/ + + | CreateStreamLimitExceededException + -- ^ /Code 400/ + + | CreateStreamResourceInUseException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + diff --git a/src/Aws/Kinesis/Commands/DeleteStream.hs b/src/Aws/Kinesis/Commands/DeleteStream.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/DeleteStream.hs @@ -0,0 +1,104 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | +-- Module: Aws.Kinesis.Commands.DeleteStream +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- This operation deletes a stream and all of its shards and data. You must +-- shut down any applications that are operating on the stream before you +-- delete the stream. If an application attempts to operate on a deleted +-- stream, it will receive the exception ResourceNotFoundException. +-- +-- If the stream is in the ACTIVE state, you can delete it. After a +-- DeleteStream request, the specified stream is in the DELETING state until +-- Amazon Kinesis completes the deletion. +-- +-- Note: Amazon Kinesis might continue to accept data read and write +-- operations, such as PutRecord and GetRecords, on a stream in the DELETING +-- state until the stream deletion is complete. +-- +-- When you delete a stream, any shards in that stream are also deleted. +-- +-- You can use the DescribeStream operation to check the state of the stream, +-- which is returned in StreamStatus. +-- +-- DeleteStream has a limit of 5 transactions per second per account. +-- +-- +-- +module Aws.Kinesis.Commands.DeleteStream +( DeleteStream(..) +, DeleteStreamResponse(..) +, DeleteStreamExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Core +import Aws.Kinesis.Types + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +deleteStreamAction :: KinesisAction +deleteStreamAction = KinesisDeleteStream + +data DeleteStream = DeleteStream + { deleteStreamStreamName :: !StreamName + -- ^ The name of the stream to delete. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON DeleteStream where + toJSON DeleteStream{..} = object + [ "StreamName" .= deleteStreamStreamName + ] + +data DeleteStreamResponse = DeleteStreamResponse + deriving (Show, Read, Eq, Ord, Typeable) + +instance FromJSON DeleteStreamResponse where + parseJSON _ = return DeleteStreamResponse + +instance ResponseConsumer r DeleteStreamResponse where + type ResponseMetadata DeleteStreamResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance SignQuery DeleteStream where + type ServiceConfiguration DeleteStream = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = deleteStreamAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction DeleteStream DeleteStreamResponse + +instance AsMemoryResponse DeleteStreamResponse where + type MemoryResponse DeleteStreamResponse = DeleteStreamResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data DeleteStreamExceptions + = DeleteStreamLimitExceededException + -- ^ /Code 400/ + + | DeleteStreamResourceNotFoundException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + diff --git a/src/Aws/Kinesis/Commands/DescribeStream.hs b/src/Aws/Kinesis/Commands/DescribeStream.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/DescribeStream.hs @@ -0,0 +1,124 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | +-- Module: Aws.Sns.Commands.DescribeStream +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-03-31/ +-- +-- This operation returns the following information about the stream: the +-- current status of the stream, the stream Amazon Resource Name (ARN), and an +-- array of shard objects that comprise the stream. For each shard object there +-- is information about the hash key and sequence number ranges that the shard +-- spans, and the IDs of any earlier shards that played in a role in a +-- MergeShards or SplitShard operation that created the shard. A sequence +-- number is the identifier associated with every record ingested in the Amazon +-- Kinesis stream. The sequence number is assigned by the Amazon Kinesis +-- service when a record is put into the stream. +-- +-- You can limit the number of returned shards using the Limit parameter. The +-- number of shards in a stream may be too large to return from a single call +-- to DescribeStream. You can detect this by using the HasMoreShards flag in +-- the returned output. HasMoreShards is set to true when there is more data +-- available. +-- +-- If there are more shards available, you can request more shards by using the +-- shard ID of the last shard returned by the DescribeStream request, in the +-- ExclusiveStartShardId parameter in a subsequent request to DescribeStream. +-- DescribeStream is a paginated operation. +-- +-- DescribeStream has a limit of 10 transactions per second per account. +-- +-- +-- +module Aws.Kinesis.Commands.DescribeStream +( DescribeStream(..) +, DescribeStreamResponse(..) +, DescribeStreamExceptions(..) +) where + +import Control.Applicative + +import Aws.Core +import Aws.Kinesis.Core +import Aws.Kinesis.Types + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +describeStreamAction :: KinesisAction +describeStreamAction = KinesisDescribeStream + +data DescribeStream = DescribeStream + { describeStreamExclusiveStartShardId:: !(Maybe ShardId) + -- ^ The shard ID of the shard to start with for the stream description. + + , describeStreamLimit :: !(Maybe Int) + -- ^ The maximum number of shards to return. + + , describeStreamStreamName :: !StreamName + -- ^ The name of the stream to describe. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON DescribeStream where + toJSON DescribeStream{..} = object + [ "ExclusiveStartShardId" .= describeStreamExclusiveStartShardId + , "Limit" .= describeStreamLimit + , "StreamName" .= describeStreamStreamName + ] + +data DescribeStreamResponse = DescribeStreamResponse + { describeStreamResStreamDescription :: !StreamDescription + -- ^ Contains the current status of the stream, the stream ARN, an array of + -- shard objects that comprise the stream, and states whether there are + -- more shards available. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ResponseConsumer r DescribeStreamResponse where + type ResponseMetadata DescribeStreamResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance FromJSON DescribeStreamResponse where + parseJSON = withObject "DescribeStreamResponse" $ \o -> DescribeStreamResponse + <$> o .: "StreamDescription" + +instance SignQuery DescribeStream where + type ServiceConfiguration DescribeStream = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = describeStreamAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction DescribeStream DescribeStreamResponse + +instance AsMemoryResponse DescribeStreamResponse where + type MemoryResponse DescribeStreamResponse = DescribeStreamResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data DescribeStreamExceptions + = DescribeStreamLimitExceededException + -- ^ /Code 400/ + + | DescribeStreamResourceNotFoundException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + + diff --git a/src/Aws/Kinesis/Commands/GetRecords.hs b/src/Aws/Kinesis/Commands/GetRecords.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/GetRecords.hs @@ -0,0 +1,143 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | +-- Module: Aws.Kinesis.Commands.GetRecords +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- This operation returns one or more data records from a shard. A GetRecords +-- operation request can retrieve up to 10 MB of data. +-- +-- You specify a shard iterator for the shard that you want to read data from +-- in the ShardIterator parameter. The shard iterator specifies the position in +-- the shard from which you want to start reading data records sequentially. A +-- shard iterator specifies this position using the sequence number of a data +-- record in the shard. For more information about the shard iterator, see +-- GetShardIterator. +-- +-- GetRecords may return a partial result if the response size limit is +-- exceeded. You will get an error, but not a partial result if the shard's +-- provisioned throughput is exceeded, the shard iterator has expired, or an +-- internal processing failure has occurred. Clients can request a smaller +-- amount of data by specifying a maximum number of returned records using the +-- Limit parameter. The Limit parameter can be set to an integer value of up to +-- 10,000. If you set the value to an integer greater than 10,000, you will +-- receive InvalidArgumentException. +-- +-- A new shard iterator is returned by every GetRecords request in +-- NextShardIterator, which you use in the ShardIterator parameter of the next +-- GetRecords request. When you repeatedly read from an Amazon Kinesis stream +-- use a GetShardIterator request to get the first shard iterator to use in +-- your first GetRecords request and then use the shard iterator returned in +-- NextShardIterator for subsequent reads. +-- +-- GetRecords can return null for the NextShardIterator to reflect that the +-- shard has been closed and that the requested shard iterator would never have +-- returned more data. +-- +-- If no items can be processed because of insufficient provisioned throughput +-- on the shard involved in the request, GetRecords throws +-- ProvisionedThroughputExceededException. +-- +-- +-- +module Aws.Kinesis.Commands.GetRecords +( GetRecords(..) +, GetRecordsResponse(..) +, GetRecordsExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Core +import Aws.Kinesis.Types + +import Control.Applicative + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +getRecordsAction :: KinesisAction +getRecordsAction = KinesisGetRecords + +data GetRecords = GetRecords + { getRecordsLimit :: !(Maybe Int) + -- ^ The maximum number of records to return, which can be set to a value + -- of up to 10,000. + + , getRecordsShardIterator :: !ShardIterator + -- ^ The position in the shard from which you want to start sequentially + -- reading data records. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON GetRecords where + toJSON GetRecords{..} = object + [ "Limit" .= getRecordsLimit + , "ShardIterator" .= getRecordsShardIterator + ] + +data GetRecordsResponse = GetRecordsResponse + { getRecordsResNextShardIterator :: !(Maybe ShardIterator) + -- ^ The next position in the shard from which to start sequentially + -- reading data records. If set to null, the shard has been closed and the + -- requested iterator will not return any more data. + + , getRecordsResRecords :: ![Record] + -- ^ List of Records + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance FromJSON GetRecordsResponse where + parseJSON = withObject "GetRecordsResponse" $ \o -> GetRecordsResponse + <$> o .:? "NextShardIterator" .!= Nothing + <*> o .: "Records" + +instance ResponseConsumer r GetRecordsResponse where + type ResponseMetadata GetRecordsResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance SignQuery GetRecords where + type ServiceConfiguration GetRecords = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = getRecordsAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction GetRecords GetRecordsResponse + +instance AsMemoryResponse GetRecordsResponse where + type MemoryResponse GetRecordsResponse = GetRecordsResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data GetRecordsExceptions + = GetRecordsExpiredIteratorException + -- ^ /Code 400/ + + | GetRecordsInvalidArgumentException + -- ^ /Code 400/ + + | GetRecordsProvisionedThroughputExceededException + -- ^ /Code 400/ + + | GetRecordsResourceNotFoundException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + + diff --git a/src/Aws/Kinesis/Commands/GetShardIterator.hs b/src/Aws/Kinesis/Commands/GetShardIterator.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/GetShardIterator.hs @@ -0,0 +1,151 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module: Aws.Kinesis.Commands.GetShardIterator +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- This operation returns a shard iterator in ShardIterator. The shard iterator +-- specifies the position in the shard from which you want to start reading +-- data records sequentially. A shard iterator specifies this position using +-- the sequence number of a data record in a shard. A sequence number is the +-- identifier associated with every record ingested in the Amazon Kinesis +-- stream. The sequence number is assigned by the Amazon Kinesis service when a +-- record is put into the stream. +-- +-- You must specify the shard iterator type in the GetShardIterator request. +-- For example, you can set the ShardIteratorType parameter to read exactly +-- from the position denoted by a specific sequence number by using the +-- AT_SEQUENCE_NUMBER shard iterator type, or right after the sequence number +-- by using the AFTER_SEQUENCE_NUMBER shard iterator type, using sequence +-- numbers returned by earlier PutRecord, GetRecords or DescribeStream +-- requests. You can specify the shard iterator type TRIM_HORIZON in the +-- request to cause ShardIterator to point to the last untrimmed record in the +-- shard in the system, which is the oldest data record in the shard. Or you +-- can point to just after the most recent record in the shard, by using the +-- shard iterator type LATEST, so that you always read the most recent data in +-- the shard. +-- +-- Note: Each shard iterator expires five minutes after it is returned to the +-- requester. +-- +-- When you repeatedly read from an Amazon Kinesis stream use a +-- GetShardIterator request to get the first shard iterator to to use in your +-- first GetRecords request and then use the shard iterator returned by the +-- GetRecords request in NextShardIterator for subsequent reads. A new shard +-- iterator is returned by every GetRecords request in NextShardIterator, which +-- you use in the ShardIterator parameter of the next GetRecords request. +-- +-- If a GetShardIterator request is made too often, you will receive a +-- ProvisionedThroughputExceededException. For more information about +-- throughput limits, see the Amazon Kinesis Developer Guide. +-- +-- GetShardIterator can return null for its ShardIterator to indicate that the +-- shard has been closed and that the requested iterator will return no more +-- data. A shard can be closed by a SplitShard or MergeShards operation. +-- +-- GetShardIterator has a limit of 5 transactions per second per account per +-- open shard. +-- +-- +-- +module Aws.Kinesis.Commands.GetShardIterator +( GetShardIterator(..) +, GetShardIteratorResponse(..) +, GetShardIteratorExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Types +import Aws.Kinesis.Core + +import Control.Applicative + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +getShardIteratorAction :: KinesisAction +getShardIteratorAction = KinesisGetShardIterator + +data GetShardIterator = GetShardIterator + { getShardIteratorShardId :: !ShardId + -- ^ The shard ID of the shard to get the iterator for. + + , getShardIteratorShardIteratorType :: !ShardIteratorType + -- ^ Determines how the shard iterator is used to start reading data + -- records from the shard. + + , getShardIteratorStartingSequenceNumber :: !(Maybe SequenceNumber) + -- ^ The sequence number of the data record in the shard from which to + -- start reading from. + + , getShardIteratorStreamName :: !StreamName + -- ^ The name of the stream. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON GetShardIterator where + toJSON GetShardIterator{..} = object + [ "ShardId" .= getShardIteratorShardId + , "ShardIteratorType" .= getShardIteratorShardIteratorType + , "SequenceNumber" .= getShardIteratorStartingSequenceNumber + , "StreamName" .= getShardIteratorStreamName + ] + +data GetShardIteratorResponse = GetShardIteratorResponse + { getShardIteratorResShardIterator :: !ShardIterator + -- ^ The position in the shard from which to start reading data records + -- sequentially. A shard iterator specifies this position using the + -- sequence number of a data record in a shard. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance FromJSON GetShardIteratorResponse where + parseJSON = withObject "GetShardIteratorResponse" $ \o -> GetShardIteratorResponse + <$> o .: "ShardIterator" + +instance ResponseConsumer r GetShardIteratorResponse where + type ResponseMetadata GetShardIteratorResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance SignQuery GetShardIterator where + type ServiceConfiguration GetShardIterator = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = getShardIteratorAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction GetShardIterator GetShardIteratorResponse + +instance AsMemoryResponse GetShardIteratorResponse where + type MemoryResponse GetShardIteratorResponse = GetShardIteratorResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data GetShardIteratorExceptions + = GetShardIteratorInvalidArgumentException + -- ^ /Code 400/ + + | GetShardIteratorProvisionedThroughputExceededException + -- ^ /Code 400/ + + | GetShardIteratorResourceNotFoundException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + diff --git a/src/Aws/Kinesis/Commands/ListStreams.hs b/src/Aws/Kinesis/Commands/ListStreams.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/ListStreams.hs @@ -0,0 +1,116 @@ +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | +-- Module: Aws.Kinesis.Commands.ListStreams +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- This operation returns an array of the names of all the streams that are +-- associated with the AWS account making the ListStreams request. A given AWS +-- account can have many streams active at one time. +-- +-- The number of streams may be too large to return from a single call to +-- ListStreams. You can limit the number of returned streams using the Limit +-- parameter. If you do not specify a value for the Limit parameter, Amazon +-- Kinesis uses the default limit, which is currently 10. +-- +-- You can detect if there are more streams available to list by using the +-- HasMoreStreams flag from the returned output. If there are more streams +-- available, you can request more streams by using the name of the last stream +-- returned by the ListStreams request in the ExclusiveStartStreamName +-- parameter in a subsequent request to ListStreams. The group of stream names +-- returned by the subsequent request is then added to the list. You can +-- continue this process until all the stream names have been collected in the +-- list. +-- +-- ListStreams has a limit of 5 transactions per second per account. +-- +-- +-- +module Aws.Kinesis.Commands.ListStreams +( ListStreams(..) +, ListStreamsResponse(..) +, ListStreamsExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Types +import Aws.Kinesis.Core + +import Control.Applicative + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +listStreamsAction :: KinesisAction +listStreamsAction = KinesisListStreams + +data ListStreams = ListStreams + { listStreamsExclusiveStartStreamName :: !(Maybe StreamName) + -- ^ The name of the stream to start the list with. + + , listStreamsLimit :: !(Maybe Int) + -- ^ The maximum number of streams to list. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON ListStreams where + toJSON ListStreams{..} = object + [ "ExclusiveStartStreamName" .= listStreamsExclusiveStartStreamName + , "StreamName" .= listStreamsLimit + ] + +data ListStreamsResponse = ListStreamsResponse + { listStreamsResHasMoreStreams :: !Bool + -- ^ If set to true, there are more streams available to list. + + , listStreamsResStreamNames :: ![StreamName] + -- ^ The names of the streams that are associated with the AWS account + -- making the ListStreams request. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance FromJSON ListStreamsResponse where + parseJSON = withObject "ListStreamsResponse" $ \o -> ListStreamsResponse + <$> o .: "HasMoreStreams" + <*> o .: "StreamNames" + +instance ResponseConsumer r ListStreamsResponse where + type ResponseMetadata ListStreamsResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance SignQuery ListStreams where + type ServiceConfiguration ListStreams = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = listStreamsAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction ListStreams ListStreamsResponse + +instance AsMemoryResponse ListStreamsResponse where + type MemoryResponse ListStreamsResponse = ListStreamsResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data ListStreamsExceptions + = ListStreamsLimitExceededException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + diff --git a/src/Aws/Kinesis/Commands/MergeShards.hs b/src/Aws/Kinesis/Commands/MergeShards.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/MergeShards.hs @@ -0,0 +1,140 @@ +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | +-- Module: Aws.Kinesis.Commands.MergeShards +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- This operation merges two adjacent shards in a stream and combines them into +-- a single shard to reduce the stream's capacity to ingest and transport data. +-- Two shards are considered adjacent if the union of the hash key ranges for +-- the two shards form a contiguous set with no gaps. For example, if you have +-- two shards, one with a hash key range of 276...381 and the other with a hash +-- key range of 382...454, then you could merge these two shards into a single +-- shard that would have a hash key range of 276...454. After the merge, the +-- single child shard receives data for all hash key values covered by the two +-- parent shards. +-- +-- MergeShards is called when there is a need to reduce the overall capacity of +-- a stream because of excess capacity that is not being used. The operation +-- requires that you specify the shard to be merged and the adjacent shard for +-- a given stream. For more information about merging shards, see the Amazon +-- Kinesis Developer Guide. +-- +-- If the stream is in the ACTIVE state, you can call MergeShards. If a stream +-- is in CREATING or UPDATING or DELETING states, then Amazon Kinesis returns a +-- ResourceInUseException. If the specified stream does not exist, Amazon +-- Kinesis returns a ResourceNotFoundException. +-- +-- You can use the DescribeStream operation to check the state of the stream, +-- which is returned in StreamStatus. +-- +-- MergeShards is an asynchronous operation. Upon receiving a MergeShards +-- request, Amazon Kinesis immediately returns a response and sets the +-- StreamStatus to UPDATING. After the operation is completed, Amazon Kinesis +-- sets the StreamStatus to ACTIVE. Read and write operations continue to work +-- while the stream is in the UPDATING state. +-- +-- You use the DescribeStream operation to determine the shard IDs that are +-- specified in the MergeShards request. +-- +-- If you try to operate on too many streams in parallel using CreateStream, +-- DeleteStream, MergeShards or SplitShard, you will receive a +-- LimitExceededException. +-- +-- MergeShards has limit of 5 transactions per second per account. +-- +-- +-- +module Aws.Kinesis.Commands.MergeShards +( MergeShards(..) +, MergeShardsResponse(..) +, MergeShardsExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Core +import Aws.Kinesis.Types + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +mergeShardsAction :: KinesisAction +mergeShardsAction = KinesisMergeShards + +data MergeShards = MergeShards + { mergeShardsAdjacentShardToMerge :: !ShardId + -- ^ The shard ID of the adjacent shard for the merge. + + , mergeShardsShardToMerge :: !ShardId + -- ^ The shard ID of the shard to combine with the adjacent shard for the + -- merge. + + , mergeShardsStreamName :: !StreamName + -- ^ The name of the stream for the merge. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON MergeShards where + toJSON MergeShards{..} = object + [ "AdjacentShardToMerge" .= mergeShardsAdjacentShardToMerge + , "ShardToMerge" .= mergeShardsShardToMerge + , "StreamName" .= mergeShardsStreamName + ] + +data MergeShardsResponse = MergeShardsResponse + deriving (Show, Read, Eq, Ord, Typeable) + +instance ResponseConsumer r MergeShardsResponse where + type ResponseMetadata MergeShardsResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance FromJSON MergeShardsResponse where + parseJSON _ = return MergeShardsResponse + +instance SignQuery MergeShards where + type ServiceConfiguration MergeShards = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = mergeShardsAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction MergeShards MergeShardsResponse + +instance AsMemoryResponse MergeShardsResponse where + type MemoryResponse MergeShardsResponse = MergeShardsResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data MergeShardsExceptions + = MergeShardsInvalidArgumentException + -- ^ /Code 400/ + + | MergeShardsLimitExceededException + -- ^ /Code 400/ + + | MergeShardsResourceInUseException + -- ^ /Code 400/ + + | MergeShardsResourceNotFoundException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + + + diff --git a/src/Aws/Kinesis/Commands/PutRecord.hs b/src/Aws/Kinesis/Commands/PutRecord.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/PutRecord.hs @@ -0,0 +1,168 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | +-- Module: Aws.Kinesis.Commands.PutRecord +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- This operation puts a data record into an Amazon Kinesis stream from a +-- producer. This operation must be called to send data from the producer into +-- the Amazon Kinesis stream for real-time ingestion and subsequent processing. +-- The PutRecord operation requires the name of the stream that captures, +-- stores, and transports the data; a partition key; and the data blob itself. +-- The data blob could be a segment from a log file, geographic/location data, +-- website clickstream data, or any other data type. +-- +-- The partition key is used to distribute data across shards. Amazon Kinesis +-- segregates the data records that belong to a data stream into multiple +-- shards, using the partition key associated with each data record to +-- determine which shard a given data record belongs to. +-- +-- Partition keys are Unicode strings, with a maximum length limit of 256 +-- bytes. An MD5 hash function is used to map partition keys to 128-bit integer +-- values and to map associated data records to shards using the hash key +-- ranges of the shards. You can override hashing the partition key to +-- determine the shard by explicitly specifying a hash value using the +-- ExplicitHashKey parameter. For more information, see the Amazon Kinesis +-- Developer Guide. +-- +-- PutRecord returns the shard ID of where the data record was placed and the +-- sequence number that was assigned to the data record. +-- +-- Sequence numbers generally increase over time. To guarantee strictly +-- increasing ordering, use the SequenceNumberForOrdering parameter. For more +-- information, see the Amazon Kinesis Developer Guide. +-- +-- If a PutRecord request cannot be processed because of insufficient +-- provisioned throughput on the shard involved in the request, PutRecord +-- throws ProvisionedThroughputExceededException. +-- +-- Data records are accessible for only 24 hours from the time that they are +-- added to an Amazon Kinesis stream. +-- +-- +-- +module Aws.Kinesis.Commands.PutRecord +( PutRecord(..) +, PutRecordResponse(..) +, PutRecordExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Types +import Aws.Kinesis.Core + +import Control.Applicative + +import Data.Aeson +import Data.ByteString as B +import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString.Base64 as B64 +import qualified Data.Text.Encoding as T +import Data.Typeable + +putRecordAction :: KinesisAction +putRecordAction = KinesisPutRecord + +data PutRecord = PutRecord + { putRecordData :: !B.ByteString + -- ^ The data blob to put into the record. The maximum size of the data + -- blob is 50 kilobytes (KB) + + , putRecordExplicitHashKey :: !(Maybe PartitionHash) + -- ^ The hash value used to explicitly determine the shard the data record + -- is assigned to by overriding the partition key hash. + -- + -- FIXME the specification is rather vague about the precise encoding + -- of this value. The default is to compute it as an MD5 hash of + -- the partition key. The API reference describes it as an Int128. + -- However, it is not clear how the result of the hash function is + -- encoded (big-endian or small endian, word size?) and how it is + -- serialized to text, which is the type in the JSON serialization. + + , putRecordPartitionKey :: !PartitionKey + -- ^ Determines which shard in the stream the data record is assigned to. + + , putRecordSequenceNumberForOrdering :: !(Maybe SequenceNumber) + -- ^ Guarantees strictly increasing sequence numbers, for puts from the + -- same client and to the same partition key. Usage: set the + -- SequenceNumberForOrdering of record n to the sequence number of record + -- n-1 (as returned in the PutRecordResult when putting record n-1). If + -- this parameter is not set, records will be coarsely ordered based on + -- arrival time. + + , putRecordStreamName :: !StreamName + -- ^ The name of the stream to put the data record into. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON PutRecord where + toJSON PutRecord{..} = object + [ "Data" .= T.decodeUtf8 (B64.encode putRecordData) + , "ExplicitHashKey" .= putRecordExplicitHashKey + , "PartitionKey" .= putRecordPartitionKey + , "SequenceNumberForOrdering" .= putRecordSequenceNumberForOrdering + , "StreamName" .= putRecordStreamName + ] + +data PutRecordResponse = PutRecordResponse + { putRecordResSequenceNumber :: !SequenceNumber + -- ^ The sequence number identifier that was assigned to the put data + -- record. The sequence number for the record is unique across all records + -- in the stream. A sequence number is the identifier associated with every + -- record put into the stream. + + , putRecordResShardId :: !ShardId + -- ^ The shard ID of the shard where the data record was placed. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance FromJSON PutRecordResponse where + parseJSON = withObject "PutRecordResponse" $ \o -> PutRecordResponse + <$> o .: "SequenceNumber" + <*> o .: "ShardId" + +instance ResponseConsumer r PutRecordResponse where + type ResponseMetadata PutRecordResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance SignQuery PutRecord where + type ServiceConfiguration PutRecord = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = putRecordAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction PutRecord PutRecordResponse + +instance AsMemoryResponse PutRecordResponse where + type MemoryResponse PutRecordResponse = PutRecordResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data PutRecordExceptions + = PutRecordInvalidArgumentException + -- ^ /Code 400/ + + | PutRecordProvisionedThroughputExceededException + -- ^ /Code 400/ + + | PutRecordResourceNotFoundException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + diff --git a/src/Aws/Kinesis/Commands/SplitShard.hs b/src/Aws/Kinesis/Commands/SplitShard.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Commands/SplitShard.hs @@ -0,0 +1,152 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | +-- Module: Aws.Sns.Commands.SplitShard +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-03-31/ +-- This operation splits a shard into two new shards in the stream, to increase +-- the stream's capacity to ingest and transport data. SplitShard is called +-- when there is a need to increase the overall capacity of stream because of +-- an expected increase in the volume of data records being ingested. +-- +-- SplitShard can also be used when a given shard appears to be approaching its +-- maximum utilization, for example, when the set of producers sending data +-- into the specific shard are suddenly sending more than previously +-- anticipated. You can also call the SplitShard operation to increase stream +-- capacity, so that more Amazon Kinesis applications can simultaneously read +-- data from the stream for real-time processing. +-- +-- The SplitShard operation requires that you specify the shard to be split and +-- the new hash key, which is the position in the shard where the shard gets +-- split in two. In many cases, the new hash key might simply be the average of +-- the beginning and ending hash key, but it can be any hash key value in the +-- range being mapped into the shard. For more information about splitting +-- shards, see the Amazon Kinesis Developer Guide. +-- +-- You can use the DescribeStream operation to determine the shard ID and hash +-- key values for the ShardToSplit and NewStartingHashKey parameters that are +-- specified in the SplitShard request. +-- +-- SplitShard is an asynchronous operation. Upon receiving a SplitShard +-- request, Amazon Kinesis immediately returns a response and sets the stream +-- status to UPDATING. After the operation is completed, Amazon Kinesis sets +-- the stream status to ACTIVE. Read and write operations continue to work +-- while the stream is in the UPDATING state. +-- +-- You can use DescribeStream to check the status of the stream, which is +-- returned in StreamStatus. If the stream is in the ACTIVE state, you can call +-- SplitShard. If a stream is in CREATING or UPDATING or DELETING states, then +-- Amazon Kinesis returns a ResourceInUseException. +-- +-- If the specified stream does not exist, Amazon Kinesis returns a +-- ResourceNotFoundException. If you try to create more shards than are +-- authorized for your account, you receive a LimitExceededException. +-- +-- Note: The default limit for an AWS account is 10 shards per stream. If you +-- need to create a stream with more than 10 shards, contact AWS Support to +-- increase the limit on your account. +-- +-- If you try to operate on too many streams in parallel using CreateStream, +-- DeleteStream, MergeShards or SplitShard, you will receive a +-- LimitExceededException. +-- +-- SplitShard has limit of 5 transactions per second per account. +-- +-- +-- +module Aws.Kinesis.Commands.SplitShard +( SplitShard(..) +, SplitShardResponse(..) +, SplitShardExceptions(..) +) where + +import Aws.Core +import Aws.Kinesis.Core +import Aws.Kinesis.Types + +import Data.Aeson +import qualified Data.ByteString.Lazy as LB +import Data.Typeable + +splitShardAction :: KinesisAction +splitShardAction = KinesisSplitShard + +data SplitShard = SplitShard + { splitShardNewStartingHashKey :: !PartitionHash + -- ^ A hash key value for the starting hash key of one of the child shards + -- created by the split. The hash key range for a given shard constitutes a + -- set of ordered contiguous positive integers. The value for + -- NewStartingHashKey must be in the range of hash keys being mapped into + -- the shard. The NewStartingHashKey hash key value and all higher hash key + -- values in hash key range are distributed to one of the child shards. All + -- the lower hash key values in the range are distributed to the other + -- child shard. + + , splitShardShardToSplit :: !ShardId + -- ^ The shard ID of the shard to split. + + , splitShardStreamName :: !StreamName + -- ^ The name of the stream for the shard split. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON SplitShard where + toJSON SplitShard{..} = object + [ "NewStartingHashKey" .= splitShardNewStartingHashKey + , "ShardToSplit" .= splitShardShardToSplit + , "StreamName" .= splitShardStreamName + ] + +data SplitShardResponse = SplitShardResponse + deriving (Show, Read, Eq, Ord, Typeable) + +instance ResponseConsumer r SplitShardResponse where + type ResponseMetadata SplitShardResponse = KinesisMetadata + responseConsumer _ = kinesisResponseConsumer + +instance FromJSON SplitShardResponse where + parseJSON _ = return SplitShardResponse + +instance SignQuery SplitShard where + type ServiceConfiguration SplitShard = KinesisConfiguration + signQuery cmd = kinesisSignQuery KinesisQuery + { kinesisQueryAction = splitShardAction + , kinesisQueryBody = Just $ LB.toStrict $ encode cmd + } + +instance Transaction SplitShard SplitShardResponse + +instance AsMemoryResponse SplitShardResponse where + type MemoryResponse SplitShardResponse = SplitShardResponse + loadToMemory = return + +-- -------------------------------------------------------------------------- -- +-- Exceptions +-- +-- Currently not used for requests. It's included for future usage +-- and as reference. + +data SplitShardExceptions + = SplitShardInvalidArgumentException + -- ^ /Code 400/ + + | SplitShardLimitExceededException + -- ^ /Code 400/ + + | SplitShardResourceInUseException + -- ^ /Code 400/ + + | SplitShardResourceNotFoundException + -- ^ /Code 400/ + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + diff --git a/src/Aws/Kinesis/Core.hs b/src/Aws/Kinesis/Core.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Core.hs @@ -0,0 +1,510 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE CPP #-} + +-- | +-- Module: Aws.Kinesis.Core +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- +-- +module Aws.Kinesis.Core +( + KinesisVersion(..) + +-- * Kinesis Client Configuration +, KinesisConfiguration(..) + +-- * Kinesis Client Metadata +, KinesisMetadata(..) + +-- * Kinesis Exceptions +, KinesisErrorResponse(..) + +-- * Internal + +-- ** Kinesis Actions +, KinesisAction(..) +, kinesisActionToText +, parseKinesisAction + +-- ** Kinesis AWS Service Endpoints +, kinesisServiceEndpoint + +-- ** Kinesis Queries +, KinesisQuery(..) +, kinesisSignQuery + +-- ** Kinesis Responses +, kinesisResponseConsumer +, jsonResponseConsumer + +-- ** Kinesis Errors and Common Parameters +, KinesisError(..) +, KinesisCommonParameters(..) +, KinesisCommonError(..) +) where + +import Aws.Core +import Aws.General +import Aws.SignatureV4 + +import qualified Blaze.ByteString.Builder as BB + +import Control.Applicative +import Control.Exception +import Control.Monad.IO.Class +import Control.Monad.Trans.Resource (throwM) + +import Data.Aeson +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString.Char8 as B8 +import Data.Conduit (($$+-)) +import Data.Conduit.Binary (sinkLbs) +import Data.IORef +import Data.Maybe +import Data.Monoid +import Data.String +import Data.Time.Clock +import Data.Typeable +import qualified Data.Text as T +import qualified Data.Text.Encoding as T + +import qualified Network.HTTP.Types as HTTP +import qualified Network.HTTP.Conduit as HTTP + +import qualified Test.QuickCheck as Q + +import qualified Text.Parser.Char as P +import Text.Parser.Combinators (()) + +data KinesisVersion + = KinesisVersion_2013_12_02 + +kinesisTargetVersion :: IsString a => a +kinesisTargetVersion = "Kinesis_20131202" + +-- -------------------------------------------------------------------------- -- +-- Kinesis Actions + +data KinesisAction + = KinesisCreateStream + | KinesisDeleteStream + | KinesisDescribeStream + | KinesisGetRecords + | KinesisGetShardIterator + | KinesisListStreams + | KinesisMergeShards + | KinesisPutRecord + | KinesisSplitShard + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + +kinesisActionToText :: IsString a => KinesisAction -> a +kinesisActionToText KinesisCreateStream = "CreateStream" +kinesisActionToText KinesisDeleteStream = "DeleteStream" +kinesisActionToText KinesisDescribeStream = "DescribeStream" +kinesisActionToText KinesisGetRecords = "GetRecords" +kinesisActionToText KinesisGetShardIterator = "GetShardIterator" +kinesisActionToText KinesisListStreams = "ListStreams" +kinesisActionToText KinesisMergeShards = "MergeShards" +kinesisActionToText KinesisPutRecord = "PutRecord" +kinesisActionToText KinesisSplitShard = "SplitShard" + +parseKinesisAction :: P.CharParsing m => m KinesisAction +parseKinesisAction = + KinesisCreateStream <$ P.text "CreateStream" + <|> KinesisDeleteStream <$ P.text "DeleteStream" + <|> KinesisDescribeStream <$ P.text "DescribeStream" + <|> KinesisGetRecords <$ P.text "GetRecords" + <|> KinesisGetShardIterator <$ P.text "GetShardIterator" + <|> KinesisListStreams <$ P.text "ListStreams" + <|> KinesisMergeShards <$ P.text "MergeShards" + <|> KinesisPutRecord <$ P.text "PutRecord" + <|> KinesisSplitShard <$ P.text "SplitShard" + "KinesisAction" + +instance AwsType KinesisAction where + toText = kinesisActionToText + parse = parseKinesisAction + +instance Q.Arbitrary KinesisAction where + arbitrary = Q.elements [minBound..maxBound] + +kinesisTargetHeader :: KinesisAction -> HTTP.Header +kinesisTargetHeader a = ("X-Amz-Target", kinesisTargetVersion <> "." <> toText a) + +-- -------------------------------------------------------------------------- -- +-- Kinesis AWS Service Endpoints + +-- | Kinesis Endpoints as specified in AWS General API version 0.1 +-- +-- +-- +kinesisServiceEndpoint :: Region -> B8.ByteString +kinesisServiceEndpoint ApNortheast1 = "kinesis.ap-northeast-1.amazonaws.com" +kinesisServiceEndpoint ApSoutheast1 = "kinesis.ap-southeast-1.amazonaws.com" +kinesisServiceEndpoint ApSoutheast2 = "kinesis.ap-southeast-2.amazonaws.com" +kinesisServiceEndpoint EuWest1 = "kinesis.eu-west-1.amazonaws.com" +kinesisServiceEndpoint UsEast1 = "kinesis.us-east-1.amazonaws.com" +kinesisServiceEndpoint UsWest2 = "kinesis.us-west-2.amazonaws.com" +kinesisServiceEndpoint r = error $ "Aws.Kinesis.Core.kinesisServiceEndpoint: unsupported region " <> show r -- FIXME + +-- -------------------------------------------------------------------------- -- +-- Kinesis Metadata + +data KinesisMetadata = KinesisMetadata + { kinesisMAmzId2 :: Maybe T.Text + , kinesisMRequestId :: Maybe T.Text + } + deriving (Show) + +instance Loggable KinesisMetadata where + toLogText (KinesisMetadata rid id2) = + "Kinesis: request ID=" <> fromMaybe "" rid + <> ", x-amz-id-2=" <> fromMaybe "" id2 + +instance Monoid KinesisMetadata where + mempty = KinesisMetadata Nothing Nothing + KinesisMetadata id1 r1 `mappend` KinesisMetadata id2 r2 = KinesisMetadata (id1 <|> id2) (r1 <|> r2) + +-- -------------------------------------------------------------------------- -- +-- Kinesis Configuration + +data KinesisConfiguration qt = KinesisConfiguration + { kinesisConfRegion :: Region + } + deriving (Show) + +-- -------------------------------------------------------------------------- -- +-- Kinesis Query + +data KinesisQuery = KinesisQuery + { kinesisQueryAction :: !KinesisAction + , kinesisQueryBody :: !(Maybe B.ByteString) + } + deriving (Show, Eq) + +-- | Creates a signed query. +-- +-- Uses AWS Signature V4. All requests are POST requests +-- with the signature placed in an HTTP header +-- +kinesisSignQuery :: KinesisQuery -> KinesisConfiguration qt -> SignatureData -> SignedQuery +kinesisSignQuery query conf sigData = SignedQuery + { sqMethod = Post + , sqProtocol = HTTPS + , sqHost = host + , sqPort = port + , sqPath = BB.toByteString $ HTTP.encodePathSegments path + , sqQuery = reqQuery + , sqDate = Nothing + , sqAuthorization = authorization + , sqContentType = contentType + , sqContentMd5 = Nothing + , sqAmzHeaders = amzHeaders + , sqOtherHeaders = [] -- we put everything into amzHeaders + , sqBody = HTTP.RequestBodyBS <$> body + , sqStringToSign = mempty -- Let me know if you really need this... + } + where + path = [] + reqQuery = [] + host = kinesisServiceEndpoint $ kinesisConfRegion conf + headers = [("host", host), kinesisTargetHeader (kinesisQueryAction query)] + port = 443 + contentType = Just "application/x-amz-json-1.1" + body = kinesisQueryBody query + + amzHeaders = filter ((/= "Authorization") . fst) sig + authorization = return <$> lookup "authorization" sig + sig = either error id $ signPostRequest + (cred2cred $ signatureCredentials sigData) + (kinesisConfRegion conf) + ServiceNamespaceKinesis + (signatureTime sigData) + "POST" + path + reqQuery + headers + (fromMaybe "" body) + +#if MIN_VERSION_aws(0,9,2) + cred2cred (Credentials a b c _) = SignatureV4Credentials a b c +#else + cred2cred (Credentials a b c) = SignatureV4Credentials a b c +#endif + +-- -------------------------------------------------------------------------- -- +-- Kinesis Response Consumer + +-- | Create a complete 'HTTPResponseConsumer' for response types with an FromJSON instance +-- +jsonResponseConsumer + :: FromJSON a + => HTTPResponseConsumer a +jsonResponseConsumer res = do + doc <- HTTP.responseBody res $$+- sinkLbs + case eitherDecode (if doc == mempty then "{}" else doc) of + Left err -> throwM . KinesisResponseJsonError $ T.pack err + Right v -> return v + +kinesisResponseConsumer + :: FromJSON a + => IORef KinesisMetadata + -> HTTPResponseConsumer a +kinesisResponseConsumer metadata resp = do + + let headerString = fmap T.decodeUtf8 . flip lookup (HTTP.responseHeaders resp) + amzId2 = headerString "x-amz-id-2" + requestId = headerString "x-amz-request-id" + m = KinesisMetadata + { kinesisMAmzId2 = amzId2 + , kinesisMRequestId = requestId + } + + liftIO $ tellMetadataRef metadata m + + if HTTP.responseStatus resp >= HTTP.status400 + then errorResponseConsumer resp + else jsonResponseConsumer resp + +-- | Parse Error Responses +-- +errorResponseConsumer :: HTTPResponseConsumer a +errorResponseConsumer resp = do + doc <- HTTP.responseBody resp $$+- sinkLbs + if HTTP.responseStatus resp == HTTP.status400 + then kinesisError doc + else throwM KinesisOtherError + { kinesisOtherErrorStatus = HTTP.responseStatus resp + , kinesisOtherErrorMessage = T.decodeUtf8 $ LB.toStrict doc + } + where + kinesisError doc = case eitherDecode doc of + Left e -> throwM . KinesisResponseJsonError $ T.pack e + Right a -> do + liftIO $ print doc + throwM (a :: KinesisErrorResponse) + +-- -------------------------------------------------------------------------- -- +-- Kinesis Errors + +-- | TODO integrate typed errors +-- +data KinesisError a + = KinesisErrorCommon KinesisCommonError + | KinesisErrorCommand a + deriving (Show, Read, Eq, Ord, Typeable) + +-- | All Kinesis exceptions have HTTP status code 400 and include +-- a JSON body with an exception type property and a short message. +-- +data KinesisErrorResponse + = KinesisErrorResponse + { kinesisErrorCode :: !T.Text + , kinesisErrorMessage :: !T.Text + } + | KinesisResponseJsonError T.Text + | KinesisOtherError + { kinesisOtherErrorStatus :: !HTTP.Status + , kinesisOtherErrorMessage :: !T.Text + } + deriving (Show, Eq, Ord, Typeable) + +instance Exception KinesisErrorResponse + +-- | This instance captures only the 'KinesisErrorResponse' constructor +-- +instance FromJSON KinesisErrorResponse where + parseJSON = withObject "KinesisErrorResponse" $ \o -> KinesisErrorResponse + <$> o .: "__type" + <*> o .: "message" + +-- | Common Kinesis Errors +-- +-- +-- +-- TODO add function to provide info about the error (content of haddock comments) +-- +data KinesisCommonError + + -- | The request signature does not conform to AWS standards. + -- + -- /Code 400/ + -- + = ErrorIncompleteSignature + + -- | The request processing has failed because of an unknown error, + -- exception or failure. + -- + -- /Code 500/ + -- + | ErrorInternalFailure + + -- | The action or operation requested is invalid. Verify that the action + -- is typed correctly. + -- + -- /Code 400/ + -- + | ErrorInvalidAction + + -- | The X.509 certificate or AWS access key ID provided does not exist in + -- our records. + -- + -- /Code 403/ + -- + | ErrorInvalidClientTokenId + + -- | Parameters that must not be used together were used together. + -- + -- /Code 400/ + -- + | ErrorInvalidParameterCombination + + -- | An invalid or out-of-range value was supplied for the input parameter. + -- + -- /Code 400/ + -- + | ErrorInvalidParameterValue + + -- | The AWS query string is malformed or does not adhere to AWS standards. + -- + -- /Code 400/ + -- + | ErrorInvalidQueryParamter + + -- | The query string contains a syntax error. + -- + -- /Code 404/ + -- + | ErrorMalformedQueryString + + -- | The request is missing an action or a required parameter. + -- + -- /Code 400/ + -- + | ErrorMissingAction + + -- | The request must contain either a valid (registered) AWS access key ID + -- or X.509 certificate. + -- + -- /Code 403/ + -- + | ErrorMissingAuthenticationToken + + -- | A required parameter for the specified action is not supplied. + -- + -- /Code 400/ + -- + | ErrorMissingParameter + + -- | The AWS access key ID needs a subscription for the service. + -- + -- /Code 403/ + -- + | ErrorOptInRequired + + -- | The request reached the service more than 15 minutes after the date + -- stamp on the request or more than 15 minutes after the request + -- expiration date (such as for pre-signed URLs), or the date stamp on the + -- request is more than 15 minutes in the future. + -- + -- /Code 400/ + -- + | ErrorRequestExpired + + -- | The request has failed due to a temporary failure of the server. + -- + -- /Code 503/ + -- + | ErrorServiceUnavailable + + -- | The request was denied due to request throttling. + -- + -- /Code 400/ + -- + | ErrorThrottling + + -- | The input fails to satisfy the constraints specified by an AWS + -- service. + -- + -- /Code 400/ + -- + | ErrorValidationError + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + +-- -------------------------------------------------------------------------- -- +-- Common Parameters + +-- | Common Kinesis Parameters +-- +-- +-- +-- The user of this API hardy needs to deal with the data type directly. +-- +-- This API supports only signature version 4 with signature method @AWS4-HMAC-SHA256@. +-- +-- /This is not currently used for computing the requests, but serves as +-- documentation and reference for the implementation of yet missing features./ +-- +data KinesisCommonParameters = KinesisCommonParameters + { kinesisAction :: !KinesisAction + -- ^ The action to be performed. + + , kinesisAuthParams :: !(Maybe ()) -- + -- ^ The parameters that are required to authenticate a Conditional request. Contains: + -- + -- * AWSAccessKeyID + -- + -- * SignatureVersion + -- + -- * Timestamp + -- + -- * Signature + + , kinesisAWSAccessKeyId :: !B8.ByteString + -- ^ The access key ID that corresponds to the secret access key that you used to sign the request. + + , kinesisExpires :: !UTCTime + -- ^ The date and time when the request signature expires. + -- Precisely one of snsExpires or snsTimestamp must be present. + -- + -- format: @YYYY-MM-DDThh:mm:ssZ@ (ISO 8601) + + , kinesisTimestamp :: !UTCTime + -- ^ The date and time of the request. + -- Precisely one of snsExpires or snsTimestamp must be present. + -- + -- format: @YYYY-MM-DDThh:mm:ssZ@ (ISO 8601) + + , kinesisSecurityToken :: () -- !(Maybe SecurityToken) + -- ^ TODO + -- + -- The temporary security token that was obtained through a call to AWS + -- Security Token Service. For a list of services that support AWS Security + -- Token Service, go to Using Temporary Security Credentials to Access AWS + -- in Using Temporary Security Credentials. + + , kinesisSignature :: !Signature + -- ^ The digital signature that you created for the request. For + -- information about generating a signature, go to the service's developer + -- documentation. + + , kinesisSignatureMethod :: !SignatureMethod + -- ^ The hash algorithm that you used to create the request signature. + -- + -- Valid Values: @HmacSHA256@ | @HmacSHA1@ + + , kinesisSignatureVersion :: !SignatureVersion + -- ^ The signature version you use to sign the request. Set this to the value that is recommended for your service. + + , kinesisVersion :: KinesisVersion + -- ^ The API version that the request is written for. + } + diff --git a/src/Aws/Kinesis/Types.hs b/src/Aws/Kinesis/Types.hs new file mode 100644 --- /dev/null +++ b/src/Aws/Kinesis/Types.hs @@ -0,0 +1,500 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveDataTypeable #-} + +-- | +-- Module: Aws.Kinesis.Types +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- /API Version: 2013-12-02/ +-- +-- The Amazon Kinesis Service API Reference API contains several data types +-- that various actions use. +-- +-- +-- +module Aws.Kinesis.Types +( StreamName +, streamName +, streamNameText + +, ShardId +, SequenceNumber + +, PartitionHash +, partitionHash +, partitionHashInteger + +, PartitionKey +, partitionKey +, partitionKeyText + +, ShardIterator +, ShardIteratorType(..) +, Record(..) +, StreamDescription(..) +, StreamStatus(..) +, Shard(..) +) where + +import Aws.General + +import Control.Applicative + +import Data.Aeson +import qualified Data.ByteString as B +import qualified Data.ByteString.Base64 as B64 +import Data.Monoid +import Data.String +import qualified Data.Text as T +import qualified Data.Text.Encoding as T +import qualified Data.Text.Read as T +import Data.Typeable + +import Test.QuickCheck +import Test.QuickCheck.Instances () + +-- -------------------------------------------------------------------------- -- +-- Internal Utils + +sshow :: (Show a, IsString b) => a -> b +sshow = fromString . show + +tryM :: Monad m => Either T.Text a -> m a +tryM = either (fail . T.unpack) return + +-- -------------------------------------------------------------------------- -- +-- | Stream Name +-- +-- The stream name is scoped to the AWS account used by the application that +-- creates the stream. It is also scoped by region. That is, two streams in two +-- different AWS accounts can have the same name, and two streams in the same +-- AWS account, but in two different regions, can have the same name. +-- +-- Length constraints: Minimum length of 1. Maximum length of 128. +-- +-- Error responses from AWS Kinesis indicate that stream names must +-- match the following regular expressions: +-- +-- > [a-zA-Z0-9_.-]+ +-- +newtype StreamName = StreamName { streamNameText :: T.Text } + deriving (Show, Read, Eq, Ord, Typeable) + +-- | Smart Constructor for 'StreamName' that enforces size constraints. +-- +-- For static construction you may also use the 'IsString' instance +-- that throws an 'error' on invalid stream names. +-- +streamName :: T.Text -> Either T.Text StreamName +streamName t + | T.length t < 1 = Left $ "Illegal StreamName " <> sshow t <> "; StreamName must be of length at least 1" + | T.length t > 128 = Left $ "Illegal StreamName " <> sshow t <> "; StreamName must be of length at most 128" + | otherwise = Right $ StreamName t + +-- | The 'fromString' function of this instance raises and asynchronous 'error' +-- if the argument 'String' is not a legal 'StreamName'. +-- +instance IsString StreamName where + fromString = either (error . T.unpack) id . streamName . T.pack + +instance ToJSON StreamName where + toJSON = toJSON . streamNameText + +instance FromJSON StreamName where + parseJSON = withText "StreamName" $ tryM . streamName + +instance Arbitrary StreamName where + arbitrary = StreamName . T.pack <$> (resize 128 . listOf1 $ elements chars) + where + chars = ['a' .. 'z'] <> ['A' .. 'Z'] <> ['0' .. '9'] <> "_.-" + +-- -------------------------------------------------------------------------- -- +-- | Identifier for a shard as returned by PutRecord. +-- +-- Length constraints: Minimum length of 1. Maximum length of 128. +-- +newtype ShardId = ShardId { shardIdText :: T.Text } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON ShardId where + toJSON = toJSON . shardIdText + +instance FromJSON ShardId where + parseJSON = withText "ShardId" $ return . ShardId + +instance Arbitrary ShardId where + arbitrary = ShardId <$> (resize 128 arbitrary `suchThat` (not . T.null)) + +-- -------------------------------------------------------------------------- -- +-- | Opaque sequence number as returned by PutRecord. +-- +newtype SequenceNumber = SequenceNumber { sequenceNumberText :: T.Text } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON SequenceNumber where + toJSON = toJSON . sequenceNumberText + +instance FromJSON SequenceNumber where + parseJSON = withText "SequenceNumber" $ return . SequenceNumber + +instance Arbitrary SequenceNumber where + arbitrary = SequenceNumber . T.pack . show . getPositive + <$> (arbitrary :: Gen (Positive Integer)) + +-- -------------------------------------------------------------------------- -- +-- | An 128 bit interger that identifies the partition. The default +-- is the MD5 hash of the partition key. +-- +-- FIXME the specification is rather vague about the precise encoding +-- of this value. The default is to compute it as an MD5 hash of +-- the partition key. The API reference describes it as an Int128. +-- However, it is not clear how the result of the hash function is +-- encoded (big-endian or small endian, word size?) and how it is +-- serialized to text, which is the type in the JSON serialization. +-- +-- +-- seems to indicate that the value is actually unsigned. +-- +newtype PartitionHash = PartitionHash { partitionHashInteger :: Integer {- FIXME -} } + deriving (Show, Read, Eq, Ord, Typeable) + +-- | Smart Constructor for 'PartitionHash' that enforces size constraints. +-- +partitionHash :: Integer -> Either T.Text PartitionHash +partitionHash i + | i >= 2^(128 :: Int) = Left $ "partition hash " <> sshow i <> "is out of range; a partition hash must be a 128 bit unsigned integer" + | i < 0 = Left $ "partition hash " <> sshow i <> "is out of range; a partition hash must be a 128 bit unsigned integer" + | otherwise = Right $ PartitionHash i + +instance ToJSON PartitionHash where + toJSON = toJSON . show . partitionHashInteger + +instance FromJSON PartitionHash where + parseJSON = withText "PartitionHash" $ \t -> case T.decimal t of + Right (a, "") -> tryM $ partitionHash a + Right (_, x) -> fail $ "trailing text: " <> T.unpack x + Left e -> fail e + +instance Arbitrary PartitionHash where + arbitrary = PartitionHash . getPositive <$> resize (2^(128 :: Int) - 1) arbitrary + +-- -------------------------------------------------------------------------- -- +-- | PartitionKey +-- +-- Identifies which shard in the stream the data record is assigned to. +-- +-- Partition keys are Unicode strings with a maximum length limit of 256 +-- bytes. Amazon Kinesis uses the partition key as input to a hash function +-- that maps the partition key and associated data to a specific shard. +-- Specifically, an MD5 hash function is used to map partition keys to +-- 128-bit integer values and to map associated data records to shards. As +-- a result of this hashing mechanism, all data records with the same +-- partition key will map to the same shard within the stream. +-- +-- Length constraints: Minimum length of 1. Maximum length of 256. +-- +newtype PartitionKey = PartitionKey { partitionKeyText :: T.Text } + deriving (Show, Read, Eq, Ord, IsString, Typeable) + +partitionKey :: T.Text -> Either T.Text PartitionKey +partitionKey t + | T.length t < 1 = Left $ "Illegal PartitionKey " <> sshow t <> "; PartitionKey must be of length at least 1" + | T.length t > 256 = Left $ "Illegal PartitionKey " <> sshow t <> "; PartitionKey must be of length at most 256" + | otherwise = Right $ PartitionKey t + +instance ToJSON PartitionKey where + toJSON = toJSON . partitionKeyText + +instance FromJSON PartitionKey where + parseJSON = withText "PartitionKey" $ tryM . partitionKey + +instance Arbitrary PartitionKey where + arbitrary = PartitionKey + <$> (resize 256 arbitrary `suchThat` (not . T.null)) + +-- -------------------------------------------------------------------------- -- +-- | Iterator for records within a shard. +-- +-- Length constraints: Minimum length of 1. Maximum length of 512. +-- +newtype ShardIterator = ShardIterator { shardIteratorText :: T.Text } + deriving (Show, Read, Eq, Ord, Typeable) + +shardIterator :: T.Text -> Either T.Text ShardIterator +shardIterator t + | T.length t < 1 = Left $ "Illegal ShardIterator " <> sshow t <> "; ShardIterator must be of length at least 1" + | T.length t > 512 = Left $ "Illegal ShardIterator " <> sshow t <> "; ShardIterator must be of length at most 512" + | otherwise = Right $ ShardIterator t + +instance ToJSON ShardIterator where + toJSON = toJSON . shardIteratorText + +instance FromJSON ShardIterator where + parseJSON = withText "ShardIterator" $ tryM . shardIterator + +instance Arbitrary ShardIterator where + arbitrary = ShardIterator + <$> (resize 512 arbitrary `suchThat` (not . T.null)) + +-- -------------------------------------------------------------------------- -- +-- | ShardIteratorType +-- +-- Determines how the shard iterator is used to start reading data records from +-- the shard. +-- +data ShardIteratorType + = AtSequenceNumber + -- ^ Start reading exactly from the position denoted by a specific sequence + -- number. + + | AfterSequenceNumber + -- ^ Start reading right after the position denoted by a specific sequence + -- number. + + | TrimHorizon + -- ^ Start reading at the last untrimmed record in the shard in the system, + -- which is the oldest data record in the shard. + + | Latest + -- ^ Start reading just after the most recent record in the shard, so that + -- you always read the most recent data in the shard. + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + +instance ToJSON ShardIteratorType where + toJSON AtSequenceNumber = "AT_SEQUENCE_NUMBER" + toJSON AfterSequenceNumber = "AFTER_SEQUENCE_NUMBER" + toJSON TrimHorizon = "TRIM_HORIZON" + toJSON Latest = "LATEST" + +instance FromJSON ShardIteratorType where + parseJSON = withText "SharedIteratorType" $ \t -> case t of + "AT_SEQUENCE_NUMBER" -> return AtSequenceNumber + "AFTER_SEQUENCE_NUMBER" -> return AfterSequenceNumber + "TRIM_HORIZON" -> return TrimHorizon + "LATEST" -> return Latest + e -> fail $ "unexpected value for SharedIteratorType: " <> T.unpack e + +instance Arbitrary ShardIteratorType where + arbitrary = elements [minBound .. maxBound] + +-- -------------------------------------------------------------------------- -- +-- | Record +-- +-- The unit of data of the Amazon Kinesis stream, which is composed of a +-- sequence number, a partition key, and a data blob. +-- +-- +-- +data Record = Record + { recordData :: !B.ByteString + -- ^ The data blob. The data in the blob is both opaque and immutable to + -- the Amazon Kinesis service, which does not inspect, interpret, or change + -- the data in the blob in any way. The maximum size of the data blob (the + -- payload after Base64-decoding) is 50 kilobytes (KB) + -- + -- Length constraints: Minimum length of 0. Maximum length of 51200. + -- + -- Note that the size constraint is not currently enforced in the code. + + , recordPartitionKey :: !PartitionKey + -- ^ Identifies which shard in the stream the data record is assigned to. + + , recordSequenceNumber :: !SequenceNumber + -- ^ The unique identifier for the record in the Amazon Kinesis stream. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON Record where + toJSON Record{..} = object + [ "Data" .= T.decodeUtf8 (B64.encode recordData) + , "PartitionKey" .= recordPartitionKey + , "SequenceNumber" .= recordSequenceNumber + ] + +instance FromJSON Record where + parseJSON = withObject "Record" $ \o -> Record + <$> (from64 =<< o .: "Data") + <*> o .: "PartitionKey" + <*> o .: "SequenceNumber" + where + from64 x = case B64.decode (T.encodeUtf8 x) of + Left e -> fail $ "failed to decode base64 encoded data: " <> e + Right a -> return a + +instance Arbitrary Record where + arbitrary = Record + <$> (arbitrary `suchThat` ((<= 51200) . B.length)) + <*> arbitrary + <*> arbitrary + +-- -------------------------------------------------------------------------- -- +-- Stream Description +-- +-- Represents the output of a DescribeStream operation. +-- +-- +-- +data StreamDescription = StreamDescription + { streamDescriptionHasMoreShards :: !Bool + -- ^ If set to true there are more shards in the stream available to + -- describe. + + , streamDescriptionShards :: ![Shard] + -- ^ The shards that comprise the stream. + + , streamDescriptionStreamARN :: !Arn + -- ^The Amazon Resource Name (ARN) for the stream being described. + + , streamDescriptionStreamName :: !StreamName + -- ^ The name of the stream being described. + + , streamDescriptionStreamStatus :: !StreamStatus + -- ^ The current status of the stream being described. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON StreamDescription where + toJSON StreamDescription{..} = object + [ "HasMoreShards" .= streamDescriptionHasMoreShards + , "Shards" .= streamDescriptionShards + , "StreamARN" .= streamDescriptionStreamARN + , "StreamName" .= streamDescriptionStreamName + , "StreamStatus" .= streamDescriptionStreamStatus + ] + +instance FromJSON StreamDescription where + parseJSON = withObject "StreamDescription" $ \o -> StreamDescription + <$> o .: "HasMoreShards" + <*> o .: "Shards" + <*> o .: "StreamARN" + <*> o .: "StreamName" + <*> o .: "StreamStatus" + +instance Arbitrary StreamDescription where + arbitrary = StreamDescription + <$> arbitrary + <*> arbitrary + <*> arbitrary + <*> arbitrary + <*> arbitrary + +-- | Stream Status +-- +-- +-- +data StreamStatus + = StreamStatusCreating + -- ^ The stream is being created. Upon receiving a CreateStream request, + -- Amazon Kinesis immediately returns and sets StreamStatus to CREATING. + + | StreamStatusDeleting + -- ^ The stream is being deleted. After a DeleteStream request, the + -- specified stream is in the DELETING state until Amazon Kinesis completes + -- the deletion. + + | StreamStatusActive + -- ^ The stream exists and is ready for read and write operations or + -- deletion. You should perform read and write operations only on an ACTIVE + -- stream. + + | StreamStatusUpdating + -- ^ Shards in the stream are being merged or split. Read and write + -- operations continue to work while the stream is in the UPDATING state. + + deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable) + +instance ToJSON StreamStatus where + toJSON StreamStatusCreating = "CREATING" + toJSON StreamStatusDeleting = "DELETING" + toJSON StreamStatusActive = "ACTIVE" + toJSON StreamStatusUpdating = "UPDATING" + +instance FromJSON StreamStatus where + parseJSON = withText "StreamStatus" $ \t -> case t of + "CREATING" -> return StreamStatusCreating + "DELETING" -> return StreamStatusDeleting + "ACTIVE" -> return StreamStatusActive + "UPDATING" -> return StreamStatusUpdating + e -> fail $ "unexpected value for StreamStatus: " <> T.unpack e + +instance Arbitrary StreamStatus where + arbitrary = elements [minBound .. maxBound] + +-- | Shard +-- +-- +-- +data Shard = Shard + { shardAdjacentParentShardId :: !(Maybe ShardId) + -- ^ The shard Id of the shard adjacent to the shard's parent. + + , shardHashKeyRange :: !(PartitionHash, PartitionHash) + -- ^ The (inclusive) range of possible hash key values for the shard, which is a set of + -- ordered contiguous positive integers. + -- + -- + + , shardParentShardId :: !(Maybe ShardId) + -- ^ The shard Id of the shard's parent. + + , shardSequenceNumberRange :: !(SequenceNumber, Maybe SequenceNumber) + -- ^ The (inclusive) range of possible sequence numbers for the shard. + -- + -- Shards that are in the OPEN state have an ending sequence number of 'Nothing'. + -- + -- + + , shardShardId :: ShardId + -- ^ The unique identifier of the shard within the Amazon Kinesis stream. + } + deriving (Show, Read, Eq, Ord, Typeable) + +instance ToJSON Shard where + toJSON Shard{..} = object + [ "AdjacentParentShardId" .= shardAdjacentParentShardId + , "HashKeyRange" .= object + [ "StartingHashKey" .= fst shardHashKeyRange + , "EndingHashKey" .= snd shardHashKeyRange + ] + , "ParentShardId" .= shardParentShardId + , "SequenceNumberRange" .= object + [ "StartingSequenceNumber" .= fst shardSequenceNumberRange + , "EndingSequenceNumber" .= snd shardSequenceNumberRange + ] + , "ShardId" .= shardShardId + ] + +instance FromJSON Shard where + parseJSON = withObject "Shard" $ \o -> Shard + <$> o .:? "AdjacentParentShardId" .!= Nothing + <*> (hashKeyRange =<< o .: "HashKeyRange") + <*> o .:? "ParentShardId" .!= Nothing + <*> (sequenceNumberRange =<< o .: "SequenceNumberRange") + <*> o .: "ShardId" + where + hashKeyRange = withObject "HashKeyRange" $ \o -> (,) + <$> o .: "StartingHashKey" + <*> o .: "EndingHashKey" + sequenceNumberRange = withObject "SequenceNumberRange" $ \o -> (,) + <$> o .: "StartingSequenceNumber" + <*> o .:? "EndingSequenceNumber" .!= Nothing + +instance Arbitrary Shard where + arbitrary = Shard + <$> arbitrary + <*> do + u <- arbitrary + l <- resize (fromIntegral $ partitionHashInteger u) arbitrary + return (l,u) + <*> arbitrary + <*> do + u <- arbitrary + l <- resize (maybe maxBound (read . T.unpack . sequenceNumberText) u) arbitrary + return (l,u) + <*> arbitrary diff --git a/tests/Main.hs b/tests/Main.hs new file mode 100644 --- /dev/null +++ b/tests/Main.hs @@ -0,0 +1,286 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE GADTs #-} + +-- | +-- Module: Main +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- Tests for Haskell Kinesis bindings +-- +module Main +( main +) where + +import Aws +import Aws.Kinesis + +import Control.Error +import Control.Exception +import Control.Monad +import Control.Monad.IO.Class + +import qualified Data.ByteString as B +import qualified Data.List as L +import Data.Monoid +import Data.Proxy +import qualified Data.Text as T + +import Test.Tasty + +import System.Exit +import System.Environment + +import Utils + +defaultStreamName :: StreamName +defaultStreamName = "test-stream" + +-- -------------------------------------------------------------------------- -- +-- Main + +-- | Since these tests generate costs there should be a warning and +-- we also should require an explicit command line argument that expresses +-- the concent of the user. +-- +main :: IO () +main = getArgs >>= runMain + where + runMain args + | "--help" `elem` args || "-h" `elem` args = defaultMain tests + | "--run-with-aws-credentials" `elem` args = + withArgs (filter (/= "--run-with-aws-credentials") args) $ defaultMain tests + | otherwise = putStrLn help >> exitFailure + +help :: String +help = L.intercalate "\n" + [ "" + , "NOTE" + , "" + , "This test suite accesses the AWS account that is associated with" + , "the default credentials from the credential file ~/.aws-keys." + , "" + , "By running the tests in this test-suite costs for usage of AWS" + , "services may incur." + , "" + , "In order to actually excute the tests in this test-suite you must" + , "provide the command line option:" + , "" + , " --run-with-aws-credentials" + , "" + , "When running this test-suite through cabal you may use the following" + , "command:" + , "" + , " cabal test kinesis-tests --test-option=--run-with-aws-credentials" + , "" + ] + +tests :: TestTree +tests = testGroup "Kinesis Tests" + [ test_jsonRoundtrips + , test_createStream + , test_stream1 + ] + +-- -------------------------------------------------------------------------- -- +-- Kinesis Utils + +kinesisConfiguration :: KinesisConfiguration qt +kinesisConfiguration = KinesisConfiguration testRegion + +simpleKinesis + :: (AsMemoryResponse a, Transaction r a, ServiceConfiguration r ~ KinesisConfiguration, MonadIO m) + => r + -> m (MemoryResponse a) +simpleKinesis command = do + c <- baseConfiguration + simpleAws c kinesisConfiguration command + +simpleKinesisT + :: (AsMemoryResponse a, Transaction r a, ServiceConfiguration r ~ KinesisConfiguration, MonadIO m) + => r + -> EitherT T.Text m (MemoryResponse a) +simpleKinesisT = tryT . simpleKinesis + +testStreamName :: StreamName -> StreamName +testStreamName = either (error . T.unpack) id + . streamName . T.take 128 . testData . streamNameText + +-- | +-- +withStream + :: StreamName -- ^ Stream Name + -> Int -- ^ Shard count + -> IO a + -> IO a +withStream stream shardCount = bracket_ createStream deleteStream + where + createStream = simpleKinesis $ CreateStream shardCount stream + deleteStream = void $ simpleKinesis (DeleteStream stream) + +-- | The function 'withResource' from "Tasty" synchronizes the aquired +-- resource through a 'TVar'. We don't need that for a stream. So instead +-- of passing the 'IO StreamName' from 'withResource' we directly pass +-- 'StreamName' to the inner function. +-- +withStreamTest + :: StreamName -- ^ stream name suffix + -> Int -- ^ shard count + -> (StreamName -> TestTree) + -> TestTree +withStreamTest stream shardCount f = withResource createStream deleteStream + $ const (f tstream) + where + createStream = do + void . simpleKinesis $ CreateStream shardCount tstream + return tstream + deleteStream = const . void . simpleKinesis $ DeleteStream tstream + tstream = testStreamName stream + + +-- | Wait for a stream to become active +-- +waitActiveT + :: Int + -- ^ upper bound on the number of seconds to wait. + -- The actual maximal number of seconds is closest smaller + -- power of two. + -> StreamName + -> EitherT T.Text IO StreamDescription +waitActiveT sec stream = retryT maxRetry $ do + DescribeStreamResponse d <- simpleKinesisT + $ DescribeStream Nothing Nothing stream + unless (streamDescriptionStreamStatus d == StreamStatusActive) + $ left "Stream is not active" + return d + where + maxRetry = floor $ logBase 2 (fromIntegral sec :: Double) + +-- -------------------------------------------------------------------------- -- +-- Types + +test_jsonRoundtrips :: TestTree +test_jsonRoundtrips = testGroup "JSON encoding roundtrips" + [ test_jsonRoundtrip (Proxy :: Proxy StreamName) + , test_jsonRoundtrip (Proxy :: Proxy ShardId) + , test_jsonRoundtrip (Proxy :: Proxy SequenceNumber) + , test_jsonRoundtrip (Proxy :: Proxy PartitionHash) + , test_jsonRoundtrip (Proxy :: Proxy PartitionKey) + , test_jsonRoundtrip (Proxy :: Proxy ShardIterator) + , test_jsonRoundtrip (Proxy :: Proxy ShardIteratorType) + , test_jsonRoundtrip (Proxy :: Proxy Record) + , test_jsonRoundtrip (Proxy :: Proxy StreamDescription) + , test_jsonRoundtrip (Proxy :: Proxy StreamStatus) + , test_jsonRoundtrip (Proxy :: Proxy Shard) + ] + +-- -------------------------------------------------------------------------- -- +-- Stream Tests + +test_stream1 :: TestTree +test_stream1 = withStreamTest defaultStreamName 1 $ \stream -> + testGroup "Perform a series of tests on a single stream" + [ eitherTOnceTest0 "list streams" (prop_streamList stream) + , eitherTOnceTest0 "describe stream" (prop_streamDescribe 1 stream) + , eitherTOnceTest2 "put and get stream" (prop_streamPutGet stream) + ] + +prop_streamList :: StreamName -> EitherT T.Text IO () +prop_streamList stream = do + ListStreamsResponse _ streams <- simpleKinesisT $ ListStreams Nothing Nothing + unless (stream `elem` streams) $ + left $ "stream " <> streamNameText stream <> " is not listed" + +prop_streamDescribe + :: Int -- ^ expected number of shards + -> StreamName + -> EitherT T.Text IO () +prop_streamDescribe shardNum stream = do + desc <- waitActiveT 64 stream + + unless (streamDescriptionStreamName desc == stream) + . left $ "unexpected stream name in description: " + <> streamNameText (streamDescriptionStreamName desc) + + let l = length $ streamDescriptionShards desc + unless (l == shardNum) + . left $ "unexpected number of shards in stream description: " <> sshow l + +prop_streamPutGet + :: StreamName + -> B.ByteString -- ^ Message data + -> PartitionKey + -> EitherT T.Text IO () +prop_streamPutGet stream dat key = do + desc <- waitActiveT 64 stream + + let shards = streamDescriptionShards desc + + PutRecordResponse putSeqNr putShard <- simpleKinesisT PutRecord + { putRecordData = dat + , putRecordExplicitHashKey = Nothing + , putRecordPartitionKey = key + , putRecordSequenceNumberForOrdering = Nothing + , putRecordStreamName = stream + } + + let shardIds = map shardShardId shards + unless (putShard `elem` shardIds) . left + $ "unexpected shard id: expected on of " <> sshow shardIds <> "; got " <> sshow putShard + + record <- retryT 5 $ do + GetShardIteratorResponse it <- simpleKinesisT GetShardIterator + { getShardIteratorShardId = putShard + , getShardIteratorShardIteratorType = TrimHorizon + , getShardIteratorStartingSequenceNumber = Nothing + , getShardIteratorStreamName = stream + } + GetRecordsResponse _nextIt records <- simpleKinesisT GetRecords + { getRecordsLimit = Nothing + , getRecordsShardIterator = it + } + case records of + [] -> left "no record found in stream" + [r] -> return r + t -> left $ "unexpected records found in stream: " <> sshow t + + let getData = recordData record + unless (getData == dat) . left + $ "data does not match: expected " <> sshow dat <> "; got " <> sshow getData + + let getSeqNr = recordSequenceNumber record + unless (getSeqNr == putSeqNr) . left + $ "sequence numbers don't match: expected " <> sshow putSeqNr + <> "; got " <> sshow getSeqNr + + let getPartKey = recordPartitionKey record + unless (getPartKey == key) . left + $ "partition keys don't match: expected " <> sshow key + <> "; got " <> sshow getPartKey + +-- -------------------------------------------------------------------------- -- +-- Stream Creation Tests + +test_createStream :: TestTree +test_createStream = testGroup "Stream creation" + [ eitherTOnceTest1 "create list delete" prop_createListDelete + ] + +prop_createListDelete + :: StreamName -- ^ stream name + -> EitherT T.Text IO () +prop_createListDelete stream = do + CreateStreamResponse <- simpleKinesisT $ CreateStream 1 tstream + handleT (\e -> deleteStream >> left e) $ do + ListStreamsResponse _ allStreams <- simpleKinesisT + $ ListStreams Nothing Nothing + unless (tstream `elem` allStreams) + . left $ "stream " <> streamNameText tstream <> " not listed" + deleteStream + where + deleteStream = void $ simpleKinesisT (DeleteStream tstream) + tstream = testStreamName stream + diff --git a/tests/Utils.hs b/tests/Utils.hs new file mode 100644 --- /dev/null +++ b/tests/Utils.hs @@ -0,0 +1,154 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE CPP #-} + +-- | +-- Module: Utils +-- Copyright: Copyright © 2014 AlephCloud Systems, Inc. +-- License: MIT +-- Maintainer: Lars Kuhtz +-- Stability: experimental +-- +-- Utils for Tests for Haskell AWS bindints +-- +module Utils +( +-- * Parameters + testRegion +, testDataPrefix + +-- * General Utils +, sshow +, tryT +, retryT +, testData + +, evalTestT +, evalTestTM +, eitherTOnceTest0 +, eitherTOnceTest1 +, eitherTOnceTest2 + +-- * Generic Tests +, test_jsonRoundtrip +, prop_jsonRoundtrip +) where + +import Aws.General + +import Control.Concurrent (threadDelay) +import Control.Error +import Control.Monad +import Control.Monad.Identity +import Control.Monad.IO.Class + +import Data.Aeson (FromJSON, ToJSON, encode, eitherDecode) +import Data.Monoid +import Data.Proxy +import Data.String +import qualified Data.Text as T +import Data.Typeable + +import Test.QuickCheck.Property +import Test.QuickCheck.Monadic +import Test.Tasty +import Test.Tasty.QuickCheck + +import System.IO + +-- -------------------------------------------------------------------------- -- +-- Static Test parameters +-- +-- TODO make these configurable + +testRegion :: Region +testRegion = UsWest2 + +-- | This prefix is used for the IDs and names of all entities that are +-- created in the AWS account. +-- +testDataPrefix :: IsString a => a +testDataPrefix = "__TEST_AWSHASKELLBINDINGS__" + +-- -------------------------------------------------------------------------- -- +-- General Utils + +tryT :: MonadIO m => IO a -> EitherT T.Text m a +tryT = fmapLT (T.pack . show) . syncIO + +testData :: (IsString a, Monoid a) => a -> a +testData a = testDataPrefix <> a + +retryT :: MonadIO m => Int -> EitherT T.Text m a -> EitherT T.Text m a +retryT i f = go 1 + where + go x + | x >= i = fmapLT (\e -> "error after " <> sshow x <> " retries: " <> e) f + | otherwise = f `catchT` \_ -> do + liftIO $ threadDelay (1000000 * min 60 (2^(x-1))) + go (succ x) + +sshow :: (Show a, IsString b) => a -> b +sshow = fromString . show + +evalTestTM + :: Functor f + => String -- ^ test name + -> f (EitherT T.Text IO a) -- ^ test + -> f (IO Bool) +evalTestTM name = fmap $ + runEitherT >=> \r -> case r of + Left e -> do + hPutStrLn stderr $ "failed to run stream test \"" <> name <> "\": " <> show e + return False + Right _ -> return True + +evalTestT + :: String -- ^ test name + -> EitherT T.Text IO a -- ^ test + -> IO Bool +evalTestT name = runIdentity . evalTestTM name . Identity + +eitherTOnceTest0 + :: String -- ^ test name + -> EitherT T.Text IO a -- ^ test + -> TestTree +eitherTOnceTest0 name = testProperty name . once . ioProperty . evalTestT name + +eitherTOnceTest1 + :: (Arbitrary a, Show a) + => String -- ^ test name + -> (a -> EitherT T.Text IO b) + -> TestTree +eitherTOnceTest1 name test = testProperty name . once $ monadicIO . liftIO + . evalTestTM name test + +eitherTOnceTest2 + :: (Arbitrary a, Show a, Arbitrary b, Show b) + => String -- ^ test name + -> (a -> b -> EitherT T.Text IO c) + -> TestTree +eitherTOnceTest2 name test = testProperty name . once $ \a b -> monadicIO . liftIO + $ (evalTestTM name $ uncurry test) (a, b) + +-- -------------------------------------------------------------------------- -- +-- Generic Tests + +test_jsonRoundtrip + :: forall a . (Eq a, Show a, FromJSON a, ToJSON a, Typeable a, Arbitrary a) + => Proxy a + -> TestTree +test_jsonRoundtrip proxy = testProperty msg (prop_jsonRoundtrip :: a -> Property) + where + msg = "JSON roundtrip for " <> show typ +#if MIN_VERSION_base(4,7,0) + typ = typeRep proxy +#else + typ = typeOf (undefined :: a) +#endif + +prop_jsonRoundtrip :: forall a . (Eq a, Show a, FromJSON a, ToJSON a) => a -> Property +prop_jsonRoundtrip a = either (const $ property False) (\(b :: [a]) -> [a] === b) $ + eitherDecode $ encode [a] +