Kinesis streaming API proposal

[semenodm]

The new Kinesis fs2 integration API proposed
The implementation
https://github.com/laserdisc-io/fs2-aws/blob/master/fs2-aws/src/main/scala/fs2/aws/kinesis/Kinesis.scala
Usage example https://github.com/laserdisc-io/fs2-aws/blob/master/fs2-aws-examples/src/main/scala/KinesisExample.scala

I found most of the issues we experience is that sometimes we need to customize AWS SDK clients with different parameters like, sts client, region, service endpoint (for localstack) etc. One of the key point is to give user to define the clients (Kinesis, DynamoDB and Cloudwatch) outside and inject it into stream api.
You can find that we created the number of so called Pure wrappers on top of AWS Async SDK clients under pure-aws folder in this repository. These wrappers are used to create Client Resources for any supported AWS service. This approach is also better, since we can share AWS client across different programs and be sure that at the end all resources will be destroyed properly in correct order.

[gregorath]

I changed my code to use the new way and I think it is nicer.
It is a big advantage to be able to configure the AWS clients since there are various options available. And it was also helpful with the tests, for example we have an integration test with localstack that looks at dynamodb for the checkpoint and it was easier to do it now using the same client without worrying too much about the dynamodb integration.

I don't know if it would make sense to have a default instance, something that would take a Blocker and return probably a Resource[Kinesis[F]] with default AWS clients like KinesisAsyncClient.builder(),...?

In any case, I will stick with the new API now. Thank you for this improvement 👍

[semenodm]

thank you for your response, especially for positive one :)
as for default params, i'm not sure i want it to be, because the variety of the places this library may be used makes it difficult to choose the defaults, for example AWS region.

[gregorath]

yes that makes sense.

for context, in our case we are using environment variables for the credentials so I thought something like

def apply[F[_]: ConcurrentEffect: ContextShift: Timer](blocker: Blocker): Resource[F, Kinesis[F]] =
    for {
      k <- KinesisInterpreter[F](blocker).KinesisAsyncClientResource(KinesisAsyncClient.builder())
      d <- DynamoDbInterpreter[F](blocker).DynamoDbAsyncClientResource(DynamoDbAsyncClient.builder())
      c <- CloudwatchInterpreter[F](blocker).CloudWatchAsyncClientResource(CloudWatchAsyncClient.builder())
    } yield Kinesis.create(k, d, c, blocker)

could be used as a default implementation.
although to be fair it is not offering much apart from saving a few lines in some cases.

[semenodm]

Hey, just to understand the library community, where do you use it?
Of course if you can answer

[gregorath]

Hey, yes we use it in a team in OVO Energy (https://www.ovoenergy.com)

[kubukoz]

def apply[F[_]: ConcurrentEffect: ContextShift: Timer](blocker: Blocker)

I love how that signature becomes like 3x shorter with CE3...