MsgHub

// Without MsgHub (verbose and error-prone)
Msg aliceReply = alice.call().block();
bob.observe(aliceReply).block();
charlie.observe(aliceReply).block();

Msg bobReply = bob.call().block();
alice.observe(bobReply).block();
charlie.observe(bobReply).block();

// With MsgHub (clean and automatic)
try (MsgHub hub = MsgHub.builder()
        .participants(alice, bob, charlie)
        .build()) {
    hub.enter().block();

    alice.call().block();  // Bob and Charlie automatically receive this
    bob.call().block();    // Alice and Charlie automatically receive this
}

import io.agentscope.core.ReActAgent;
import io.agentscope.extensions.model.dashscope.formatter.DashScopeMultiAgentFormatter;
import io.agentscope.core.memory.InMemoryMemory;
import io.agentscope.core.message.Msg;
import io.agentscope.core.message.MsgRole;
import io.agentscope.core.message.TextBlock;
import io.agentscope.extensions.model.dashscope.DashScopeChatModel;
import io.agentscope.core.pipeline.MsgHub;

// Create model with MultiAgentFormatter (important!)
DashScopeChatModel model = DashScopeChatModel.builder()
        .apiKey(System.getenv("DASHSCOPE_API_KEY"))
        .modelName("qwen3-max")
        .formatter(new DashScopeMultiAgentFormatter())
        .build();

// Create agents
ReActAgent alice = ReActAgent.builder()
        .name("Alice")
        .sysPrompt("You are Alice, a friendly teacher. Be brief in your responses.")
        .model(model)
        .memory(new InMemoryMemory())
        .build();

ReActAgent bob = ReActAgent.builder()
        .name("Bob")
        .sysPrompt("You are Bob, a curious student. Be brief in your responses.")
        .model(model)
        .memory(new InMemoryMemory())
        .build();

ReActAgent charlie = ReActAgent.builder()
        .name("Charlie")
        .sysPrompt("You are Charlie, a thoughtful observer. Be brief in your responses.")
        .model(model)
        .memory(new InMemoryMemory())
        .build();

// Create announcement message
Msg announcement = Msg.builder()
        .name("system")
        .role(MsgRole.SYSTEM)
        .content(TextBlock.builder()
                .text("Welcome to the discussion! Please introduce yourself briefly.")
                .build())
        .build();

// Use MsgHub with try-with-resources
try (MsgHub hub = MsgHub.builder()
        .name("Introduction")
        .participants(alice, bob, charlie)
        .announcement(announcement)
        .enableAutoBroadcast(true)  // Default is true
        .build()) {

    // Enter the hub (broadcasts announcement to all participants)
    hub.enter().block();

    // Each agent introduces themselves
    // Their responses are automatically broadcast to others
    Msg aliceReply = alice.call().block();
    System.out.println("Alice: " + aliceReply.getTextContent());

    Msg bobReply = bob.call().block();
    System.out.println("Bob: " + bobReply.getTextContent());

    Msg charlieReply = charlie.call().block();
    System.out.println("Charlie: " + charlieReply.getTextContent());
}
// Hub is automatically closed, subscribers are cleaned up

MsgHub hub = MsgHub.builder()
        .participants(alice, bob)
        .build();

// Enter: sets up subscriptions and broadcasts announcements
hub.enter().block();

// ... conversation happens ...

// Exit: cleans up subscriptions
hub.exit().block();

try (MsgHub hub = MsgHub.builder()
        .participants(alice, bob)
        .build()) {

    hub.enter().block();

    // Alice and Bob talk
    alice.call().block();
    bob.call().block();

    // Add Charlie mid-conversation
    hub.add(charlie).block();

    // Now Charlie receives messages too
    alice.call().block();  // Charlie receives this
    charlie.call().block(); // Alice and Bob receive this
}

try (MsgHub hub = MsgHub.builder()
        .participants(alice, bob, charlie)
        .build()) {

    hub.enter().block();

    // All three talk
    alice.call().block();
    bob.call().block();
    charlie.call().block();

    // Remove Bob from the conversation
    hub.delete(bob).block();

    // Bob won't receive these messages
    alice.call().block();
    charlie.call().block();
}

try (MsgHub hub = MsgHub.builder()
        .participants(alice, bob, charlie)
        .enableAutoBroadcast(false)  // Disable automatic broadcasting
        .build()) {

    hub.enter().block();

    // Alice speaks, but others don't automatically receive it
    Msg aliceReply = alice.call().block();

    // Manually broadcast to all participants
    hub.broadcast(aliceReply).block();

    // Now Bob and Charlie have received Alice's message
    bob.call().block();
}

try (MsgHub hub = MsgHub.builder()
        .participants(alice, bob)
        .enableAutoBroadcast(true)
        .build()) {

    hub.enter().block();

    // Auto-broadcast is on
    alice.call().block();  // Bob automatically receives

    // Turn off auto-broadcast
    hub.setAutoBroadcast(false);

    Msg bobReply = bob.call().block();
    // Alice doesn't automatically receive Bob's reply

    // Manually broadcast specific messages
    hub.broadcast(bobReply).block();

    // Turn auto-broadcast back on
    hub.setAutoBroadcast(true);
}

List<Msg> messages = List.of(msg1, msg2, msg3);
hub.broadcast(messages).block();

MsgHub hub = MsgHub.builder()
        .participants(alice, bob, charlie)
        .announcement(announcement)
        .build();

// Fully reactive chain
hub.enter()
    .then(alice.call())
    .doOnSuccess(msg -> System.out.println("Alice: " + msg.getTextContent()))
    .then(bob.call())
    .doOnSuccess(msg -> System.out.println("Bob: " + msg.getTextContent()))
    .then(charlie.call())
    .doOnSuccess(msg -> System.out.println("Charlie: " + msg.getTextContent()))
    .then(hub.exit())
    .block();  // Only block once at the end