Background
Luo9's plugin system is built on an FFI message bus. Each plugin is an independent dynamic library, such as a DLL or SO, running in its own OS thread. Communication happens inside the same process through extern "C" functions exposed by luo9_core, using a pub/sub model.
The original dispatcher used a pure broadcast model: once the host called publish, every subscriber received the message at roughly the same time. That model is simple and fast, but it has an obvious limitation: there is no way to define delivery order, and no plugin can intercept a message to prevent later plugins from processing it.
The new implementation adds plugin priority and message blocking to Luo9. The important part is not only what it adds, but how it is added: existing plugin code does not need to change at all.
From Broadcast to Targeted Delivery
The previous flow looked like this:
宿主 dispatch_message()
│
▼
Bus::topic("luo9_message").publish(json)
│
├──► 插件A 的队列 (fan-out 副本)
├──► 插件B 的队列 (fan-out 副本)
└──► 插件C 的队列 (fan-out 副本)
Every plugin received the same message. There was no ordering, and there was no point at which delivery could be stopped.
The new flow is different:
宿主 priority_dispatch_message()
│
▼
按优先级降序遍历
│
├──► Bus::publish_to("luo9_message", json, [插件A的sub_id])
│ (插件A 是高优先级，block_enabled=true)
│
│ → 插件A 收到消息后，停止分发
│
✗ 插件B 和 插件C 不会收到这条消息
The key change is replacing publish, which broadcasts to everyone, with publish_to, which pushes a message only to selected subscriber IDs.
Bus Layer: publish_to, unsubscribe, and Dead Subscribers
Two functions were added to the Bus implementation in luo9_core.
publish_to performs targeted publishing. Unlike publish, it only pushes the payload into the queues of the specified subscriber IDs. During delivery it skips subscribers already marked as dead, and after the push it calls notify_all so any blocked consumer threads can wake up.
unsubscribe removes a subscriber from normal operation. Internally, it marks the subscriber as dead, removes the subscriber queue, and wakes any thread blocked on wait_pop. The awakened thread can then detect the dead state and receive a sentinel message, allowing the plugin to exit cleanly.
The Bus keeps a per-topic dead set. publish, publish_to, pop, and wait_pop all check this set before doing their work, which guarantees that an unsubscribed subscriber will not receive more messages.
Host Layer: Pre-Creating Subscribers
In the old model, each plugin called Bus::topic("luo9_message").subscribe() by itself when it started.
The new model moves subscriber creation to the host. When a plugin is loaded, the host creates its subscribers in advance:
fn create_subscribers(plugin_name: &str) -> HashMap<String, usize> {
let topics = ["luo9_message", "luo9_notice", "luo9_meta_event", "luo9_task", "luo9_send"];
let mut ids = HashMap::new();
for topic in &topics {
let id = Bus::topic(topic).subscribe().unwrap();
ids.insert(topic.to_string(), id);
}
ids
}
Those subscriber IDs are then passed into the plugin SDK through FFI:
#[repr(C)]
pub struct PluginSubscribers {
pub message_sub_id: i32,
pub meta_event_sub_id: i32,
pub notice_sub_id: i32,
pub task_sub_id: i32,
pub send_sub_id: i32,
}
When the host spawns the plugin thread, it first calls luo9_init_subscribers to hand over the mapping, then calls the plugin entry point:
fn run_plugin(lib: Arc<Library>, plugin_name: &str, subscriber_ids: HashMap<String, usize>) {
// 1. 传递预分配的 subscriber ID
if let Ok(init_fn) = lib.get::<InitSubscribersFn>(b"luo9_init_subscribers\0") {
let subs = PluginSubscribersRaw {
message_sub_id: subscriber_ids.get("luo9_message").copied().unwrap_or(0) as i32,
// ... 其他 topic
};
init_fn(&subs);
}
// 2. 调用插件入口
let plugin_main: Symbol<unsafe extern "C" fn()> = lib.get(b"plugin_main\0").unwrap();
plugin_main();
}
This gives the host full knowledge of which plugin owns which subscriber ID, which is exactly what priority-based dispatch needs.
SDK Layer: Swapping the Subscriber Transparently
The plugin still calls Topic::subscribe() as before. The difference is hidden inside the SDK: it first checks whether the host has already provided a subscriber ID for the topic.
pub static PRECREATED_SUBSCRIBERS: OnceLock<Mutex<HashMap<String, usize>>> = OnceLock::new();
impl<'a> Topic<'a> {
pub fn subscribe(&self) -> Result<usize, BusError> {
// 检查是否有预分配的 subscriber_id
if let Some(map) = PRECREATED_SUBSCRIBERS.get() {
if let Some(&id) = map.lock().unwrap().get(self.name) {
return Ok(id); // 直接返回预分配的 ID
}
}
// 否则正常创建 subscriber
let ret = unsafe { luo9_bus_subscribe(topic.as_ptr()) };
// ...
}
}
This is the reason existing plugins need no modification. A plugin still writes the same subscription code, but the SDK returns the host-created ID instead of creating a fresh subscriber. From the plugin's perspective, nothing has changed.
Priority Dispatch in the Host
The host keeps a dispatch list sorted by priority. It is stored in an RwLock, so the dispatcher can read the list quickly while updates remain controlled.
static DISPATCH_LIST: RwLock<Vec<DispatchEntry>> = RwLock::new(Vec::new());
pub struct DispatchEntry {
pub name: String,
pub priority: i32,
pub block_enabled: bool,
pub message_sub_id: Option<usize>,
pub notice_sub_id: Option<usize>,
pub meta_event_sub_id: Option<usize>,
}
One detail matters here: message_sub_id is Option<usize>, not usize. That choice exists because subscriber IDs start from 0.
The dispatch logic walks through the sorted entries, sends the message to one plugin at a time, and stops if the current plugin has blocking enabled:
pub fn priority_dispatch_message(msg: Message) {
let payload = serde_json::to_string(&PluginData::Message(msg)).unwrap();
let list = DISPATCH_LIST.read().unwrap();
for entry in list.iter() {
// 跳过未订阅该 topic 的插件
let Some(sub_id) = entry.message_sub_id else {
continue;
};
// 定向推送到这个插件的 subscriber 队列
Bus::topic("luo9_message")
.publish_to(&payload, &[sub_id])
.unwrap();
// 如果这个插件启用了阻断，停止分发
if entry.block_enabled {
break;
}
}
}
A high-priority plugin can therefore receive a message first. If its block_enabled flag is set, lower-priority plugins never see that message.
The Subscriber ID 0 Trap
A subtle bug appeared during implementation: the first loaded plugin received subscriber ID 0, and the dispatcher treated it as if it had not subscribed.
The original DispatchEntry used a plain usize:
pub struct DispatchEntry {
pub message_sub_id: usize, // 0 表示"未订阅"
}
The dispatcher skipped entries with ID 0:
if entry.message_sub_id == 0 {
continue; // 跳过未订阅的插件
}
That would be harmless in systems where IDs start at 1. But in luo9_core, subscriber IDs are per-topic and start from 0. The first loaded plugin, plugin_doro, got subscriber_id = 0 on luo9_message, so it was incorrectly skipped.
The fix was to use Option<usize>: None means there is no subscription, while Some(0) is a valid subscriber ID.
// 之前：message_sub_id: h.subscriber_ids.get("luo9_message").copied().unwrap_or(0),
// 之后：message_sub_id: h.subscriber_ids.get("luo9_message").copied(),
The dispatch code then uses pattern matching rather than numeric comparison:
// 之前：if entry.message_sub_id == 0 { continue; }
// 之后：
let Some(sub_id) = entry.message_sub_id else { continue };
The lesson is straightforward: do not use a magic value to represent absence. Use Option.
Hot Reloading with a Sentinel Message
Disabling a plugin requires its thread to exit. The difficulty is that plugin_main is usually blocked in wait_pop, waiting on a condition variable. It needs to be awakened without forcing plugin authors to write new shutdown logic.
The solution is a sentinel. luo9_core defines a special string, __luo9_unsubscribed__. When a subscriber is unsubscribed, wait_pop returns that sentinel instead of a normal message.
The disable flow starts by unsubscribing all subscriber IDs held by the plugin handle:
// PluginHandle
pub fn unsubscribe_all(&self) {
for (topic, &sub_id) in &self.subscriber_ids {
Bus::topic(topic).unsubscribe(sub_id);
}
}
unsubscribe does three things:
- Adds the subscriber ID to the dead set.
- Removes the subscriber and drops its queue.
- Calls
notify_allto wake any thread blocked inwait_pop.
Once awakened, wait_pop sees that the subscriber has been removed and is present in the dead set, so it returns the sentinel. The SDK recognizes the sentinel and returns Err(BusError::Unsubscribed). The plugin's normal loop can then exit naturally:
// 插件代码（无需修改）
loop {
let msg = match Bus::topic("luo9_message").wait_pop(sub_id) {
Ok(msg) => msg,
Err(BusError::Unsubscribed) => break, // 收到取消订阅信号，退出
Err(e) => {
error!("错误: {:?}", e);
continue;
}
};
// 处理消息...
}
After the plugin thread exits, the reference count of Arc<Library> drops to zero, dlclose runs automatically, and the dynamic library is unloaded. Hot reload can then load the DLL again, create fresh subscribers, and spawn a new thread.
Persisting Priority and Blocking Settings
Plugin priority and blocking configuration are persisted in the main configuration file:
[[plugins.plugins]]
name = "plugin_doro"
priority = 3
block_enabled = true
[[plugins.plugins]]
name = "plugin_epic"
priority = 1
block_enabled = false
During startup, init_global_manager applies priority and block_enabled from the configuration to each plugin's PluginHandle and PluginInfo:
for mut handle in handles {
if let Some(entry) = config_entries.iter().find(|e| e.name == handle.name) {
handle.priority = entry.priority;
handle.block_enabled = entry.block_enabled;
}
manager.register_handle(handle);
}
When priority or blocking is changed from the WebUI, the same values are written back to the configuration file, so the settings survive a restart.
Overall Shape
┌─────────────────────────────────────┐
│ 宿主 (luo9_bot) │
│ │
│ PluginManager │
│ ├── PluginHandle A (priority=10) │
│ │ ├── subscriber_ids: { │
│ │ │ message: 0, notice: 0 │
│ │ │ } │
│ │ ├── block_enabled: true │
│ │ └── thread_handle: JoinHandle│
│ ├── PluginHandle B (priority=5) │
│ │ └── subscriber_ids: { │
│ │ message: 1, notice: 1 │
│ │ } │
│ └── PluginHandle C (priority=0) │
│ └── subscriber_ids: { │
│ message: 2, notice: 2 │
│ } │
│ │
│ DISPATCH_LIST: RwLock<Vec<...>> │
│ (按 priority 降序排列) │
│ │
│ priority_dispatch_message() │
│ │ │
│ ├─► publish_to(msg, [0]) ← A │
│ │ A.block_enabled → break│
│ ├─► publish_to(msg, [1]) ← B │
│ └─► publish_to(msg, [2]) ← C │
└──────────────┬──────────────────────┘
│
┌──────────────▼──────────────────────┐
│ luo9_core (FFI 消息总线) │
│ │
│ 每个 topic 维护独立的 subscriber │
│ 队列，per-topic 编号从 0 开始。 │
│ │
│ publish_to: 定向推送到指定队列 │
│ unsubscribe: 标记 dead + 唤醒线程 │
│ wait_pop: dead 时返回哨兵消息 │
└─────────────────────────────────────┘
Subscriber IDs are scoped per topic. Plugin A can be subscriber 0 on luo9_message and also subscriber 0 on luo9_notice; those are separate subscriber queues under separate topics.
What Changed Where
<table> <thead> <tr> <th>Layer</th> <th>Change</th> <th>Visible to plugins</th> </tr> </thead> <tbody> <tr> <td>Core Bus</td> <td>Addedpublish_to, unsubscribe, dead set, sentinel</td>
<td>No</td>
</tr>
<tr>
<td>SDK</td>
<td>subscribe() returns pre-created IDs; wait_pop recognizes the sentinel</td>
<td>No</td>
</tr>
<tr>
<td>Host</td>
<td>Priority dispatch, hot reload, handle management, config persistence</td>
<td>No</td>
</tr>
<tr>
<td>Plugin</td>
<td>No changes</td>
<td>Fully transparent</td>
</tr>
</tbody>
</table>
The design keeps the complexity in the infrastructure layer. Plugin authors continue to focus on business logic, while priority, blocking, hot reload, subscriber cleanup, and configuration persistence are handled by the host and SDK.