kafka-consumer.fsx

#nowarn "760"
#r "nuget:Confluent.Kafka"

open System
open System.Diagnostics
open System.Threading
open Confluent.Kafka

[<Literal>]
let Topic = "test-topic"

let config = ConsumerConfig(BootstrapServers = "localhost:9092",
                            GroupId = "test-group",
                            AutoOffsetReset = AutoOffsetReset.Earliest,
                            SocketTimeoutMs = 5000
                            )

let consumer = ConsumerBuilder<string,string>(config).Build()

consumer.Subscribe Topic

let cancel = CancellationTokenSource()
Console.CancelKeyPress.AddHandler(fun _ e ->
    cancel.Cancel()
    e.Cancel <- true
    )

printfn "Consumer started!"
try
    while true do
        let m = consumer.Consume(cancel.Token)
        let message = m.Message
        let timestamp = message.Timestamp.UtcDateTime.ToLocalTime()
        printfn $"{timestamp} Received {m.Partition}/{m.Offset}: %s{message.Key} -> %s{message.Value}"
with
| :? OperationCanceledException -> ()
| e -> printfn $"Error! {e}"

printfn "Consumer quited!"
let watch = Stopwatch.StartNew()
consumer.Close()
printfn $"Consumer closed in {watch.ElapsedMilliseconds}ms!"