License	BSD3
Stability	stable
Safe Haskell	Safe-Inferred
Language	Haskell2010

AWS.Lambda.Events.Kafka

Contents

Internal

Description

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:

Synopsis

data KafkaEvent = KafkaEvent {
- eventSource :: !EventSource
- eventSourceArn :: !(Maybe Text)
- bootstrapServers :: !(NonEmpty Text)
- records :: !(Map Text [Record])
}
data EventSource
- = AwsKafka
- | SelfManagedKafka
type Record = Record' ByteString
data Record' a = Record {
- topic :: !Text
- partition :: !Int32
- offset :: !Int64
- timestamp :: !Timestamp
- headers :: [Header]
- key :: !(Maybe ByteString)
- value :: !(Maybe a)
}
data Header = Header !Text !ByteString
data Timestamp
- = NoTimestampType
- | CreateTime !UTCTime
- | LogAppendTime !UTCTime
parseTimestamp :: Object -> Parser Timestamp
unparseTimestamp :: KeyValue kv => Timestamp -> [kv]
int64ToUTCTime :: Int64 -> UTCTime
utcTimeToInt64 :: UTCTime -> Int64

Documentation

data KafkaEvent Source #

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 AWS documentation for a sample payload.

Constructors

KafkaEvent
Fields eventSource :: !EventSource eventSourceArn :: !(Maybe Text) Only present when using the "Amazon MSK" event source mapping. bootstrapServers :: !(NonEmpty Text) records :: !(Map Text [Record]) The map's keys look like `"${topicName}-${partitionNumber}"`

Instances

Instances details

FromJSON KafkaEvent Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods parseJSON :: Value -> Parser KafkaEvent # parseJSONList :: Value -> Parser [KafkaEvent] #
ToJSON KafkaEvent Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods toJSON :: KafkaEvent -> Value # toEncoding :: KafkaEvent -> Encoding # toJSONList :: [KafkaEvent] -> Value # toEncodingList :: [KafkaEvent] -> Encoding #
Generic KafkaEvent Source #
Instance details Defined in AWS.Lambda.Events.Kafka Associated Types type Rep KafkaEvent :: Type -> Type # Methods from :: KafkaEvent -> Rep KafkaEvent x # to :: Rep KafkaEvent x -> KafkaEvent #
Show KafkaEvent Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods showsPrec :: Int -> KafkaEvent -> ShowS # show :: KafkaEvent -> String # showList :: [KafkaEvent] -> ShowS #
Eq KafkaEvent Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods (==) :: KafkaEvent -> KafkaEvent -> Bool # (/=) :: KafkaEvent -> KafkaEvent -> Bool #
type Rep KafkaEvent Source #
Instance details Defined in AWS.Lambda.Events.Kafka type Rep KafkaEvent = D1 ('MetaData "KafkaEvent" "AWS.Lambda.Events.Kafka" "hal-1.1-IwuGgwGQvpsJOoRlidLUAk" 'False) (C1 ('MetaCons "KafkaEvent" 'PrefixI 'True) ((S1 ('MetaSel ('Just "eventSource") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 EventSource) :: S1 ('MetaSel ('Just "eventSourceArn") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Maybe Text))) :: (S1 ('MetaSel ('Just "bootstrapServers") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (NonEmpty Text)) :*: S1 ('MetaSel ('Just "records") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Map Text [Record])))))

data EventSource Source #

Constructors

AwsKafka	"aws:kafka"
SelfManagedKafka	SelfManagedKafka

Instances

Instances details

FromJSON EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods parseJSON :: Value -> Parser EventSource # parseJSONList :: Value -> Parser [EventSource] #
ToJSON EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods toJSON :: EventSource -> Value # toEncoding :: EventSource -> Encoding # toJSONList :: [EventSource] -> Value # toEncodingList :: [EventSource] -> Encoding #
Bounded EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods minBound :: EventSource # maxBound :: EventSource #
Enum EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods succ :: EventSource -> EventSource # pred :: EventSource -> EventSource # toEnum :: Int -> EventSource # fromEnum :: EventSource -> Int # enumFrom :: EventSource -> [EventSource] # enumFromThen :: EventSource -> EventSource -> [EventSource] # enumFromTo :: EventSource -> EventSource -> [EventSource] # enumFromThenTo :: EventSource -> EventSource -> EventSource -> [EventSource] #
Generic EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Associated Types type Rep EventSource :: Type -> Type # Methods from :: EventSource -> Rep EventSource x # to :: Rep EventSource x -> EventSource #
Show EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods showsPrec :: Int -> EventSource -> ShowS # show :: EventSource -> String # showList :: [EventSource] -> ShowS #
Eq EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods (==) :: EventSource -> EventSource -> Bool # (/=) :: EventSource -> EventSource -> Bool #
Ord EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods compare :: EventSource -> EventSource -> Ordering # (<) :: EventSource -> EventSource -> Bool # (<=) :: EventSource -> EventSource -> Bool # (>) :: EventSource -> EventSource -> Bool # (>=) :: EventSource -> EventSource -> Bool # max :: EventSource -> EventSource -> EventSource # min :: EventSource -> EventSource -> EventSource #
type Rep EventSource Source #
Instance details Defined in AWS.Lambda.Events.Kafka type Rep EventSource = D1 ('MetaData "EventSource" "AWS.Lambda.Events.Kafka" "hal-1.1-IwuGgwGQvpsJOoRlidLUAk" 'False) (C1 ('MetaCons "AwsKafka" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "SelfManagedKafka" 'PrefixI 'False) (U1 :: Type -> Type))

type Record = Record' ByteString Source #

Convenience alias: most of the time you will parse the records straight into some app-specific structure.

data Record' a Source #

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)

Constructors

Record
Fields topic :: !Text partition :: !Int32 offset :: !Int64 timestamp :: !Timestamp headers :: [Header] NOTE: there can be multiple headers for a given key. key :: !(Maybe ByteString) value :: !(Maybe a)

Instances

Instances details

Foldable Record' Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods fold :: Monoid m => Record' m -> m # foldMap :: Monoid m => (a -> m) -> Record' a -> m # foldMap' :: Monoid m => (a -> m) -> Record' a -> m # foldr :: (a -> b -> b) -> b -> Record' a -> b # foldr' :: (a -> b -> b) -> b -> Record' a -> b # foldl :: (b -> a -> b) -> b -> Record' a -> b # foldl' :: (b -> a -> b) -> b -> Record' a -> b # foldr1 :: (a -> a -> a) -> Record' a -> a # foldl1 :: (a -> a -> a) -> Record' a -> a # toList :: Record' a -> [a] # null :: Record' a -> Bool # length :: Record' a -> Int # elem :: Eq a => a -> Record' a -> Bool # maximum :: Ord a => Record' a -> a # minimum :: Ord a => Record' a -> a # sum :: Num a => Record' a -> a # product :: Num a => Record' a -> a #
Traversable Record' Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods traverse :: Applicative f => (a -> f b) -> Record' a -> f (Record' b) # sequenceA :: Applicative f => Record' (f a) -> f (Record' a) # mapM :: Monad m => (a -> m b) -> Record' a -> m (Record' b) # sequence :: Monad m => Record' (m a) -> m (Record' a) #
Functor Record' Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods fmap :: (a -> b) -> Record' a -> Record' b # (<$) :: a -> Record' b -> Record' a #
FromJSON (Record' ByteString) Source #	Decodes base64-encoded keys and values, where present.
Instance details Defined in AWS.Lambda.Events.Kafka Methods parseJSON :: Value -> Parser (Record' ByteString) # parseJSONList :: Value -> Parser [Record' ByteString] #
ToJSON (Record' ByteString) Source #	Encodes keys and values into base64.
Instance details Defined in AWS.Lambda.Events.Kafka Methods toJSON :: Record' ByteString -> Value # toEncoding :: Record' ByteString -> Encoding # toJSONList :: [Record' ByteString] -> Value # toEncodingList :: [Record' ByteString] -> Encoding #
Generic (Record' a) Source #
Instance details Defined in AWS.Lambda.Events.Kafka Associated Types type Rep (Record' a) :: Type -> Type # Methods from :: Record' a -> Rep (Record' a) x # to :: Rep (Record' a) x -> Record' a #
Show a => Show (Record' a) Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods showsPrec :: Int -> Record' a -> ShowS # show :: Record' a -> String # showList :: [Record' a] -> ShowS #
Eq a => Eq (Record' a) Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods (==) :: Record' a -> Record' a -> Bool # (/=) :: Record' a -> Record' a -> Bool #
type Rep (Record' a) Source #
Instance details Defined in AWS.Lambda.Events.Kafka type Rep (Record' a) = D1 ('MetaData "Record'" "AWS.Lambda.Events.Kafka" "hal-1.1-IwuGgwGQvpsJOoRlidLUAk" 'False) (C1 ('MetaCons "Record" 'PrefixI 'True) ((S1 ('MetaSel ('Just "topic") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Text) :: (S1 ('MetaSel ('Just "partition") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Int32) :: S1 ('MetaSel ('Just "offset") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Int64))) :: ((S1 ('MetaSel ('Just "timestamp") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Timestamp) :: S1 ('MetaSel ('Just "headers") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 [Header])) :: (S1 ('MetaSel ('Just "key") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Maybe ByteString)) :: S1 ('MetaSel ('Just "value") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 (Maybe a))))))

data Header Source #

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"

Constructors

!Text !ByteString

Instances

data Timestamp Source #

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

Constructors

NoTimestampType
CreateTime !UTCTime
LogAppendTime !UTCTime

Instances

Instances details

Generic Timestamp Source #
Instance details Defined in AWS.Lambda.Events.Kafka Associated Types type Rep Timestamp :: Type -> Type # Methods from :: Timestamp -> Rep Timestamp x # to :: Rep Timestamp x -> Timestamp #
Show Timestamp Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods showsPrec :: Int -> Timestamp -> ShowS # show :: Timestamp -> String # showList :: [Timestamp] -> ShowS #
Eq Timestamp Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods (==) :: Timestamp -> Timestamp -> Bool # (/=) :: Timestamp -> Timestamp -> Bool #
type Rep Timestamp Source #
Instance details Defined in AWS.Lambda.Events.Kafka type Rep Timestamp = D1 ('MetaData "Timestamp" "AWS.Lambda.Events.Kafka" "hal-1.1-IwuGgwGQvpsJOoRlidLUAk" 'False) (C1 ('MetaCons "NoTimestampType" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "CreateTime" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 UTCTime)) :+: C1 ('MetaCons "LogAppendTime" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 UTCTime))))

Internal

parseTimestamp :: Object -> Parser Timestamp Source #

unparseTimestamp :: KeyValue kv => Timestamp -> [kv] Source #

int64ToUTCTime :: Int64 -> UTCTime Source #

utcTimeToInt64 :: UTCTime -> Int64 Source #

FromJSON Header Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods parseJSON :: Value -> Parser Header # parseJSONList :: Value -> Parser [Header] #
ToJSON Header Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods toJSON :: Header -> Value # toEncoding :: Header -> Encoding # toJSONList :: [Header] -> Value # toEncodingList :: [Header] -> Encoding #
Generic Header Source #
Instance details Defined in AWS.Lambda.Events.Kafka Associated Types type Rep Header :: Type -> Type # Methods from :: Header -> Rep Header x # to :: Rep Header x -> Header #
Show Header Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods showsPrec :: Int -> Header -> ShowS # show :: Header -> String # showList :: [Header] -> ShowS #
Eq Header Source #
Instance details Defined in AWS.Lambda.Events.Kafka Methods (==) :: Header -> Header -> Bool # (/=) :: Header -> Header -> Bool #
type Rep Header Source #
Instance details Defined in AWS.Lambda.Events.Kafka type Rep Header = D1 ('MetaData "Header" "AWS.Lambda.Events.Kafka" "hal-1.1-IwuGgwGQvpsJOoRlidLUAk" 'False) (C1 ('MetaCons "Header" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Text) :*: S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 ByteString)))