forked from Kweebec_Network/core-vitals
140 lines
4.6 KiB
Markdown
140 lines
4.6 KiB
Markdown
# MessageBus
|
|
|
|
A thin wrapper over Redis pub/sub. Use it for fire-and-forget cross-server events (party invites, chat broadcasts, "player joined queue X"). For request/response, see [07_RPC.md](07_RPC.md).
|
|
|
|
## API
|
|
|
|
```java
|
|
public interface MessageBus {
|
|
void publish(String channel, Object payload);
|
|
<T> Subscription subscribe(String channel, Class<T> type, Consumer<T> handler);
|
|
|
|
interface Subscription extends AutoCloseable {
|
|
@Override void close();
|
|
}
|
|
}
|
|
```
|
|
|
|
## Publishing
|
|
|
|
```java
|
|
PartyInvite invite = new PartyInvite(fromUuid, toUuid);
|
|
core.getMessageBus().publish("party.invite", invite);
|
|
```
|
|
|
|
`payload` can be any class Gson can serialize. The bus wraps it in an envelope before sending.
|
|
|
|
The call is non-blocking - it hands off to Lettuce's async API and returns immediately. There is no acknowledgement that another server received it. Pub/sub is at-most-once delivery: if no server is subscribed when you publish, the message disappears.
|
|
|
|
## Subscribing
|
|
|
|
```java
|
|
MessageBus.Subscription sub = core.getMessageBus().subscribe(
|
|
"party.invite",
|
|
PartyInvite.class,
|
|
invite -> handleInvite(invite)
|
|
);
|
|
```
|
|
|
|
Save the subscription if you want to unsubscribe later (`sub.close()`). NetworkCore's own subscriptions are cleaned up on shutdown, but yours aren't tied to anything by default.
|
|
|
|
### Thread safety - read this
|
|
|
|
Callbacks run on an internal worker pool (`networkcore-bus-*` threads), **not** on a Hytale world thread. If your handler touches world or entity state, hop to the right thread first:
|
|
|
|
```java
|
|
core.getMessageBus().subscribe("party.invite", PartyInvite.class, invite -> {
|
|
World world = playerRef.getWorld();
|
|
world.execute(() -> {
|
|
// Now safe to touch entities, components, etc.
|
|
playerRef.sendMessage(Message.raw("Party invite from " + invite.from));
|
|
});
|
|
});
|
|
```
|
|
|
|
Failing to hop will mostly work most of the time and then crash unpredictably under load. Always hop.
|
|
|
|
The worker pool is sized 2 core / 8 max, queue 1024. A slow handler will not block other subscribers - it'll just back up its own queue. If the queue fills, the publisher thread runs the task itself (CallerRunsPolicy), which slows down message intake until the backlog clears.
|
|
|
|
## The envelope
|
|
|
|
What actually goes over Redis:
|
|
|
|
```json
|
|
{
|
|
"v": 1,
|
|
"from": "lobby-3f2a91bd",
|
|
"auth": "<token from config>",
|
|
"ts": 1716610000000,
|
|
"type": "com.example.PartyInvite",
|
|
"body": { ... your payload ... }
|
|
}
|
|
```
|
|
|
|
You don't construct this. The bus does it. `type` is informational only - dispatch is by channel, not by class.
|
|
|
|
## Channel naming
|
|
|
|
A loose convention to keep things grep-able:
|
|
|
|
- `party.invite`, `party.disband` - simple cross-server events
|
|
- `chat.global` - broadcast channels
|
|
- `rpc:req:<service>`, `rpc:resp:<replyTo>` - reserved by RPC, don't touch
|
|
- `network:events` - reserved by NetworkCore for hello/goodbye, don't touch
|
|
|
|
## Auth filtering
|
|
|
|
If `network.auth_token` is set (see [03_Configuration.md](03_Configuration.md)), inbound messages whose envelope `auth` doesn't match are dropped silently (with a rate-limited warning log and an increment of `messagebus_rejected_total`). Outbound messages get stamped automatically.
|
|
|
|
If a misconfigured server keeps spamming the bus with the wrong token, you'll see one log line every 10s rather than one per message.
|
|
|
|
## Performance and limits
|
|
|
|
- Lettuce uses one connection for publish, one for subscribe. They share threading via Netty.
|
|
- Pub/sub is in-memory in Redis. No persistence. If Redis restarts, in-flight messages are lost.
|
|
- Redis pub/sub bandwidth is real but small - the practical bottleneck is usually subscriber-side processing, not network.
|
|
- Payloads are JSON. Keep them small. Don't put 10MB worlds through the bus - put them in MySQL or object storage and bus a reference.
|
|
|
|
## Common patterns
|
|
|
|
### Broadcast to everyone
|
|
|
|
Just publish - every server subscribed to the channel gets it.
|
|
|
|
```java
|
|
core.getMessageBus().publish("chat.global", chatMessage);
|
|
```
|
|
|
|
### Target a specific server
|
|
|
|
Encode the target in the payload, let receivers filter:
|
|
|
|
```java
|
|
class TargetedAction {
|
|
String targetServerId;
|
|
String action;
|
|
Map<String, Object> data;
|
|
}
|
|
|
|
bus.subscribe("admin.action", TargetedAction.class, action -> {
|
|
if (!action.targetServerId.equals(core.getServerId())) return;
|
|
apply(action);
|
|
});
|
|
```
|
|
|
|
Or use a per-server channel: `admin.action.<server-id>`.
|
|
|
|
### Avoid receiving your own messages
|
|
|
|
Filter on `serverId`:
|
|
|
|
```java
|
|
String myId = core.getServerId();
|
|
bus.subscribe("chat.global", ChatMessage.class, msg -> {
|
|
if (myId.equals(msg.fromServerId)) return;
|
|
relay(msg);
|
|
});
|
|
```
|
|
|
|
The bus does NOT filter your own publishes - all subscribers (including the publisher) see the message.
|