Mayfly.Events.Kinesis (Mayfly v1.0.0-rc.1)

Copy Markdown View Source

Amazon Kinesis Data Streams events.

kinesis = Mayfly.Events.Kinesis.decode(event)

{:ok,
 Mayfly.Events.Kinesis.process_batch(kinesis, fn record ->
   MyApp.Ingest.handle(record.data)
 end)}

data is base64-decoded and JSON-decoded when it parses, otherwise the raw bytes. Partial batch responses use the record's sequence_number; the event source mapping needs FunctionResponseTypes=ReportBatchItemFailures.

Summary

Types

t()

Functions

batch_failures(sequence_numbers)

Partial batch response for the given sequence numbers.

decode(e)

Decodes a Kinesis event.

process_batch(kinesis, fun)

Runs fun per record, collecting failures by sequence number.

Types

t()

@type t() :: %Mayfly.Events.Kinesis{
  raw: map(),
  records: [Mayfly.Events.Kinesis.Record.t()]
}

Functions

batch_failures(sequence_numbers)

@spec batch_failures([String.t()]) :: %{
  batchItemFailures: [%{itemIdentifier: String.t()}]
}

Partial batch response for the given sequence numbers.

decode(e)

@spec decode(map()) :: t()

Decodes a Kinesis event.

process_batch(kinesis, fun)

@spec process_batch(t(), (Mayfly.Events.Kinesis.Record.t() -> term())) :: %{
  batchItemFailures: [%{itemIdentifier: String.t()}]
}

Runs fun per record, collecting failures by sequence number.