FarmáciasAPP Kinesis-Streams

Biblioteca para padronizar e facilitar a utilização do Kinesis como um broker de mensagens.

Como utilizar

Produtor

A classe KinesisStreamClient possui uma implementação default que instancia baseado na variável de ambiente BROKER_NAME, considerando que o broker utilizado esteja na mesma região e conta da AWS.

Se necessitar customizar, basta criar uma nova instância.

Sua utilização é feita por 2 métodos

• sendStream - serializa 1 objeto para o stream.
• sendBatch - envia N objetos para o stream.

A chave de partição é sempre opcional, e no caso de omissão é utilizado o nome do evento + uuid randomico.

Exemplo de utilização

import {
  defaultKinesisClient,
  KinesisStreamClient,
} from '@develop-fapp/kinesis-streams';

interface Example {
    id: string;
    name: string;
}

export class ExampleRepository {
  constructor(
    private readonly client: KinesisStreamClient = defaultKinesisClient,
  ) {}

  sendBatch(examples: Example[]): Promise<void> {
    return this.client.sendBatch(
      'my-example-changed',
      examples.map(example => ({
        data: example,
        key: `CUSTOM_PARTITION##${example.id}`,
      })),
    );
  }

  send(example: Example): Promise<void> {
    return this.client.sendStream(
      'my-example-changed',
      example,
      key: `CUSTOM_PARTITION##${example.id}`,
    );
  }
}

Consumidor

Para consumir os eventos disparados via lambda, é disponibilizado os wrappers

• kinesisLambdaWrapper - Retona somente a lista de objetos do stream
• kinesisLambdaRawWrapper - Retorna o payload completo do stream (quando necessário tratar nomes e posições)

Exemplos

import {
  kinesisLambdaWrapper,
  kinesisLambdaRawWrapper,
  Stream
} from '@develop-fapp/kinesis-streams';

interface Example {
    id: string;
    name: string;
}

export const simpleHandler = kinesisLambdaWrapper<Example>(
  async (inputs: Example[]) => {
    console.log(inputs); // print [{ id, name }]
    // ...
  }
);

export const complexHandler = kinesisLambdaRawWrapper<Example>(
  async (inputs: Stream<Example>[]) => {
    console.log(inputs); // print [{ streamName, data: { id, name } }]
    // ...
  }
);