Ciris

Standard environment variables and system properties for configuring a KCLConsumer, via Ciris

Installation

libraryDependencies += "io.github.etspaceman" %% "kinesis4cats-kcl-ciris" % "0.0.32"

Usage

import cats.effect._
import cats.syntax.all._

import kinesis4cats.kcl._
import kinesis4cats.kcl.ciris.KCLCiris
import kinesis4cats.syntax.bytebuffer._

object MyApp extends ResourceApp.Forever {
    override def run(args: List[String]) = for {
        consumer <- KCLCiris.consumer[IO](){ 
            case records: List[CommittableRecord[IO]] => 
                records.traverse_(r => IO.println(r.data.asString)) 
        }
        _ <- consumer.run()
    } yield ()
}

Configuration

The configuration below is intended to only work with single-stream consumers at this time.

Common

Checkpoint

Currently the CheckpointConfig does not have any configuration, but this section is here in case it is provided at a later date.

Coordinator

CoordinatorConfig values.

Lease

LeaseManagementConfig values.

Lifecycle

LifecycleConfig values.

Metrics

MetricsConfig values.

Retrieval

RetrievalConfig values.

FanOut

Configuration values if KCL_RETRIEVAL_TYPE/kcl.retrieval.type is fanout. Used for constructing the FanOutConfig

Polling

Configuration values if KCL_RETRIEVAL_TYPE/kcl.retrieval.type is polling. Used for constructing the PollingConfig

Processor

ProcessorConfig values.

Duration

Some of these environment variables are loaded as a Scala Duration, and are parsed through its apply method.

For example, if you wanted to set the duration to a day, you could use: - 1d - 1 day - 24 hours - 24h

FS2 Configuration

Standard environment variables and system properties for configuring a KCLConsumerFS2, via Ciris

Usage

import cats.effect._

import kinesis4cats.kcl.fs2.ciris.KCLCirisFS2
import kinesis4cats.syntax.bytebuffer._

object MyApp extends ResourceApp.Forever {
    override def run(args: List[String]) = for {
        consumer <- KCLCirisFS2.consumer[IO]()
        _ <- consumer.stream()
            .flatMap(stream =>
                stream
                .evalTap(x => IO.println(x.data.asString))
                .through(consumer.commitRecords)
                .compile
                .resource
                .drain
            )
    } yield ()
}

Configuration

The configuration below is intended to only work with single-stream consumers at this time.