Bus plugin

Bus gives processes logical addresses and loads application types, process endpoints and connection relationships. bus_module exposes configuration queries. msg_module sends messages by application type, Bus ID or session, while the Net plugin manages the underlying connections.

Addresses and responsibilities

A Bus address has four components, channel.zone.app.instance, such as 1.1.31.1. Each occupies 8 bits in the combined bus_id_t. It identifies a process, rather than an IP address or listening port.

Configure processes and relationships

bus_module::init() reads four sections from bus.plugin.conf:

The following example describes Game and World processes. External and intranet endpoints use different ports to avoid conflicts when both listeners start.

<xml>
    <applications>
        <application name="world" type="30" />
        <application name="game" type="31" />
    </applications>
    <relations>
        <relation source="game" target="world" />
    </relations>
    <processes>
        <proc busid="1.1.30.1" endpoint_server="tcp://127.0.0.1:9101"
            endpoint_intranet="tcp://127.0.0.1:9111" max_connection="100" thread_num="1" />
        <proc busid="1.1.31.1" endpoint_server="tcp://127.0.0.1:9102"
            endpoint_intranet="tcp://127.0.0.1:9112" max_connection="100" thread_num="1" />
    </processes>
    <center ip="127.0.0.1" port="8500" service_name="ark-service"
        check_interval="10s" check_timeout="3s" />
</xml>

This is a Bus configuration example, not a complete two-process launch configuration. Applications must also load Bus, Net and their dependencies, set each process’s Bus ID, start listeners and initiate connections. relations supplies target types; reading a relationship does not automatically create a connection.

Configuring center does not enable dynamic registration or discovery. The Consul registration and health-based discovery paths in the current network service manager remain commented out. This configuration alone cannot discover target processes.

Query local and target processes

After module initialization, obtain the Bus interface from the plugin manager. This function collects statically configured processes for the current application’s target types.

#include <vector>

#include "ark/bus/bus_module.hpp"
#include "ark/plugin_manager.hpp"

std::vector<ark::proc_config> configured_targets(ark::plugin_manager& manager) {
    auto* bus = manager.find_module<ark::bus_module>();
    if (bus == nullptr)
    {
        return {};
    }

    std::vector<ark::app_type> types;
    if (!bus->get_target_bus_relations(types))
    {
        return {};
    }

    std::vector<ark::proc_config> result;
    for (auto type : types)
    {
        std::vector<const ark::proc_config*> processes;
        bus->get_other_proc(type, processes);
        for (const auto* process : processes)
        {
            result.push_back(*process);
        }
    }
    return result;
}

get_other_proc(type, list) appends pointers to its output list and returns the resulting total list length. The example starts with an empty list each time and copies the records, avoiding externally retained pointers into module state. This query filters by application type without also restricting channel or zone; applications can apply those filters when needed.

Establish a TCP connection

The target process first starts its listeners with net_service_manager_module::create_server(). The following function reads the target’s configured intranet endpoint and queues a TCP connection through the network service manager.

#include "ark/bus/bus_module.hpp"
#include "ark/net/net_service_manager_module.hpp"
#include "ark/plugin_manager.hpp"

bool queue_tcp_connection(ark::plugin_manager& manager, ark::bus_id_t target_id) {
    auto* bus = manager.find_module<ark::bus_module>();
    auto* network = manager.find_module<ark::net_service_manager_module>();
    if (bus == nullptr || network == nullptr)
    {
        return false;
    }

    ark::proc_config target;
    if (!bus->get_other_proc(target_id, target))
    {
        return false;
    }

    const auto& endpoint = target.intranet_ep;
    if (endpoint.proto() != ark::proto_type::tcp || endpoint.get_ip().empty() || endpoint.get_port() == 0)
    {
        return false;
    }

    return network->create_client_service(
        ark::bus_addr(target_id), ark::proto_type::tcp, endpoint.get_ip(), endpoint.get_port());
}

true means the connection request was queued in the client service and will be processed by subsequent update() calls. It does not mean the connection has succeeded. Observe connection events with the client service’s reg_net_event_callback() and register message handlers with reg_msg_callback(). The receiver registers handlers on the corresponding server service.

The current client service handles reconnection through its connection state machine, with a 30-second retry interval in the implementation. This is not a configurable backoff policy in the XML file. Reconnection does not replay application messages whose earlier send failed.

Send and receive messages

Once a connection is available, send application-defined Protobuf messages through msg_module. The following function accepts the caller’s message ID and message type, without defining an application protocol.

#include "ark/bus/msg_module.hpp"

bool send_to_process(ark::msg_module& messages, ark::bus_id_t target_id, ark::msg_id_t message_id,
    const google::protobuf::Message& body) {
    return messages.send_msg_by_bus_id(target_id, message_id, body);
}

Inter-process headers carry the message ID, source / target Bus IDs and Actor ID. Sending returns false on failure, for example when a service or connection is unavailable or serialization fails. true does not mean the remote application has processed the message. Request/response handling, timeouts and idempotency belong to the application protocol.

The receive flow is: the network service extracts a net_msg → invokes the handler registered for its message ID → the handler parses its application message with msg_module::decode_protobuf(). The parser returns false on a Protobuf parse failure. read_message_body() copies the raw body and retrieves the Actor ID. A callback’s net_msg* is borrowed data; copy the required contents before deferring work.

Coverage and planned work

These examples cover configuration queries, an explicit TCP connection request and sending through an existing connection. They are not an accepted multi-process deployment example, and do not imply that every protocol is integrated with the service manager. See the Net plugin for transport integration.

TODO: provide a standalone two-process example and tests covering connection setup, message round trips, process restarts, reconnection and cross-host routing. Implement and validate dynamic registration / discovery separately.

API sources

• Bus addresses and process configuration
• Bus interface
• Message interface
• Network service management interface
• Network build and tests