diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,14 @@ The format is based on [Keep a Changelog][chg] and this project adheres to [Haskell's Package Versioning Policy][pvp] +## `0.4.9` - 2022-02-28 + + - Add `KafkaEvent` type for subscribing Lambda functions to Kafka + topics + - Support `aeson ^>=2.0.0.0` + - Drop official support for GHC 8.0.2 + - Expand hedgehog bounds to include 1.1 to keep ahead of Stack nightly + ## `0.4.8` - 2021-05-07 - Add `ToJSON` instances for `ProxyRequest` types, and test diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,4 +1,5 @@ -[![Build Status](https://travis-ci.org/Nike-Inc/hal.svg?branch=master)](https://travis-ci.org/Nike-Inc/hal) +[![Stack Haskell Builds](https://github.com/Nike-Inc/hal/actions/workflows/haskell_stack.yaml/badge.svg)](https://github.com/Nike-Inc/hal/actions/workflows/haskell_stack.yaml) +[![Cabal Haskell Builds](https://github.com/Nike-Inc/hal/actions/workflows/haskell_cabal.yaml/badge.svg)](https://github.com/Nike-Inc/hal/actions/workflows/haskell_cabal.yaml) # hal @@ -13,8 +14,8 @@ It accepts a handler with the signature `(FromJSON a, ToJSON b) => a -> b`. This runtime guarantees that side-effects cannot occur. -For advanced use cases `mRuntimeWithContext` unlocks the full power of Monad Transformers. -It accepts handlers with the signature `(HasLambdaContext r, MonadCatch m, MonadReader r m, MonadIO m, FromJSON event, ToJSON result) => (event -> m result)` +For advanced use cases `mRuntime` unlocks the full power of Monad Transformers. +It accepts handlers with the signature `(MonadCatch m, MonadIO m, FromJSON event, ToJSON result) => (event -> m result)` This enables users to add caching logic or expose complex environments. With numerous options in between these two, developers can choose the right balance of flexibility vs simplicity. @@ -63,9 +64,8 @@ ```yaml #... -packages: - - '.' - - hal-0.1.2 +extra-deps: + - hal-${DESIRED_VERSION} # ... docker: enable: true @@ -85,10 +85,10 @@ import GHC.Generics (Generic) data IdEvent = IdEvent { input :: String } deriving Generic -instance FromJSON IdEvent where +instance FromJSON IdEvent data IdResult = IdResult { output :: String } deriving Generic -instance ToJSON IdResult where +instance ToJSON IdResult handler :: IdEvent -> IdResult handler IdEvent { input } = IdResult { output = input } @@ -153,7 +153,7 @@ # Create your function package aws cloudformation package \ - --template-file template.yaml + --template-file template.yaml \ --s3-bucket your-existing-bucket > \ deployment_stack.yaml @@ -167,8 +167,8 @@ # Take it for a spin! aws lambda invoke \ --function-name your-function-name \ - --region us-west-2 - --payload '{"input": "foo"}' + --region us-west-2 \ + --payload '{"input": "foo"}' \ output.txt ``` @@ -213,7 +213,7 @@ [aws-sam-cli]: https://github.com/awslabs/aws-sam-cli [Rust Runtime]: https://github.com/awslabs/aws-lambda-rust-runtime [lambda-env]: https://docs.aws.amazon.com/lambda/latest/dg/current-supported-versions.html -[ci]: https://travis-ci.org/Nike-Inc/hal +[ci]: https://github.com/Nike-Inc/hal/actions [stackage]: https://www.stackage.org/ [GHC]: https://www.haskell.org/ghc/download.html [Cabal]: https://www.haskell.org/cabal/download.html diff --git a/hal.cabal b/hal.cabal --- a/hal.cabal +++ b/hal.cabal @@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.34.4. +-- This file has been generated from package.yaml by hpack version 0.33.0. -- -- see: https://github.com/sol/hpack -- --- hash: e76a8db81a6c7a721b849d1efca110b833ece7559f999740e5ac5c2d6cdd36e1 +-- hash: 804d66701a670f6c063313f96b6b5c9754cf0a4a67b7fc4651e11ed8ab059608 name: hal -version: 0.4.8 +version: 0.4.9 synopsis: A runtime environment for Haskell applications running on AWS Lambda. description: This library uniquely supports different types of AWS Lambda Handlers for your needs/comfort with advanced Haskell. Instead of exposing a single function @@ -35,6 +35,7 @@ AWS.Lambda.Context AWS.Lambda.Events.ApiGateway.ProxyRequest AWS.Lambda.Events.ApiGateway.ProxyResponse + AWS.Lambda.Events.Kafka AWS.Lambda.Events.S3 AWS.Lambda.Events.SQS AWS.Lambda.Internal @@ -81,6 +82,7 @@ , http-client , http-types , mtl + , scientific >=0.3.3.0 && <0.4 , text , time , unordered-containers @@ -93,7 +95,11 @@ AWS.Lambda.Events.ApiGateway.ProxyRequest.Gen AWS.Lambda.Events.ApiGateway.ProxyRequest.Gen.Parameters AWS.Lambda.Events.ApiGateway.ProxyRequest.Gen.Resource + AWS.Lambda.Events.ApiGateway.ProxyRequest.Spec AWS.Lambda.Events.ApiGateway.ProxyResponse.Gen + AWS.Lambda.Events.ApiGateway.ProxyResponse.Spec + AWS.Lambda.Events.Kafka.Gen + AWS.Lambda.Events.Kafka.Spec Gen.Header Paths_hal hs-source-dirs: @@ -119,22 +125,23 @@ ScopedTypeVariables ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints -fno-warn-partial-type-signatures -fno-warn-name-shadowing -fwarn-tabs -fwarn-unused-imports -fwarn-missing-signatures -fwarn-incomplete-patterns -threaded -rtsopts -with-rtsopts=-N build-depends: - aeson + aeson >=1.2.0.0 && <1.6 || >=2.0.0.0 && <2.1 , base >=4.7 && <5 , base64-bytestring , bytestring , case-insensitive , containers , hal - , hedgehog >=1.0.3 && <1.1 + , hedgehog >=1.0.3 && <1.2 , hspec , hspec-hedgehog , http-client , http-types >=0.12.2 && <0.13 + , raw-strings-qq ==1.1.* , scientific , text , time , transformers , unordered-containers - , vector + , vector >=0.12.0.0 && <0.13 default-language: Haskell2010 diff --git a/src/AWS/Lambda/Events/Kafka.hs b/src/AWS/Lambda/Events/Kafka.hs new file mode 100644 --- /dev/null +++ b/src/AWS/Lambda/Events/Kafka.hs @@ -0,0 +1,273 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE DeriveTraversable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE NamedFieldPuns #-} + +{-| +Module : AWS.Lambda.Events.Kafka +Description : Data types for consuming Kafka events. +License : BSD3 +Stability : stable + +It is possible to subscribe Lambda functions to Kafka topics. You can +subscribe to topics from Amazon Managed Streaming for Kafka (MSK) as +well as self-managed Kafka clusters. + +Lambda considers Amazon Managed Streaming for Kafka (MSK) to be a +different event source type from a self-managed Apache Kafka cluster, +but their payloads are very similar. The types in this module are +derived from inspecting invocation payloads, and from reading the +following links: + + * https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html + * https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html + * https://aws.amazon.com/blogs/compute/using-amazon-msk-as-an-event-source-for-aws-lambda/ + * https://aws.amazon.com/blogs/compute/using-self-hosted-apache-kafka-as-an-event-source-for-aws-lambda/ + +-} + +module AWS.Lambda.Events.Kafka ( + KafkaEvent(..), + EventSource(..), + Record, + Record'(..), + Header (..), + Timestamp(..), + -- * Internal + parseTimestamp, + unparseTimestamp, + int64ToUTCTime, + utcTimeToInt64 +) where + +import Data.Aeson +import Data.Aeson.Encoding (text) +import Data.Aeson.Types +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import qualified Data.ByteString.Base64 as B64 +import Data.List.NonEmpty (NonEmpty) +import qualified Data.List.NonEmpty as NE +import Data.Function ((&)) +import Data.Int (Int32, Int64) +import Data.Map (Map) +import Data.Maybe (catMaybes, maybeToList) +import Data.Scientific (toBoundedInteger) +import Data.Semigroup ((<>)) +import Data.Text (Text) +import qualified Data.Text as T +import qualified Data.Text.Encoding as TE +import Data.Time (UTCTime) +import Data.Time.Clock.POSIX (posixSecondsToUTCTime, + utcTimeToPOSIXSeconds) +import GHC.Generics (Generic) + +#if MIN_VERSION_aeson(2,0,0) +import qualified Data.Aeson.Key as Key +import qualified Data.Aeson.KeyMap as KM + +keyToText :: Key -> Text +keyToText = Key.toText + +keyFromText :: Text -> Key +keyFromText = Key.fromText +#else +import qualified Data.HashMap.Strict as KM +type Key = Text + +keyToText :: Key -> Text +keyToText = id + +keyFromText :: Text -> Key +keyFromText = id +#endif + +-- | Represents an event from either Amazon MSK or a self-managed +-- Apache Kafka cluster, as the payloads are very similar. +-- +-- The 'ToJSON' and 'FromJSON' instances on 'Record' perform base64 +-- conversion for you. +-- +-- See the +-- for a sample payload. +data KafkaEvent = KafkaEvent { + eventSource :: !EventSource, + -- | Only present when using the "Amazon MSK" event source mapping. + eventSourceArn :: !(Maybe Text), + bootstrapServers :: !(NonEmpty Text), + -- | The map's keys look like @"${topicName}-${partitionNumber}"@ + records :: !(Map Text [Record]) +} deriving (Eq, Show, Generic) + +instance FromJSON KafkaEvent where + parseJSON = withObject "KafkaEvent" $ \o -> + let parseBootstrapServers + = maybe (fail "bootstrapServers: empty string") pure + . NE.nonEmpty + . T.splitOn "," + in KafkaEvent + <$> o .: "eventSource" + <*> o .:! "eventSourceArn" + <*> (o .: "bootstrapServers" >>= parseBootstrapServers) + <*> o .: "records" + +instance ToJSON KafkaEvent where + toJSON e = object $ + [ "eventSource" .= eventSource e + , "bootstrapServers" .= T.intercalate "," (NE.toList (bootstrapServers e)) + , "records" .= records e + ] <> maybeToList (("eventSourceArn" .=) <$> eventSourceArn e) + + toEncoding e = pairs . mconcat $ + [ "eventSource" .= eventSource e + , "bootstrapServers" .= T.intercalate "," (NE.toList (bootstrapServers e)) + , "records" .= records e + ] <> maybeToList (("eventSourceArn" .=) <$> eventSourceArn e) + +data EventSource + = AwsKafka -- ^ @"aws:kafka"@ + | SelfManagedKafka -- ^ @"SelfManagedKafka"@ + deriving (Eq, Show, Bounded, Enum, Ord, Generic) + +instance FromJSON EventSource where + parseJSON = withText "EventSource" $ \case + "aws:kafka" -> pure AwsKafka + "SelfManagedKafka" -> pure SelfManagedKafka + t -> fail $ "unrecognised EventSource: \"" <> show t <> "\"" + +instance ToJSON EventSource where + toJSON = \case + AwsKafka -> String "aws:kafka" + SelfManagedKafka -> String "SelfManagedKafka" + + toEncoding = text . \case + AwsKafka -> "aws:kafka" + SelfManagedKafka -> "SelfManagedKafka" + +-- | Convenience alias: most of the time you will parse the records +-- straight into some app-specific structure. +type Record = Record' ByteString + +-- | Records from a Kafka event. This is 'Traversable', which means +-- you can do things like parse a JSON-encoded payload: +-- +-- @ +-- 'traverse' 'decodeStrict' :: 'FromJSON' a => Record -> Maybe (Record' a) +-- @ +data Record' a = Record { + topic :: !Text, + partition :: !Int32, + offset :: !Int64, + timestamp :: !Timestamp, + -- | NOTE: there can be multiple headers for a given key. + headers :: [Header], + key :: !(Maybe ByteString), + value :: !(Maybe a) +} deriving (Eq, Show, Generic, Functor, Foldable, Traversable) + +-- | Decodes base64-encoded keys and values, where present. +instance FromJSON (Record' ByteString) where + parseJSON = withObject "Record" $ \o -> do + topic <- o .: "topic" + partition <- o .: "partition" + offset <- o .: "offset" + timestamp <- parseTimestamp o + headers <- o .: "headers" + key <- fmap (B64.decodeLenient . TE.encodeUtf8) <$> o .:? "key" + value <- fmap (B64.decodeLenient . TE.encodeUtf8) <$> o .:? "value" + pure Record { topic, partition, offset, timestamp, headers, key, value } + +-- | Encodes keys and values into base64. +instance ToJSON (Record' ByteString) where + toJSON r = object $ concat + [ + [ "offset" .= offset r + , "partition" .= partition r + , "topic" .= topic r + , "headers" .= headers r + ] + , unparseTimestamp (timestamp r) + , catMaybes + [ ("key" .=) . TE.decodeUtf8 . B64.encode <$> key r + , ("value" .=) . TE.decodeUtf8 . B64.encode <$> value r + ] + ] + + toEncoding r = pairs . mconcat . concat $ + [ + [ "offset" .= offset r + , "partition" .= partition r + , "topic" .= topic r + , "headers" .= headers r + ] + , unparseTimestamp (timestamp r) + , catMaybes + [ ("key" .=) . TE.decodeUtf8 . B64.encode <$> key r + , ("value" .=) . TE.decodeUtf8 . B64.encode <$> value r + ] + ] + +-- | AWS serialises record headers to JSON as an array of +-- objects. From their docs: +-- +-- @ +-- "headers":[{"headerKey":[104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]}] +-- @ +-- +-- Note: +-- +-- >>> map chr [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101] +-- "headerValue" +data Header = Header !Text !ByteString deriving (Eq, Show, Generic) + +instance FromJSON Header where + parseJSON = withObject "header" $ \o -> + case KM.toList o of + [(key, value)] -> Header (keyToText key) . B.pack <$> parseJSON value + [] -> fail "Unexpected empty object" + _ -> fail "Unexpected additional keys in object" + +instance ToJSON Header where + toJSON (Header key value) = object [keyFromText key .= B.unpack value] + toEncoding (Header key value) = pairs $ keyFromText key .= B.unpack value + +-- | Kafka timestamp types, derived from the Java client's enum at: +-- https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/record/TimestampType.java +data Timestamp = NoTimestampType | CreateTime !UTCTime | LogAppendTime !UTCTime + deriving (Eq, Show, Generic) + +parseTimestamp :: Object -> Parser Timestamp +parseTimestamp o = explicitParseField parseSubtype o "timestampType" + where + parseSubtype = withText "timestampType" $ \case + "NO_TIMESTAMP_TYPE" -> pure NoTimestampType + "CREATE_TIME" -> CreateTime <$> parseUTCTimestamp + "LOG_APPEND_TIME" -> LogAppendTime <$> parseUTCTimestamp + t -> fail $ "unknown timestampType: \"" <> show t <> "\"" + + parseUTCTimestamp = explicitParseField convertToUTCTime o "timestamp" + + convertToUTCTime = withScientific "timestamp" $ \stamp -> + int64ToUTCTime <$> (toBoundedInteger stamp & + maybe (fail "timestamp out of range") pure :: Parser Int64) + +unparseTimestamp :: KeyValue kv => Timestamp -> [kv] +unparseTimestamp = \case + NoTimestampType -> ["timestampType" .= String "NO_TIMESTAMP_TYPE"] + CreateTime ts -> + [ "timestampType" .= String "CREATE_TIME" + , "timestamp" .= utcTimeToInt64 ts + ] + LogAppendTime ts -> + [ "timestampType" .= String "LOG_APPEND_TIME" + , "timestamp" .= utcTimeToInt64 ts + ] + +int64ToUTCTime :: Int64 -> UTCTime +int64ToUTCTime int = + let (seconds, millis) = int `divMod` 1000 + posixSeconds = fromIntegral seconds + fromIntegral millis / 1000 + in posixSecondsToUTCTime posixSeconds + +utcTimeToInt64 :: UTCTime -> Int64 +utcTimeToInt64 utc = truncate $ utcTimeToPOSIXSeconds utc * 1000 diff --git a/test/AWS/Lambda/Events/ApiGateway/ProxyRequest/Gen.hs b/test/AWS/Lambda/Events/ApiGateway/ProxyRequest/Gen.hs --- a/test/AWS/Lambda/Events/ApiGateway/ProxyRequest/Gen.hs +++ b/test/AWS/Lambda/Events/ApiGateway/ProxyRequest/Gen.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE CPP #-} + module AWS.Lambda.Events.ApiGateway.ProxyRequest.Gen where import qualified AWS.Lambda.Events.ApiGateway.ProxyRequest.Gen.Resource as Resource @@ -6,7 +8,6 @@ import Control.Applicative (liftA2) import Data.Aeson (Value(..)) import qualified Data.ByteString.Lazy as BL -import qualified Data.HashMap.Strict as H import Data.Scientific (fromFloatDigits) import qualified Data.Vector as V import qualified Gen.Header as Header @@ -17,6 +18,21 @@ import Data.Text (Text) import qualified Data.Text.Encoding as TE +#if MIN_VERSION_aeson(2,0,0) +import Data.Aeson.Key (Key) +import qualified Data.Aeson.Key as Key +import qualified Data.Aeson.KeyMap as KM + +keyFromText :: Text -> Key +keyFromText = Key.fromText +#else +import qualified Data.HashMap.Strict as KM +type Key = Text + +keyFromText :: Text -> Key +keyFromText = id +#endif + proxyRequest :: Gen (ProxyRequest NoAuthorizer) proxyRequest = do segments <- Resource.segments @@ -105,8 +121,8 @@ [string, number, Bool <$> Gen.bool] [object, array] where - object = Object . H.fromList <$> Gen.list (Range.linear 1 50) entry - entry = liftA2 (,) text authJson + object = Object . KM.fromList <$> Gen.list (Range.linear 1 50) entry + entry = liftA2 (,) (keyFromText <$> text) authJson array = Array . V.fromList <$> Gen.list (Range.linear 1 50) authJson string = String <$> Gen.text (Range.exponential 1 200) Gen.unicode number = Number . fromFloatDigits <$> diff --git a/test/AWS/Lambda/Events/ApiGateway/ProxyRequest/Spec.hs b/test/AWS/Lambda/Events/ApiGateway/ProxyRequest/Spec.hs new file mode 100644 --- /dev/null +++ b/test/AWS/Lambda/Events/ApiGateway/ProxyRequest/Spec.hs @@ -0,0 +1,15 @@ +module AWS.Lambda.Events.ApiGateway.ProxyRequest.Spec where + +import qualified AWS.Lambda.Events.ApiGateway.ProxyRequest.Gen as Gen +import Data.Aeson (decode, encode) +import Hedgehog +import Test.Hspec (Spec, describe, specify) +import Test.Hspec.Hedgehog (hedgehog) + +spec :: Spec +spec = do + describe "properties" $ + specify "tripping" $ + hedgehog $ do + request <- forAll Gen.proxyRequest + tripping request encode decode diff --git a/test/AWS/Lambda/Events/ApiGateway/ProxyResponse/Spec.hs b/test/AWS/Lambda/Events/ApiGateway/ProxyResponse/Spec.hs new file mode 100644 --- /dev/null +++ b/test/AWS/Lambda/Events/ApiGateway/ProxyResponse/Spec.hs @@ -0,0 +1,15 @@ +module AWS.Lambda.Events.ApiGateway.ProxyResponse.Spec where + +import qualified AWS.Lambda.Events.ApiGateway.ProxyResponse.Gen as Gen +import Data.Aeson (decode, encode) +import Hedgehog +import Test.Hspec (Spec, describe, specify) +import Test.Hspec.Hedgehog (hedgehog) + +spec :: Spec +spec = do + describe "properties" $ + specify "tripping" $ + hedgehog $ do + request <- forAll Gen.proxyResponse + tripping request encode decode diff --git a/test/AWS/Lambda/Events/Kafka/Gen.hs b/test/AWS/Lambda/Events/Kafka/Gen.hs new file mode 100644 --- /dev/null +++ b/test/AWS/Lambda/Events/Kafka/Gen.hs @@ -0,0 +1,108 @@ +module AWS.Lambda.Events.Kafka.Gen where + +import AWS.Lambda.Events.Kafka (KafkaEvent(KafkaEvent), + EventSource (..), + Header(..), + Record, + Record' (Record), + Timestamp (..), + int64ToUTCTime) +import Control.Applicative (liftA2) +import Data.Char (isPrint) +import Data.Foldable (fold) +import Data.List (intersperse) +import qualified Data.List.NonEmpty as NE +import Data.List.NonEmpty (NonEmpty) +import Data.Map (Map) +import Data.Semigroup ((<>)) +import Data.Text (Text) +import qualified Data.Text as T +import Data.Time (UTCTime) +import Data.Traversable (for) +import Hedgehog (Gen) +import qualified Hedgehog.Gen as Gen +import qualified Hedgehog.Range as Range + +kafkaEvent :: Gen KafkaEvent +kafkaEvent = KafkaEvent + <$> eventSource + <*> Gen.maybe eventSourceArn + <*> bootstrapServers + <*> records + +eventSource :: Gen EventSource +eventSource = Gen.enumBounded + +eventSourceArn :: Gen Text +eventSourceArn = do + region <- Gen.element ["ap-south-1", "us-east-2", "eu-west-3"] + account <- Gen.text (Range.singleton 12) Gen.digit + clusterName <- Gen.text (Range.linear 1 20) printable + uuidParts <- for [ 7, 4, 4, 4, 12, 2 ] $ \n -> + Gen.text (Range.singleton n) Gen.hexit + let uuid = fold $ intersperse "-" uuidParts + resource = T.intercalate "/" ["cluster", clusterName, uuid] + pure $ T.intercalate ":" ["arn:aws:kafka", region, account, resource] + +bootstrapServers :: Gen (NonEmpty Text) +bootstrapServers = Gen.nonEmpty (Range.linear 1 5) server + where + server :: Gen Text + server = (\h p -> h <> ":" <> p) <$> host <*> port + + host :: Gen Text + host = fold . NE.intersperse "." <$> domainParts + + domainParts :: Gen (NonEmpty Text) + domainParts = Gen.nonEmpty (Range.linear 1 5) fragment + + fragment :: Gen Text + fragment = Gen.text (Range.linear 1 20) . + Gen.element $ ['A'..'Z'] <> ['a'..'z'] <> ['0'..'9'] <> ['-', '_'] + + port :: Gen Text + port = T.pack . show <$> Gen.int (Range.linear 0 65535) + +records :: Gen (Map Text [Record]) +records = Gen.map (Range.exponential 1 3) + . liftA2 (,) topicKey + $ Gen.list (Range.linear 1 5) record + where + topicKey :: Gen Text + topicKey = do + t <- topic + n <- T.pack . show <$> Gen.int (Range.linear 0 10) + pure $ t <> "-" <> n + +record :: Gen Record +record = Record + <$> topic + <*> Gen.int32 (Range.exponential 0 maxBound) + <*> Gen.int64 (Range.linear 0 maxBound) + <*> timestamp + <*> headers + <*> Gen.maybe (Gen.bytes (Range.exponential 0 512)) + <*> Gen.maybe (Gen.bytes (Range.exponential 0 512)) + +timestamp :: Gen Timestamp +timestamp = Gen.choice + [ pure NoTimestampType + , CreateTime <$> utcTime + , LogAppendTime <$> utcTime + ] + where + utcTime :: Gen UTCTime + utcTime = int64ToUTCTime <$> Gen.int64 + (Range.exponentialFrom 1600000000000 1650000000000 1700000000000) + +headers :: Gen [Header] +headers = Gen.list (Range.linear 0 10) + (Header + <$> Gen.text (Range.linear 0 512) Gen.unicode + <*> Gen.bytes (Range.linear 0 512)) + +topic :: Gen Text +topic = Gen.text (Range.linear 1 20) printable + +printable :: Gen Char +printable = Gen.filter isPrint Gen.ascii diff --git a/test/AWS/Lambda/Events/Kafka/Spec.hs b/test/AWS/Lambda/Events/Kafka/Spec.hs new file mode 100644 --- /dev/null +++ b/test/AWS/Lambda/Events/Kafka/Spec.hs @@ -0,0 +1,199 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE QuasiQuotes #-} + +module AWS.Lambda.Events.Kafka.Spec where + +import AWS.Lambda.Events.Kafka +import qualified AWS.Lambda.Events.Kafka.Gen as Gen +import Data.Aeson (decode, eitherDecode, encode) +import Data.ByteString.Lazy (ByteString) +import Data.Int (Int64) +import qualified Data.List.NonEmpty as NE +import qualified Data.Map as M +import Data.Time (UTCTime(..), fromGregorian, + picosecondsToDiffTime) +import Hedgehog (forAll, tripping) +import Test.Hspec (Spec, describe, shouldBe, specify) +import Test.Hspec.Hedgehog (hedgehog) +import Text.RawString.QQ (r) + +spec :: Spec +spec = do + specify "read MSK payload" $ + eitherDecode sampleMskPayload + `shouldBe` Right expectedMskPayload + + specify "read Self-Managed Kafka payload" $ + eitherDecode sampleSelfManagedKafkaPayload + `shouldBe` Right expectedSelfManagedKafkaPayload + + describe "properties" $ do + specify "tripping" $ + hedgehog $ do + request <- forAll Gen.kafkaEvent + tripping request encode decode + +sampleMskPayload :: ByteString +sampleMskPayload = [r| +{ + "bootstrapServers": "b-1.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9094,b-2.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9094", + "eventSource": "aws:kafka", + "eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-test-cluster-2/669b9b55-48db-44e6-9c75-b9b9ab79ae36-14", + "records": { + "welcome-to-the-space-jam-0": [ + { + "key": "c2xhbQ==", + "offset": 1, + "partition": 0, + "headers": [ + { + "headerKey": [ + 104, + 101, + 97, + 100, + 101, + 114, + 86, + 97, + 108, + 117, + 101 + ] + } + ], + "timestamp": 1621304398263, + "timestampType": "CREATE_TIME", + "topic": "welcome-to-the-space-jam", + "value": "amFt" + } + ] + } +} +|] + +expectedMskPayload :: KafkaEvent +expectedMskPayload = KafkaEvent + { eventSource = AwsKafka + , eventSourceArn = Just "arn:aws:kafka:us-east-1:123456789012:cluster/msk-test-cluster-2/669b9b55-48db-44e6-9c75-b9b9ab79ae36-14" + , bootstrapServers = NE.fromList + [ "b-1.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9094" + , "b-2.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9094" + ] + , records = M.fromList + [ ( "welcome-to-the-space-jam-0" + , [ withHeaders [Header "headerKey" "headerValue"] + . slamJamRecord 1 + $ UTCTime + (fromGregorian 2021 05 18) + (picosecondsToDiffTime 8398263000000000) + ] + ) + ] + } + +sampleSelfManagedKafkaPayload :: ByteString +sampleSelfManagedKafkaPayload = [r| +{ + "bootstrapServers": "b-2.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9092,b-1.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9092", + "eventSource": "SelfManagedKafka", + "records": { + "welcome-to-the-space-jam-0": [ + { + "key": "c2xhbQ==", + "offset": 0, + "partition": 0, + "headers": [], + "timestamp": 1621302644353, + "timestampType": "CREATE_TIME", + "topic": "welcome-to-the-space-jam", + "value": "amFt" + }, + { + "key": "c2xhbQ==", + "offset": 1, + "partition": 0, + "headers": [], + "timestamp": 1621304398263, + "timestampType": "CREATE_TIME", + "topic": "welcome-to-the-space-jam", + "value": "amFt" + }, + { + "key": "c2xhbQ==", + "offset": 2, + "partition": 0, + "headers": [], + "timestamp": 1621309340017, + "timestampType": "CREATE_TIME", + "topic": "welcome-to-the-space-jam", + "value": "amFt" + }, + { + "key": "c2xhbQ==", + "offset": 3, + "partition": 0, + "headers": [], + "timestamp": 1621309352813, + "timestampType": "CREATE_TIME", + "topic": "welcome-to-the-space-jam", + "value": "amFt" + }, + { + "key": "c2xhbQ==", + "offset": 4, + "partition": 0, + "headers": [], + "timestamp": 1621309372115, + "timestampType": "CREATE_TIME", + "topic": "welcome-to-the-space-jam", + "value": "amFt" + } + ] + } +} +|] + +expectedSelfManagedKafkaPayload :: KafkaEvent +expectedSelfManagedKafkaPayload = KafkaEvent + { eventSource = SelfManagedKafka + , eventSourceArn = Nothing + , bootstrapServers = NE.fromList + [ "b-2.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9092" + , "b-1.msk-test-cluster-2.4jp6ci.c14.kafka.us-east-1.amazonaws.com:9092" + ] + , records = M.fromList + [ ( "welcome-to-the-space-jam-0" + , [ slamJamRecord 0 $ UTCTime + (fromGregorian 2021 05 18) + (picosecondsToDiffTime 6644353000000000) + , slamJamRecord 1 $ UTCTime + (fromGregorian 2021 05 18) + (picosecondsToDiffTime 8398263000000000) + , slamJamRecord 2 $ UTCTime + (fromGregorian 2021 05 18) + (picosecondsToDiffTime 13340017000000000) + , slamJamRecord 3 $ UTCTime + (fromGregorian 2021 05 18) + (picosecondsToDiffTime 13352813000000000) + , slamJamRecord 4 $ UTCTime + (fromGregorian 2021 05 18) + (picosecondsToDiffTime 13372115000000000) + ] + ) + ] + } + +slamJamRecord :: Int64 -> UTCTime -> Record +slamJamRecord off stamp = Record + { topic = "welcome-to-the-space-jam" + , partition = 0 + , offset = off + , headers = [] + , timestamp = CreateTime stamp + , key = Just "slam" + , value = Just "jam" + } + +withHeaders :: [Header] -> Record -> Record +withHeaders headers record = record { headers } diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -3,11 +3,13 @@ CognitoIdentity (..), LambdaContext (..)) import qualified Gen.Header as Header -import qualified AWS.Lambda.Events.ApiGateway.ProxyRequest.Gen as ProxyRequest -import qualified AWS.Lambda.Events.ApiGateway.ProxyResponse.Gen as ProxyResponse +import qualified AWS.Lambda.Events.ApiGateway.ProxyRequest.Spec as ProxyRequest +import qualified AWS.Lambda.Events.ApiGateway.ProxyResponse.Spec as ProxyResponse + +import qualified AWS.Lambda.Events.Kafka.Spec as Kafka import AWS.Lambda.Internal (StaticContext (..)) import AWS.Lambda.RuntimeClient.Internal (eventResponseToNextData) -import Data.Aeson (Value (Null), decode, encode) +import Data.Aeson (Value (Null)) import qualified Data.CaseInsensitive as CI import Data.Map (singleton) import Data.Semigroup ((<>)) @@ -25,6 +27,12 @@ main :: IO () main = hspec $ do + describe "Events" $ do + describe "ApiGateway" $ do + describe "ProxyRequest" ProxyRequest.spec + describe "ProxyResponse" ProxyResponse.spec + describe "Kafka" Kafka.spec + describe "Event Response Data" $ do let staticContext = StaticContext @@ -214,17 +222,6 @@ s <- forAll $ CI.mk <$> Gen.text (Range.linear 1 100) Gen.latin1 t <- forAll $ Header.unfoldCase s s === t - - specify "ProxyRequest tripping" $ - hedgehog $ do - request <- forAll ProxyRequest.proxyRequest - tripping request encode decode - - specify "ProxyResponse tripping" $ - hedgehog $ do - request <- forAll ProxyResponse.proxyResponse - tripping request encode decode - minResponse :: [Header] -> a -> Response a minResponse headers body =