Package eu.vstoyanov.natsy.jetstream

• Consumer information and updates

• Pause/resume operations

• Monitoring metrics

val consumerCtx = js.consumer("MY_STREAM", "MY_CONSUMER") {
    durable = "MY_CONSUMER"
    ackPolicy = AckPolicy.EXPLICIT
    maxDeliver = 5
}

// Update the consumer
consumerCtx.update {
    maxDeliver = 10
}

// Pause the consumer until it is explicitly resumed
consumerCtx.pause()

// ...or until a deadline, after which the server resumes it automatically
consumerCtx.pause(30.minutes)

// Everything worth monitoring, in one round trip
val stats = consumerCtx.stats()
println("${stats.ackPending} unacknowledged, ${stats.pending} still to come")

val streamCtx = js.stream("MY_STREAM") {
    subjects = listOf("orders.*", "payments.*")
    retentionPolicy = RetentionPolicy.LIMITS
    maxMessages = 1000000
    maxBytes = 1024 * 1024 * 1024 // 1GB
    storageType = StorageType.FILE
    replicas = 3
}