Basic Pub/Sub

Overview

In this tutorial, you'll build the "hello world" of KubeMQ messaging: a publisher and a subscriber talking over the Events pattern. Events are fire-and-forget — the broker fans a message out to every subscriber currently listening on the channel and moves on. There's no persistence, no acknowledgment, and no replay, which makes this the pattern to reach for when you need low-latency, high-throughput broadcast (metrics ticks, live status updates, cache-invalidation signals) and can tolerate losing a message if nobody is listening at the moment it's sent.

You'll wire up KubeMQ.Client.subscribe_to_events/3 with an on_event callback, give the subscription a moment to register with the server, then call KubeMQ.Client.send_event/2 to publish a KubeMQ.Event. Every connected subscriber on the channel gets its own copy, as opposed to a consumer group where only one member would receive it. Gotchas: if the subscriber isn't fully established before you publish, the event is simply gone — there's no queue catching it, which is why the sample sleeps briefly before sending; and because delivery isn't acknowledged, a crashed or disconnected subscriber never knows it missed anything.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-events.basic-pubsub"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-events-basic")

parent = self()

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, channel,
    on_event: fn event ->
      IO.puts("Received event: #{event.body}")
      send(parent, :received)
    end
  )

IO.puts("Subscribed to '#{channel}'")
Process.sleep(500)

event = KubeMQ.Event.new(channel: channel, body: "Hello from Elixir!", metadata: "greeting")

case KubeMQ.Client.send_event(client, event) do
  :ok -> IO.puts("Event sent successfully")
  {:error, err} -> IO.puts("Send failed: #{err.message}")
end

receive do
  :received -> IO.puts("Event delivery confirmed!")
after
  5_000 -> IO.puts("Timeout waiting for event")
end

KubeMQ.Subscription.cancel(sub)
KubeMQ.Client.close(client)

How It Works

• Events are fire-and-forget — subscribers must be active before publishing
• The subscriber callback receives an %KubeMQ.EventReceive{} struct
• send/2 is used to notify the parent process for synchronization
• The receive block waits for the event with a 5-second timeout

Related

• Events Pattern Overview
• Cancel Subscription
• Consumer Group