Quickstart

This guide gets you from zero to producing and consuming messages in under a minute.

Prerequisites

Download the latest danube-broker and danube-cli binaries from the releases page, or pull the container image.

Binary

1. Start the broker

   Terminal 1

   danube-broker --mode standalone --data-dir ./danube-data

   Broker ready on 127.0.0.1:6650, admin on 127.0.0.1:50051. No config file needed.

2. Produce messages

   Terminal 2

   danube-cli produce \
     -s http://127.0.0.1:6650 \
     -t /default/quickstart \
     -c 10 \
     -m "Hello, Danube!"

3. Consume messages

   Terminal 3

   danube-cli consume \
     -s http://127.0.0.1:6650 \
     -t /default/quickstart \
     -m my-subscription

   You should see the 10 messages arrive at the consumer.

Docker

1. Start the broker

   Terminal 1

   docker run -p 6650:6650 -p 50051:50051 \
     ghcr.io/danube-messaging/danube-broker:latest \
     --mode standalone --data-dir /data \
     --broker-addr 0.0.0.0:6650 --admin-addr 0.0.0.0:50051

2. Produce messages

   Terminal 2

   danube-cli produce \
     -s http://127.0.0.1:6650 \
     -t /default/quickstart \
     -c 10 \
     -m "Hello, Danube!"

3. Consume messages

   Terminal 3

   danube-cli consume \
     -s http://127.0.0.1:6650 \
     -t /default/quickstart \
     -m my-subscription

Tip

Data persists across restarts in the ./danube-data directory. To start fresh, remove it and restart the broker.

What Just Happened?

1. The broker started in standalone mode, a single-node setup with zero configuration.
2. danube-cli produce created a topic /default/quickstart and published 10 messages.
3. danube-cli consume created a subscription my-subscription and received the messages via best-effort dispatch.

Try With a Client Library

Rust

examples/quickstart.rs

use danube_client::{DanubeClient, SubType};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = DanubeClient::builder()
        .service_url("http://127.0.0.1:6650")
        .build()
        .await?;

    // Producer
    let mut producer = client
        .new_producer()
        .with_topic("/default/quickstart")
        .with_name("my-producer")
        .build()?;

    producer.create().await?;
    producer.send("Hello from Rust!".as_bytes().to_vec(), None).await?;

    // Consumer
    let mut consumer = client
        .new_consumer()
        .with_topic("/default/quickstart")
        .with_consumer_name("my-consumer")
        .with_subscription("my-sub")
        .with_subscription_type(SubType::Exclusive)
        .build()?;

    consumer.subscribe().await?;

    let mut stream = consumer.receive().await?;
    while let Some(message) = stream.recv().await {
        println!("Received: {}", String::from_utf8_lossy(&message.payload));
        consumer.ack(&message).await?;
    }

    Ok(())
}

Go

examples/quickstart.go

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/danube-messaging/danube-go"
)

func main() {
    client, err := danube.NewClient().ServiceURL("127.0.0.1:6650").Build()
    if err != nil {
        log.Fatalf("Failed to create client: %v", err)
    }

    ctx := context.Background()

    // Producer
    producer, err := client.NewProducer().
        WithName("my-producer").
        WithTopic("/default/quickstart").
        Build()
    if err != nil {
        log.Fatalf("Failed to build producer: %v", err)
    }

    if err := producer.Create(ctx); err != nil {
        log.Fatalf("Failed to create producer: %v", err)
    }

    producer.Send(ctx, []byte("Hello from Go!"), nil)

    // Consumer
    consumer, err := client.NewConsumer().
        WithConsumerName("my-consumer").
        WithTopic("/default/quickstart").
        WithSubscription("my-sub").
        WithSubscriptionType(danube.Exclusive).
        Build()
    if err != nil {
        log.Fatalf("Failed to build consumer: %v", err)
    }

    if err := consumer.Subscribe(ctx); err != nil {
        log.Fatalf("Failed to subscribe: %v", err)
    }

    stream, err := consumer.Receive(ctx)
    if err != nil {
        log.Fatalf("Failed to receive: %v", err)
    }

    for msg := range stream {
        fmt.Printf("Received: %s\n", string(msg.GetPayload()))
        consumer.Ack(ctx, msg)
    }
}

Java

examples/Quickstart.java

import com.danubemessaging.client.DanubeClient;
import com.danubemessaging.client.Producer;
import com.danubemessaging.client.Consumer;
import com.danubemessaging.client.SubType;
import com.danubemessaging.client.model.StreamMessage;
import java.util.Map;
import java.util.concurrent.Flow;

public class Quickstart {
    public static void main(String[] args) throws Exception {
        DanubeClient client = DanubeClient.builder()
                .serviceUrl("http://127.0.0.1:6650")
                .build();

        // Producer
        Producer producer = client.newProducer()
                .withTopic("/default/quickstart")
                .withName("my-producer")
                .build();

        producer.create();
        producer.send("Hello from Java!".getBytes(), Map.of());

        // Consumer
        Consumer consumer = client.newConsumer()
                .withTopic("/default/quickstart")
                .withConsumerName("my-consumer")
                .withSubscription("my-sub")
                .withSubscriptionType(SubType.EXCLUSIVE)
                .build();

        consumer.subscribe();

        consumer.receive().subscribe(new Flow.Subscriber<>() {
            @Override public void onSubscribe(Flow.Subscription s) { s.request(Long.MAX_VALUE); }

            @Override
            public void onNext(StreamMessage msg) {
                System.out.println("Received: " + new String(msg.payload()));
                consumer.ack(msg);
            }

            @Override public void onError(Throwable t) { t.printStackTrace(); }
            @Override public void onComplete() {}
        });

        Thread.sleep(5000); // Wait for messages
        client.close();
    }
}

Python

examples/quickstart.py

import asyncio
from danube import DanubeClientBuilder, SubType

async def main():
    client = await (
        DanubeClientBuilder()
        .service_url("http://127.0.0.1:6650")
        .build()
    )

    # Producer
    producer = (
        client.new_producer()
        .with_topic("/default/quickstart")
        .with_name("my-producer")
        .build()
    )

    await producer.create()
    await producer.send("Hello from Python!".encode())

    # Consumer
    consumer = (
        client.new_consumer()
        .with_topic("/default/quickstart")
        .with_consumer_name("my-consumer")
        .with_subscription("my-sub")
        .with_subscription_type(SubType.EXCLUSIVE)
        .build()
    )

    await consumer.subscribe()

    queue = await consumer.receive()
    while True:
        message = await queue.get()
        print(f"Received: {message.payload.decode()}")
        await consumer.ack(message)

asyncio.run(main())

Next Steps

• Broker Modes: Learn about Standalone, Cluster, and Edge modes.
• Topics & Partitions: Understand how messages are organized.
• Subscriptions: Explore Exclusive, Shared, Failover, and Key-Shared patterns.