# Publish & Subscribe (/learn/events/tutorials/publish-subscribe)



## What You Will Build [#what-you-will-build]

A publisher sending order events with metadata and tags, and two subscribers both receiving every event (fan-out).

<Mermaid
  chart="graph LR
  OS[&#x22;Order Service<br/>(publisher)&#x22;]
  CH{{&#x22;Events channel<br/>order-events&#x22;}}
  NS[&#x22;Notification Service&#x22;]
  AS[&#x22;Analytics Service&#x22;]

  OS -- publish --> CH
  CH -- &#x22;at-most-once&#x22; --> NS
  CH -- &#x22;at-most-once&#x22; --> AS

  class CH events
  class OS,NS,AS client"
/>

*One publisher fans out every order event to both subscribers at-most-once.*

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* SDK installed ([Getting Started](/learn/events/getting-started))

## Steps [#steps]

<Steps>
  <Step>
    ### Set Up the Publisher [#set-up-the-publisher]

    The publisher sends order events with a JSON body, metadata, and tags for downstream filtering.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="order_publisher.go"
        package main

        import (
            "context"
            "fmt"
            "log"
            "time"

            "github.com/kubemq-io/kubemq-go/v2"
        )

        func main() {
            ctx := context.Background()
            client, err := kubemq.NewClient(ctx,
                kubemq.WithAddress("localhost", 50000),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer client.Close()

            orders := []struct {
                ID     string
                Amount float64
                Region string
            }{
                {"ORD-001", 99.99, "us-east"},
                {"ORD-002", 249.50, "eu-west"},
                {"ORD-003", 15.00, "us-east"},
            }

            for _, order := range orders {
                body := fmt.Sprintf(
                    `{"orderId":"%s","amount":%.2f,"region":"%s"}`,
                    order.ID, order.Amount, order.Region)

                err = client.SendEvent(ctx, kubemq.NewEvent().
                    SetChannel("order-events").
                    SetMetadata("order.created").
                    SetBody([]byte(body)).
                    SetTags(map[string]string{"region": order.Region}),
                )
                if err != nil {
                    log.Printf("Failed to send event for %s: %v", order.ID, err)
                    continue
                }
                log.Printf("Published order event: %s", order.ID)
                time.Sleep(500 * time.Millisecond)
            }
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="order_publisher.py"
        import json
        import time
        from kubemq.pubsub import Client as PubSubClient
        from kubemq.pubsub import EventMessage

        orders = [
            {"orderId": "ORD-001", "amount": 99.99, "region": "us-east"},
            {"orderId": "ORD-002", "amount": 249.50, "region": "eu-west"},
            {"orderId": "ORD-003", "amount": 15.00, "region": "us-east"},
        ]

        client = PubSubClient(address="localhost:50000")
        for order in orders:
            try:
                client.send_event(
                    EventMessage(
                        channel="order-events",
                        metadata="order.created",
                        body=json.dumps(order).encode("utf-8"),
                        tags={"region": order["region"]},
                    )
                )
                print(f"Published order event: {order['orderId']}")
            except Exception as e:
                print(f"Failed to send event for {order['orderId']}: {e}")
            time.sleep(0.5)
        client.close()
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="order_publisher.js"
        const { KubeMQClient } = require("kubemq-js");

        const client = new KubeMQClient({ address: "localhost:50000" });

        const orders = [
          { orderId: "ORD-001", amount: 99.99, region: "us-east" },
          { orderId: "ORD-002", amount: 249.50, region: "eu-west" },
          { orderId: "ORD-003", amount: 15.00, region: "us-east" },
        ];

        for (const order of orders) {
          try {
            await client.sendEvent({
              channel: "order-events",
              metadata: "order.created",
              body: Buffer.from(JSON.stringify(order)),
              tags: { region: order.region },
            });
            console.log(`Published order event: ${order.orderId}`);
          } catch (err) {
            console.error(`Failed to send event for ${order.orderId}:`, err);
          }
          await new Promise((r) => setTimeout(r, 500));
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="OrderPublisher.java"
        PubSubClient client = PubSubClient.builder()
            .address("localhost:50000")
            .clientId("order-publisher")
            .build();

        String[][] orders = {
            {"ORD-001", "99.99", "us-east"},
            {"ORD-002", "249.50", "eu-west"},
            {"ORD-003", "15.00", "us-east"},
        };

        for (String[] order : orders) {
            String body = String.format(
                "{\"orderId\":\"%s\",\"amount\":%s,\"region\":\"%s\"}",
                order[0], order[1], order[2]);
            try {
                client.sendEventsMessage(EventMessage.builder()
                    .channel("order-events")
                    .metadata("order.created")
                    .body(body.getBytes())
                    .tags(Map.of("region", order[2]))
                    .build());
                System.out.println("Published order event: " + order[0]);
            } catch (Exception e) {
                System.err.println("Failed to send: " + e.getMessage());
            }
            Thread.sleep(500);
        }
        client.close();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="OrderPublisher.cs"
        await using var client = new KubeMQClient(new KubeMQClientOptions());
        await client.ConnectAsync();

        var orders = new[]
        {
            new { Id = "ORD-001", Amount = 99.99, Region = "us-east" },
            new { Id = "ORD-002", Amount = 249.50, Region = "eu-west" },
            new { Id = "ORD-003", Amount = 15.00, Region = "us-east" },
        };

        foreach (var order in orders)
        {
            var body = $"{{\"orderId\":\"{order.Id}\",\"amount\":{order.Amount},\"region\":\"{order.Region}\"}}";
            await client.SendEventAsync(new EventMessage
            {
                Channel = "order-events",
                Metadata = "order.created",
                Body = Encoding.UTF8.GetBytes(body),
                Tags = new Dictionary<string, string> { ["region"] = order.Region },
            });
            Console.WriteLine($"Published order event: {order.Id}");
            await Task.Delay(500);
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="OrderPublisher.kt"
        val client = PubSubClient("localhost:50000")

        data class Order(val id: String, val amount: Double, val region: String)
        val orders = listOf(
            Order("ORD-001", 99.99, "us-east"),
            Order("ORD-002", 249.50, "eu-west"),
            Order("ORD-003", 15.00, "us-east"),
        )

        for (order in orders) {
            val body = """{"orderId":"${order.id}","amount":${order.amount},"region":"${order.region}"}"""
            client.sendEvent(EventMessage(
                channel = "order-events",
                metadata = "order.created",
                body = body.toByteArray(),
                tags = mapOf("region" to order.region),
            ))
            println("Published order event: ${order.id}")
            Thread.sleep(500)
        }
        client.close()
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="order_publisher.cpp"
        #include <kubemq/client.h>
        #include <iostream>
        #include <thread>
        #include <chrono>

        auto client = kubemq::PubSubClient("localhost:50000");

        struct Order { std::string id; double amount; std::string region; };
        std::vector<Order> orders = {
            {"ORD-001", 99.99, "us-east"},
            {"ORD-002", 249.50, "eu-west"},
            {"ORD-003", 15.00, "us-east"},
        };

        for (const auto& order : orders) {
            kubemq::EventMessage event;
            event.channel = "order-events";
            event.metadata = "order.created";
            event.body = "{\"orderId\":\"" + order.id + "\",\"amount\":" +
                         std::to_string(order.amount) + ",\"region\":\"" + order.region + "\"}";
            event.tags["region"] = order.region;

            client.sendEvent(event);
            std::cout << "Published order event: " << order.id << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="order_publisher.rs"
        use kubemq::prelude::*;
        use kubemq::EventBuilder;
        use std::collections::HashMap;
        use std::time::Duration;

        #[tokio::main]
        async fn main() -> kubemq::Result<()> {
            let client = KubemqClient::builder()
                .host("localhost")
                .port(50000)
                .build()
                .await?;

            let orders = [
                ("ORD-001", 99.99, "us-east"),
                ("ORD-002", 249.50, "eu-west"),
                ("ORD-003", 15.00, "us-east"),
            ];

            for (id, amount, region) in orders {
                let body = format!(
                    r#"{{"orderId":"{id}","amount":{amount},"region":"{region}"}}"#
                );
                let mut tags = HashMap::new();
                tags.insert("region".to_string(), region.to_string());

                let event = EventBuilder::new()
                    .channel("order-events")
                    .metadata("order.created")
                    .body(body.into_bytes())
                    .tags(tags)
                    .build();

                match client.send_event(event).await {
                    Ok(_) => println!("Published order event: {id}"),
                    Err(e) => eprintln!("Failed to send event for {id}: {e}"),
                }
                tokio::time::sleep(Duration::from_millis(500)).await;
            }

            client.close().await?;
            Ok(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="order_publisher.rb"
        require 'kubemq'
        require 'json'

        client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'order-publisher')

        orders = [
          { orderId: 'ORD-001', amount: 99.99, region: 'us-east' },
          { orderId: 'ORD-002', amount: 249.50, region: 'eu-west' },
          { orderId: 'ORD-003', amount: 15.00, region: 'us-east' },
        ]

        orders.each do |order|
          begin
            msg = KubeMQ::PubSub::EventMessage.new(
              channel: 'order-events',
              metadata: 'order.created',
              body: order.to_json,
              tags: { 'region' => order[:region] }
            )
            client.send_event(msg)
            puts "Published order event: #{order[:orderId]}"
          rescue KubeMQ::Error => e
            puts "Failed to send event for #{order[:orderId]}: #{e.message}"
          end
          sleep 0.5
        end

        client.close
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="order_publisher.exs"
        {:ok, client} =
          KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-publisher")

        orders = [
          %{order_id: "ORD-001", amount: 99.99, region: "us-east"},
          %{order_id: "ORD-002", amount: 249.50, region: "eu-west"},
          %{order_id: "ORD-003", amount: 15.00, region: "us-east"}
        ]

        for order <- orders do
          body =
            ~s({"orderId":"#{order.order_id}","amount":#{order.amount},"region":"#{order.region}"})

          event =
            KubeMQ.Event.new(
              channel: "order-events",
              metadata: "order.created",
              body: body,
              tags: %{"region" => order.region}
            )

          case KubeMQ.Client.send_event(client, event) do
            :ok -> IO.puts("Published order event: #{order.order_id}")
            {:error, err} -> IO.puts("Failed to send event for #{order.order_id}: #{err.message}")
          end

          Process.sleep(500)
        end

        KubeMQ.Client.close(client)
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Set Up Subscriber A [#set-up-subscriber-a]

    The notification service processes every order event.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="notification_service.go"
        package main

        import (
            "context"
            "fmt"
            "log"

            "github.com/kubemq-io/kubemq-go/v2"
        )

        func main() {
            ctx := context.Background()
            client, err := kubemq.NewClient(ctx,
                kubemq.WithAddress("localhost", 50000),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer client.Close()

            sub, err := client.SubscribeToEvents(ctx, "order-events", "",
                kubemq.WithOnEvent(func(event *kubemq.Event) {
                    fmt.Printf("[Notification] New order: %s | metadata: %s\n",
                        string(event.Body), event.Metadata)
                }),
                kubemq.WithOnError(func(err error) {
                    log.Println("[Notification] Error:", err)
                }),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer sub.Unsubscribe()

            log.Println("[Notification] Service listening...")
            <-ctx.Done()
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="notification_service.py"
        import time
        from kubemq.pubsub import Client as PubSubClient
        from kubemq.pubsub import EventsSubscription, CancellationToken

        def on_event(event):
            print(f"[Notification] New order: "
                  f"{event.body.decode('utf-8')} | metadata: {event.metadata}")

        client = PubSubClient(address="localhost:50000")
        client.subscribe_to_events(
            subscription=EventsSubscription(
                channel="order-events",
                on_receive_event_callback=on_event,
                on_error_callback=lambda e: print(f"[Notification] Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        print("[Notification] Service listening...")
        time.sleep(300)
        client.close()
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="notification_service.js"
        const { KubeMQClient } = require("kubemq-js");

        const client = new KubeMQClient({ address: "localhost:50000" });

        client.subscribeToEvents({
          channel: "order-events",
          onEvent: (msg) =>
            console.log(
              `[Notification] New order: ${Buffer.from(msg.body).toString()}`
            ),
          onError: (err) => console.error("[Notification] Error:", err.message),
        });

        console.log("[Notification] Service listening...");
        ```
      </Tab>

      <Tab value="Java">
        ```java title="NotificationService.java"
        PubSubClient client = PubSubClient.builder()
            .address("localhost:50000")
            .clientId("notification-service")
            .build();

        client.subscribeToEvents(EventsSubscription.builder()
            .channel("order-events")
            .onReceiveEventCallback(event ->
                System.out.println("[Notification] New order: "
                    + new String(event.getBody())))
            .onErrorCallback(err ->
                System.err.println("[Notification] Error: " + err.getMessage()))
            .build());

        System.out.println("[Notification] Service listening...");
        Thread.sleep(300_000);
        client.close();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="NotificationService.cs"
        await using var client = new KubeMQClient(new KubeMQClientOptions());
        await client.ConnectAsync();

        Console.WriteLine("[Notification] Service listening...");
        await foreach (var msg in client.SubscribeToEventsAsync(
            new EventsSubscription { Channel = "order-events" }))
        {
            Console.WriteLine($"[Notification] New order: "
                + $"{Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="NotificationService.kt"
        val client = PubSubClient("localhost:50000")

        client.subscribeToEvents(
            channel = "order-events",
            onEvent = { event ->
                println("[Notification] New order: ${String(event.body)}")
            },
            onError = { err ->
                System.err.println("[Notification] Error: ${err.message}")
            }
        )

        println("[Notification] Service listening...")
        Thread.sleep(300_000)
        client.close()
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="notification_service.cpp"
        auto client = kubemq::PubSubClient("localhost:50000");

        client.subscribeToEvents("order-events", "",
            [](const kubemq::Event& event) {
                std::cout << "[Notification] New order: " << event.body << std::endl;
            },
            [](const std::string& err) {
                std::cerr << "[Notification] Error: " << err << std::endl;
            }
        );

        std::cout << "[Notification] Service listening..." << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(300));
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="notification_service.rs"
        use kubemq::prelude::*;
        use kubemq::Subscription;
        use std::time::Duration;

        #[tokio::main]
        async fn main() -> kubemq::Result<()> {
            let client = KubemqClient::builder()
                .host("localhost")
                .port(50000)
                .build()
                .await?;

            // Subscribe with an empty group -- every subscriber receives every event
            let sub: Subscription = client
                .subscribe_to_events(
                    "order-events",
                    "",
                    |event| {
                        Box::pin(async move {
                            println!(
                                "[Notification] New order: {} | metadata: {}",
                                String::from_utf8_lossy(&event.body),
                                event.metadata
                            );
                        })
                    },
                    None,
                )
                .await?;

            println!("[Notification] Service listening...");
            tokio::time::sleep(Duration::from_secs(300)).await;

            sub.unsubscribe().await;
            client.close().await?;
            Ok(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="notification_service.rb"
        require 'kubemq'

        client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'notification-service')
        cancel = KubeMQ::CancellationToken.new

        sub = KubeMQ::PubSub::EventsSubscription.new(channel: 'order-events')
        client.subscribe_to_events(
          sub,
          cancellation_token: cancel,
          on_error: ->(e) { puts "[Notification] Error: #{e.message}" }
        ) do |event|
          puts "[Notification] New order: #{event.body} | metadata: #{event.metadata}"
        end

        puts '[Notification] Service listening...'
        sleep 300

        cancel.cancel
        client.close
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="notification_service.exs"
        {:ok, client} =
          KubeMQ.Client.start_link(address: "localhost:50000", client_id: "notification-service")

        {:ok, _sub} =
          KubeMQ.Client.subscribe_to_events(client, "order-events",
            on_event: fn event ->
              IO.puts("[Notification] New order: #{event.body} | metadata: #{event.metadata}")
            end,
            on_error: fn err -> IO.puts("[Notification] Error: #{err.message}") end
          )

        IO.puts("[Notification] Service listening...")
        Process.sleep(300_000)

        KubeMQ.Client.close(client)
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Set Up Subscriber B [#set-up-subscriber-b]

    A second subscriber receives the same events independently for analytics processing.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="analytics_service.go"
        sub, err := client.SubscribeToEvents(ctx, "order-events", "",
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[Analytics] Processing: %s\n", string(event.Body))
            }),
            kubemq.WithOnError(func(err error) {
                log.Println("[Analytics] Error:", err)
            }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="analytics_service.py"
        def on_event(event):
            print(f"[Analytics] Processing: {event.body.decode('utf-8')}")

        client.subscribe_to_events(
            subscription=EventsSubscription(
                channel="order-events",
                on_receive_event_callback=on_event,
                on_error_callback=lambda e: print(f"[Analytics] Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        print("[Analytics] Service listening...")
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="analytics_service.js"
        client.subscribeToEvents({
          channel: "order-events",
          onEvent: (msg) =>
            console.log(
              `[Analytics] Processing: ${Buffer.from(msg.body).toString()}`
            ),
          onError: (err) => console.error("[Analytics] Error:", err.message),
        });

        console.log("[Analytics] Service listening...");
        ```
      </Tab>

      <Tab value="Java">
        ```java title="AnalyticsService.java"
        client.subscribeToEvents(EventsSubscription.builder()
            .channel("order-events")
            .onReceiveEventCallback(event ->
                System.out.println("[Analytics] Processing: "
                    + new String(event.getBody())))
            .onErrorCallback(err ->
                System.err.println("[Analytics] Error: " + err.getMessage()))
            .build());

        System.out.println("[Analytics] Service listening...");
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="AnalyticsService.cs"
        Console.WriteLine("[Analytics] Service listening...");
        await foreach (var msg in client.SubscribeToEventsAsync(
            new EventsSubscription { Channel = "order-events" }))
        {
            Console.WriteLine($"[Analytics] Processing: "
                + $"{Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="AnalyticsService.kt"
        client.subscribeToEvents(
            channel = "order-events",
            onEvent = { event ->
                println("[Analytics] Processing: ${String(event.body)}")
            },
            onError = { err ->
                System.err.println("[Analytics] Error: ${err.message}")
            }
        )
        println("[Analytics] Service listening...")
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="analytics_service.cpp"
        client.subscribeToEvents("order-events", "",
            [](const kubemq::Event& event) {
                std::cout << "[Analytics] Processing: " << event.body << std::endl;
            },
            [](const std::string& err) {
                std::cerr << "[Analytics] Error: " << err << std::endl;
            }
        );
        std::cout << "[Analytics] Service listening..." << std::endl;
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="analytics_service.rs"
        // Same channel, empty group -- this subscriber receives the same events
        // independently of the notification service.
        let sub: Subscription = client
            .subscribe_to_events(
                "order-events",
                "",
                |event| {
                    Box::pin(async move {
                        println!(
                            "[Analytics] Processing: {}",
                            String::from_utf8_lossy(&event.body)
                        );
                    })
                },
                None,
            )
            .await?;
        println!("[Analytics] Service listening...");
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="analytics_service.rb"
        # Same channel, no group -- this subscriber receives the same events
        # independently of the notification service.
        sub = KubeMQ::PubSub::EventsSubscription.new(channel: 'order-events')
        client.subscribe_to_events(
          sub,
          cancellation_token: cancel,
          on_error: ->(e) { puts "[Analytics] Error: #{e.message}" }
        ) do |event|
          puts "[Analytics] Processing: #{event.body}"
        end
        puts '[Analytics] Service listening...'
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="analytics_service.exs"
        # Same channel, no group -- this subscriber receives the same events
        # independently of the notification service.
        {:ok, _sub} =
          KubeMQ.Client.subscribe_to_events(client, "order-events",
            on_event: fn event ->
              IO.puts("[Analytics] Processing: #{event.body}")
            end,
            on_error: fn err -> IO.puts("[Analytics] Error: #{err.message}") end
          )

        IO.puts("[Analytics] Service listening...")
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Observe Fan-Out [#observe-fan-out]

    Start both subscribers in separate terminals, then run the publisher. Both subscribers receive every event. No acknowledgment needed.

    **Notification Service output:**

    ```text
    [Notification] New order: {"orderId":"ORD-001","amount":99.99,"region":"us-east"}
    [Notification] New order: {"orderId":"ORD-002","amount":249.50,"region":"eu-west"}
    [Notification] New order: {"orderId":"ORD-003","amount":15.00,"region":"us-east"}
    ```

    **Analytics Service output:**

    ```text
    [Analytics] Processing: {"orderId":"ORD-001","amount":99.99,"region":"us-east"}
    [Analytics] Processing: {"orderId":"ORD-002","amount":249.50,"region":"eu-west"}
    [Analytics] Processing: {"orderId":"ORD-003","amount":15.00,"region":"us-east"}
    ```
  </Step>
</Steps>

## How Fan-Out Works [#how-fan-out-works]

<Mermaid
  chart="sequenceDiagram
    participant P as Publisher
    participant K as KubeMQ
    participant A as Subscriber A
    participant B as Subscriber B
    P->>K: publish event
    K->>A: deliver
    K->>B: deliver
    Note over A,B: Both receive every event"
/>

*KubeMQ delivers each published event to every active subscriber independently — no acknowledgment, no ordering coordination between subscribers.*

## Key Points [#key-points]

* **No acknowledgment needed** — fire-and-forget delivery
* **Missed messages** — if a subscriber is offline, events sent while it was disconnected are lost
* **Metadata and tags** — propagated to all subscribers for downstream filtering and routing

<Callout type="warn">
  If a subscriber is slow to process events, messages may be dropped after the write deadline (default 2 seconds). See [Handle Slow Consumers](/learn/events/how-to/handle-slow-consumers) for mitigation strategies.
</Callout>

## Next Steps [#next-steps]

<Cards>
  <Card title="Consumer Groups" href="/learn/events/tutorials/consumer-groups" description="Distribute events across a group." />

  <Card title="Wildcard Subscriptions" href="/learn/events/tutorials/wildcard-subscriptions" description="Subscribe to multiple channels." />
</Cards>