Kafka F# example

kafka.fsx

#r "nuget: Confluent.Kafka"
#r "nuget: System.Text.Json"

open Confluent.Kafka
open System
open System.Text.Json
open System.Threading.Tasks
open Confluent.Kafka.Admin

let server = "localhost:9092"
let topicName = "topic"

let producerConfig = ProducerConfig(BootstrapServers = server)

try
  AdminClientBuilder(AdminClientConfig producerConfig)
    .Build()
    .CreateTopicsAsync([| TopicSpecification(Name = topicName) |])
    .Wait()
with _ -> ()

let producer = ProducerBuilder<byte array, byte array>(producerConfig).Build()

let consumer =
  ConsumerBuilder<byte array, byte array>(
    ConsumerConfig(BootstrapServers = server, GroupId = Guid.NewGuid().ToString())
  )
    .Build()

consumer.Subscribe topicName

Task.Run(fun () ->
  while true do
    try
      let result = consumer.Consume()

      if result <> null && result.Message <> null then
        let message = JsonSerializer.Deserialize result.Message.Value
        printfn $"""MESSAGE %A{message}"""
    with ex ->
      eprintfn $"EXCEPTION {ex.Message}")

let send (request: byte array) =
  producer.Produce(topicName, Message(Value = request))

printfn "Press any key to send"
Console.ReadKey()

JsonSerializer.SerializeToUtf8Bytes {| Now = DateTimeOffset.UtcNow |} |> send
producer.Flush()

printfn "Press any key to exit"
Console.ReadKey()