Message Bus

1. Overview

In scenarios where a game has a hosted backend system and only utilizes certain services from PGOS (such as DS Hosting), there may be a need for the game-hosted backend to communicate with the DS (Dedicated Server) or VS (Virtual Server), and to receive PGOS Events. Typically, DS/VS can communicate directly with the game-hosted backend through the public network. However, this requires the game-hosted backend to have a public network gateway. Furthermore, even if the game-hosted backend already has a public network gateway, there is still some adaptation work required for the game. Therefore, it is necessary for PGOS to provide a ready-to-use communication mechanism named Message Bus.

2. Architecture

If the game wants to utilize the Message Bus, the game-hosted backend needs to integrate the PgosMsgBusSDK. The DS can participate in the communication through PgosSDK, and the VS can participate in the communication through the PGOS backend HTTP API. Additionally, you can bind PGOS Events to the Message Bus through the PGOS portal console, enabling the pushing of PGOS Events to the game-hosted backend.

Supported scenarios:

DS push messages to the game-hosted backend (no response back): (1), (2), (3)
DS send RPC request to the game-hosted backend: (1), (2), (3), (4), (5)
VS (HTTP API) push messages to the game-hosted backend (no response back): (6), (2), (3)
VS (HTTP API) send RPC request to the game-hosted backend: (6), (2), (3), (4), (7)
PGOS Event push messages to the game-hosted backend (no response back): (8), (2), (3)
Game-hosted backend push message to DS (no response back): (3), (4), (5)
Game-hosted backend send RPC request to DS: (3), (4), (5), (1), (2), (3)

提示

The game can deploy multiple proxies, and the PGOS backend will strive to maintain a stable communication pipeline. For example, a DS instance will consistently communicate with one proxy instance rather than multiple proxy instances, unless that proxy instance goes offline.

For PGOS events, the PGOS backend will select a proxy based on a hash of the route_key, which is typically the instance ID of a business module, such as Player ID, Party ID, Lobby ID, etc. This is also aimed at keeping the communication pipeline as stable as possible.

3. PgosMsgBusSDK Integration

PgosMsgBusSDK is provided to the game in the form of source code, and it relies on gRPC. Currently, only the C++ version is available, other language versions will be provided gradually in the future. You can download the SDK from the PGOS portal download page. Here is the SDK package directory description:

├─scripts                        # some helper scripts such as installing gRPC, generating proto codes, and building the sample.
├─source                         # the c++ source codes of the sample, including the PgosMsgBusSDK source codes.
│  └─generated_proto_codes       # the generated gRPC c++ codes of the proto files.
│  └─pgos_msg_bus_sdk            # the PgosMsgBusSDK codes, you can copy this directory to your project to integrate.
│      ├─include                 # the PgosMsgBusSDK header files for the SDK user.
│      └─source                  # the PgosMsgBusSDK implementation codes.
│          └─protos              # gRPC proto files are responsible for the communication between SDK and msg bus.
├─tools                          # tools for build
│  └─cmake                       # CMake for windows & linux, convenient for users to build the sample.

Since the SDK is provided as source code, integrating the SDK only requires copying the pgos_msg_bus_sdk directory into your C++ project. Additionally, if you don't have gRPC installed locally, you will need to install it. Finally, make sure you regenerate the C++ codes based on the proto files using the local gRPC. As a reference, the sample project in the SDK package demonstrates an integration and build approach based on CMake, reading the following chapter if you are interested.

3.1 Integrating SDK with CMake

Taking the sample project in the SDK package as an example:

Step 1: Copy the {SDK Package}\source\pgos_msg_bus_sdk\ to the target project.
Step 2: Add the SDK and the generated proto c++ source codes to the CMakeLists.txt.
You can refer to the file: {SDK Package}\CMakeLists.txt.
Step 3: Install gRPC.
For Linux:
- If you have gRPC installed locally, you can use it directly.
- If you don't have gRPC installed locally, you will need to install it. Here is an installation guide. Another choice is that you can refer to: {SDK Package}\scripts\linux_install_grpc.sh, and make adjustments based on your actual needs.
For Windows:
- You can refer to the {SDK Package}\CMakeLists.txt, the gRPC installation work has been done inside.
Step 4: Generate the C++ codes based on the proto files.
The protocol file path: {SDK Package}\source\pgos_msg_bus_sdk\source\protos\pgos_msg_bus.proto
Please generate the proto source codes based on it, the target output directory: {SDK Package}\source\generated_proto_codes\
For Linux: you can refer to the {SDK Package}\scripts\linux_generate_proto_codes.sh , and make adjustments based on your actual needs.
For Windows: you can refer to the {SDK Package}\CMakeLists.txt, the code generation will be done in the PRE_BUILD event.
Step 5: Add your business with the PgosMsgBusSDK.
Please view PgosMsgBusSDK Usage for details.
Step 6: Build your C++ project
For Linux: you can refer to the {SDK Package}\scripts\linux_build_sample.sh
For Windows: you can refer to the {SDK Package}\scripts\windows_build_sample.bat , note that it relies on Visual Studio 2017.
Make adjustments based on your actual needs.

4. PgosMsgBusSDK Usage

Once you have integrated the PgosMsgBusSDK, you can create an SDK instance in your project source code and call the APIs it provides. You can refer to the MyPgosMsgBusSDKManager class in the {SDK Package}\source\sample.cpp, which provides a simple usage example for the SDK.

4.1 Get the SDK Instance

#include "pgos_msg_bus_sdk.h"

void SomeFunction() {
    // It is recommended to use the singleton instance of the 'PgosMsgBusSDK' class.
    auto pmbs_sdk = pgos::PgosMsgBusSDK::Singleton();
    
    ...
}

4.2 Initialize the SDK Instance

#include "pgos_msg_bus_sdk.h"

void SomeFunction() {
    pgos::InitSdkParams params;
    params.title_id = "your title id";                  // obtained your 'title_id' from PGOS portal console.
    params.title_region_id = "your title_region_id";    // obtained your 'title_region_id' from PGOS portal console.
    params.server_key = "your server_key";              // obtained your 'server_key' from PGOS portal console.
    params.wp_log = wp_log;
    params.wp_observer = wp_observer;
    if (!pgos::PgosMsgBusSDK::Singleton()->Init(params)) {
        LOG_ERROR("init PgosMsgBusSDK failed.");
    }
    
    ...
}

Key fields of the pgos::InitSdkParams:

wp_log: If you want to output PgosMsgBusSDK logs for future troubleshooting, you need to implement the IMsgBusSDKLog interface and assign a weak pointer to its instance to this field.

// PgosMsgBusSDK log abstract class.
// If you need to output the PgosMsgBusSDK logs, please implement and set it to InitSDKParams.
class IMsgBusSDKLog {
public:
    enum class LogLevel : uint8_t {Debug = 0, Info = 1, Warn = 2, Error = 3 };
    virtual void Log(LogLevel level, const char* log) = 0;  // Need to support multi-threaded calls.
};

wp_observer: If you want to listen to events of PgosMsgBusSDK, such as received messages from the DS or the SDK lost connection, you need to implement the IMsgBusSDKEventObserver interface and assign a weak pointer to its instance to this field.

// PgosMsgBusSDK event observer abstract class.
// If you need to output the PgosMsgBusSDK logs, please implement and set it to InitSDKParams.
class IMsgBusSDKEventObserver {
public:
    // the event will be triggered when a push message from DS is received.
    virtual void OnDsPushMsgReceived(const pgos::DsMsgEvt& evt) {};

    // the event will be triggered when a RPCRequest from DS is received, 
    // and the receiver needs to call the 'resp_cb' to give a response to the sender.
    // the 'resp_cb' can be called in the same thread or another thread.
    virtual void OnDsRpcRequestReceived(const pgos::DsMsgEvt& evt, 
        std::function<void(const std::string& resp)>&& resp_cb) {};

    // the event will be triggered when a push message from backend HTTP API is received.
    virtual void OnHttpApiPushMsgReceived(const pgos::HttpApiMsgEvt& evt) {};

    // the event will be triggered when a RPCRequest from backend HTTP API is received,
    // and the receiver needs to call the 'resp_cb' to give a response to the sender.
    // the 'resp_cb' can be called in the same thread or another thread.
    virtual void OnHttpApiRpcRequestReceived(const pgos::HttpApiMsgEvt& evt,
        std::function<void(const std::string& resp)>&& resp_cb) {};

    // the event will be triggered when a PGOS Event push message is received.
    virtual void OnPgosEventPushMsgReceived(const pgos::PgosEventMsgEvt& evt) {};

    // the event will be triggered when the connection to the msg bus is lost.
    // if this is not due to the 'Destroy' API being called, 
    // the SDK will automatically try to reconnect 3 times, and if it still fails, this event will be thrown.
    virtual void OnConnectionLost(const pgos::PmbSDKConnectionLostEvt& evt) {};
};

提示

For the events OnDsRpcRequestReceived and OnHttpApiRpcRequestReceived, if the calculation or retrieval of the response takes a significant amount of time, it is advisable to perform that task in a separate thread to prevent blocking the dispatch of other messages.

4.3 Call the APIs You Need

Currently, only the APIs for sending messages to DS is provided.

#include "pgos_msg_bus_sdk.h"

void SomeFunction() {
    auto pmbs_sdk = pgos::PgosMsgBusSDK::Singleton();

    // call API: PushMsgToDS
    std::string msg = "data_sent_via_PushMsgToDS_API";
    auto ret = pmbs_sdk->PushMsgToDS(battle_session_id, msg);
    LOG_INFO("RunSomeBusiness: PushMsgToDS, bs_id=(%s), msg=(%s), err_code=(%u), err_msg=(%s).",
        battle_session_id.c_str(), msg.c_str(), ret.err_code, ret.err_msg.c_str());

    // call API: RPCRequestToDS
    msg = "data_sent_via_RPCRequestToDS_API";
    uint32_t timeout_ms = 10 * 1000;
    auto rsp = pmbs_sdk->RPCRequestToDS(battle_session_id, msg, timeout_ms);
    LOG_INFO("RunSomeBusiness: RPCRequestToDS, bs_id=(%s), msg=(%s), err_code=(%u), err_msg=(%s), rsp=(%s).",
        battle_session_id.c_str(), msg.c_str(), rsp.err_code, rsp.err_msg.c_str(), rsp.respone.c_str());

    // call API: AsyncRPCRequestToDS
    msg = "data_sent_via_AsyncRPCRequestToDS_API";
    bool invoke_ret = pmbs_sdk->AsyncRPCRequestToDS(battle_session_id, msg, timeout_ms,
        [battle_session_id, msg](const pgos::RpcRespone& rsp) {
        LOG_INFO("RunSomeBusiness: AsyncRPCRequestToDS, bs_id=(%s), msg=(%s), err_code=(%u), err_msg=(%s), rsp=(%s).",
            battle_session_id.c_str(), msg.c_str(), rsp.err_code, rsp.err_msg.c_str(), rsp.respone.c_str());
        });
    LOG_INFO("RunSomeBusiness: AsyncRPCRequestToDS invoke %s", invoke_ret? "success": "failed");
    
    ...
}

4.4 Shutdown the SDK instance

Shutdown the PgosMsgBusSDK instance, which should be called during process shutdown if possible. NOTE: it should be called from the same thread as the Init API.

#include "pgos_msg_bus_sdk.h"

void OnProcessShutdown() {
    pgos::PgosMsgBusSDK::Singleton()->Destroy();
    
    ...
}

4.5 Key Error Handling

Error Code	Handling Suggestion
kPmbSdkNotInit (600000)	the sdk has not been initiated, please do the init before sending msg.
kPmbSdkInvalidParameter (600001)	The API input parameter or one of its field values is invalid, please view the error message for more details.
kPmbSdkBadAlloc (600002)	failed to allocate memory for the sdk, which means an os system error has occurred.
kPmbSdkWriteTimeout (600003)	the specified timeout period was reached, but the msg did not complete the write operation. and the send can be retried later if necessary.
kPmbSdkReadTimeout (600004)	the specified timeout period was reached, the msg has been written, but no response from the peer has been received. and the send can be retried later if necessary.
kPmbSdkUnavailable (600005)	the sdk instance is not available, typically, it's due to a connection issue to the msg bus.
kPmbSdkQueueIsFull (600006)	the write queue has reached its limit (100), typically, it's due to excessive write operations within a short period or network issues. and the send can be retried later if necessary.
kPmbSdkConnectionLost (600007)	the connection to the msg bus is lost.
kPmbSdkDestroyed (600008)	the sdk has been destroyed.

5. View in Portal

The term Endpoint here specifically refers to a game backend server process that holds a PgosMsgBusSDK instance, and it is hosted, started, and shutdown by the game. Log in to the web portal, enter the console, and go to Extension > Message Bus, you can view the statistics and logs of the endpoints.

5.1 Endpoint List

Click the Endpoints tab, you will get a list of active endpoints:

The endpoint list field description:

Instance ID: the instance ID of the endpoint.
Endpoint IP: the public IP address of the endpoint.
Total Messages: total ingress / egress messages of the endpoint
Last Minute Messages: last minute ingress / egress messages of the endpoint.
Connected Time: indicates when the endpoint establishes a message bus channel with the PGOS backend.

5.2 Message Bus Channel

Message Bus Channel is a virtual connection between the endpoint and PGOS backend. Click the Instance ID of the endpoint, you will get the message bus channel details of the endpoint connected:

It is mainly divided into 3 parts: Basic Information, Push Message Statistics, and RPC Message Statistics.

5.2.1 Basic Information

It lists the instance ID, status, public IP address, first connection time, and last connection time (if there was a reconnection) corresponding to this Message Bus Channel's endpoint.

5.2.2 Push Message Statistics

The list field description:

Source: specifies who sent the push message.
Total Sent Count: specifies how many push messages were sent.
Total Relay Count: specifies how many push messages were successfully relayed.
Last Minute Sent Count: specifies how many push messages were sent in the last minute.
Last Minute Relay Count: specifies how many push messages were successfully relayed in the last minute.

5.2.3 RPC Message Statistics

The list field description:

Source: specifies who sent the push message.
Total Sent Count: specifies how many RPC messages were sent.
Total Relay Count: specifies how many RPC messages were successfully relayed.
Total Response Count: specifies how many RPC messages successfully got a response.
Last Minute Sent Count: specifies how many RPC messages were sent in the last minute.
Last Minute Relay Count: specifies how many RPC messages were successfully relayed in the last minute.
Last Minute Response Count: specifies how many RPC messages successfully got a response in the last minute.

5.3 Endpoint Logs

Click the Logs tab, you will get a brief log list of the behavior for all endpoints:

1. Overview​

2. Architecture​

3. PgosMsgBusSDK Integration​

3.1 Integrating SDK with CMake​

4. PgosMsgBusSDK Usage​

4.1 Get the SDK Instance​

4.2 Initialize the SDK Instance​

4.3 Call the APIs You Need​

4.4 Shutdown the SDK instance​

4.5 Key Error Handling​

5. View in Portal​

5.1 Endpoint List​

5.2 Message Bus Channel​

5.2.1 Basic Information​

5.2.2 Push Message Statistics​

5.2.3 RPC Message Statistics​

5.3 Endpoint Logs​