mirror of
https://github.com/esx-framework/esx_core.git
synced 2026-08-29 01:08:54 +00:00
feat(esx_lib): add a pub/sub multicast module
Adds xLib.pubsub for server-driven topic multicast: a producer resource subscribes players to a topic server-side and publishes data that is pushed only to the subscribed players. Clients listen with xLib.pubsub.on and never subscribe or publish themselves. The subscription registry is a single shared instance in the esx_lib server VM, reached cross-resource through exports; the client side is a per-VM listener over one net event. Subscriptions are cleaned up on playerDropped, so a reused serverId never inherits them, and on the owning resource stopping. Also adds xLib.triggerClientEvent, which packs the payload once when sending to many players; publish uses it so a broadcast to N subscribers serialises once.
This commit is contained in:
@@ -0,0 +1,79 @@
|
||||
-- Per-VM client listener for xLib pub/sub. Routes incoming topic data to the
|
||||
-- handlers registered with xLib.pubsub.on. on() only listens; the player must be
|
||||
-- subscribed server-side to actually receive anything.
|
||||
|
||||
local PUBSUB_EVENT <const> = '__xLib_pubsub' -- must match resource/pubsub/server.lua
|
||||
|
||||
-- topic -> array of handlers
|
||||
local handlers = {}
|
||||
|
||||
local pubsub = {}
|
||||
|
||||
---Register a handler for a topic. Multiple handlers per topic are allowed.
|
||||
---@param topic string
|
||||
---@param handler fun(data: any, topic: string)
|
||||
function pubsub.on(topic, handler)
|
||||
if type(topic) ~= 'string' or type(handler) ~= 'function' then
|
||||
return
|
||||
end
|
||||
|
||||
local list = handlers[topic]
|
||||
if not list then
|
||||
list = {}
|
||||
handlers[topic] = list
|
||||
end
|
||||
list[#list + 1] = handler
|
||||
end
|
||||
|
||||
---Remove handlers for a topic. Without a handler argument, removes all of them.
|
||||
---@param topic string
|
||||
---@param handler? function
|
||||
function pubsub.off(topic, handler)
|
||||
if not handler then
|
||||
handlers[topic] = nil
|
||||
return
|
||||
end
|
||||
|
||||
local list = handlers[topic]
|
||||
if not list then
|
||||
return
|
||||
end
|
||||
|
||||
for i = #list, 1, -1 do
|
||||
if list[i] == handler then
|
||||
table.remove(list, i)
|
||||
end
|
||||
end
|
||||
|
||||
if list[1] == nil then
|
||||
handlers[topic] = nil
|
||||
end
|
||||
end
|
||||
|
||||
RegisterNetEvent(PUBSUB_EVENT, function(topic, data)
|
||||
local list = handlers[topic]
|
||||
if not list then
|
||||
return
|
||||
end
|
||||
|
||||
-- fast path: a topic almost always has a single handler, so skip the copy
|
||||
local n = #list
|
||||
if n == 1 then
|
||||
list[1](data, topic)
|
||||
return
|
||||
end
|
||||
|
||||
-- snapshot so a handler that calls off() mid-dispatch cannot shift the list
|
||||
-- and skip a sibling that has not run yet
|
||||
local snapshot = {}
|
||||
for i = 1, n do
|
||||
snapshot[i] = list[i]
|
||||
end
|
||||
|
||||
for i = 1, n do
|
||||
snapshot[i](data, topic)
|
||||
end
|
||||
end)
|
||||
|
||||
xLib.pubsub = pubsub
|
||||
return pubsub
|
||||
@@ -0,0 +1,47 @@
|
||||
-- Per-VM wrapper exposing xLib.pubsub.* on the server; forwards to the shared
|
||||
-- registry (resource/pubsub/server.lua) through its exports.
|
||||
|
||||
local pubsub = {}
|
||||
|
||||
---@param src number serverId of the player to add
|
||||
---@param topic string
|
||||
---@return boolean added
|
||||
function pubsub.subscribe(src, topic)
|
||||
return xLib.pubsub_subscribe(src, topic)
|
||||
end
|
||||
|
||||
---@param src number
|
||||
---@param topic string
|
||||
---@return boolean removed
|
||||
function pubsub.unsubscribe(src, topic)
|
||||
return xLib.pubsub_unsubscribe(src, topic)
|
||||
end
|
||||
|
||||
---@param topic string
|
||||
---@param data any server-trusted data only
|
||||
---@return integer count number of players notified
|
||||
function pubsub.publish(topic, data)
|
||||
return xLib.pubsub_publish(topic, data)
|
||||
end
|
||||
|
||||
---@param topic string
|
||||
---@return number[] serverIds
|
||||
function pubsub.subscribers(topic)
|
||||
return xLib.pubsub_subscribers(topic)
|
||||
end
|
||||
|
||||
---@param src number
|
||||
---@param topic string
|
||||
---@return boolean
|
||||
function pubsub.isSubscribed(src, topic)
|
||||
return xLib.pubsub_isSubscribed(src, topic)
|
||||
end
|
||||
|
||||
---@param topic string
|
||||
---@return integer count
|
||||
function pubsub.clear(topic)
|
||||
return xLib.pubsub_clear(topic)
|
||||
end
|
||||
|
||||
xLib.pubsub = pubsub
|
||||
return pubsub
|
||||
@@ -0,0 +1,207 @@
|
||||
-- xLib pub/sub: a producer subscribes players to a topic server-side and
|
||||
-- publishes data pushed only to them. Clients only listen, never subscribe.
|
||||
-- The registry is shared here (esx_lib server VM). Namespace topics by resource
|
||||
-- (e.g. "esx_shops:247:stock") so two resources cannot clash on a bare name.
|
||||
|
||||
---@diagnostic disable: duplicate-set-field
|
||||
|
||||
local PUBSUB_EVENT <const> = '__xLib_pubsub' -- must match imports/pubsub/client.lua
|
||||
local TOPIC_PATTERN <const> = '^[%w_:%.%-]+$'
|
||||
local MAX_TOPIC_LEN <const> = 200
|
||||
local resourceName <const> = GetCurrentResourceName()
|
||||
|
||||
-- topic -> set of subscribed serverIds: subs[topic][src] = true
|
||||
local subs = {}
|
||||
-- serverId -> set of topics: memberOf[src][topic] = true (O(degree) cleanup on drop)
|
||||
local memberOf = {}
|
||||
-- topic -> resource that created it, purged on that resource's onResourceStop
|
||||
local ownerOf = {}
|
||||
|
||||
local function isValidTopic(topic)
|
||||
return type(topic) == 'string'
|
||||
and #topic > 0
|
||||
and #topic <= MAX_TOPIC_LEN
|
||||
and topic:match(TOPIC_PATTERN) ~= nil
|
||||
end
|
||||
|
||||
local function isLivePlayer(src)
|
||||
-- a connected player has a name; rejects 0, stale or recycled serverIds
|
||||
return type(src) == 'number' and src > 0 and GetPlayerName(src) ~= nil
|
||||
end
|
||||
|
||||
local function detach(src, topic)
|
||||
local set = subs[topic]
|
||||
if set then
|
||||
set[src] = nil
|
||||
if next(set) == nil then
|
||||
subs[topic] = nil
|
||||
ownerOf[topic] = nil
|
||||
end
|
||||
end
|
||||
|
||||
local topics = memberOf[src]
|
||||
if topics then
|
||||
topics[topic] = nil
|
||||
if next(topics) == nil then
|
||||
memberOf[src] = nil
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
---Subscribe a player to a topic. Called server-side by the producer resource.
|
||||
---@param src number serverId of the player to add
|
||||
---@param topic string
|
||||
---@return boolean added false if already subscribed or input is invalid
|
||||
function xLib.pubsub_subscribe(src, topic)
|
||||
if not isLivePlayer(src) or not isValidTopic(topic) then
|
||||
return false
|
||||
end
|
||||
|
||||
local set = subs[topic]
|
||||
if set and set[src] then
|
||||
return false -- idempotent: already subscribed
|
||||
end
|
||||
|
||||
if not set then
|
||||
set = {}
|
||||
subs[topic] = set
|
||||
ownerOf[topic] = GetInvokingResource() or resourceName
|
||||
end
|
||||
set[src] = true
|
||||
|
||||
local topics = memberOf[src]
|
||||
if not topics then
|
||||
topics = {}
|
||||
memberOf[src] = topics
|
||||
end
|
||||
topics[topic] = true
|
||||
|
||||
return true
|
||||
end
|
||||
|
||||
---Unsubscribe a player from a topic.
|
||||
---@param src number
|
||||
---@param topic string
|
||||
---@return boolean removed false if it was not subscribed
|
||||
function xLib.pubsub_unsubscribe(src, topic)
|
||||
local set = subs[topic]
|
||||
if not set or not set[src] then
|
||||
return false -- idempotent
|
||||
end
|
||||
detach(src, topic)
|
||||
return true
|
||||
end
|
||||
|
||||
---Push data to every player subscribed to a topic. Server-trusted data only.
|
||||
---@param topic string
|
||||
---@param data any
|
||||
---@return integer count number of players notified
|
||||
function xLib.pubsub_publish(topic, data)
|
||||
-- no isValidTopic() here on purpose: an invalid topic never created a set,
|
||||
-- so the lookup below already returns 0 for it without a regex on the hot path
|
||||
local set = subs[topic]
|
||||
if not set then
|
||||
return 0
|
||||
end
|
||||
|
||||
local targets, count = {}, 0
|
||||
for src in pairs(set) do
|
||||
count = count + 1
|
||||
targets[count] = src
|
||||
end
|
||||
|
||||
-- packs the payload once for the whole set instead of once per client
|
||||
xLib.triggerClientEvent(PUBSUB_EVENT, targets, topic, data)
|
||||
return count
|
||||
end
|
||||
|
||||
---@param topic string
|
||||
---@return number[] serverIds a fresh copy, empty if the topic has no subscribers
|
||||
function xLib.pubsub_subscribers(topic)
|
||||
local set = subs[topic]
|
||||
if not set then
|
||||
return {}
|
||||
end
|
||||
|
||||
local list, i = {}, 0
|
||||
for src in pairs(set) do
|
||||
i = i + 1
|
||||
list[i] = src
|
||||
end
|
||||
return list
|
||||
end
|
||||
|
||||
---@param src number
|
||||
---@param topic string
|
||||
---@return boolean
|
||||
function xLib.pubsub_isSubscribed(src, topic)
|
||||
local set = subs[topic]
|
||||
return set ~= nil and set[src] == true
|
||||
end
|
||||
|
||||
---Drop a whole topic and all of its subscribers.
|
||||
---@param topic string
|
||||
---@return integer count number of subscribers that were removed
|
||||
function xLib.pubsub_clear(topic)
|
||||
local set = subs[topic]
|
||||
if not set then
|
||||
return 0
|
||||
end
|
||||
|
||||
local count = 0
|
||||
for src in pairs(set) do
|
||||
local topics = memberOf[src]
|
||||
if topics then
|
||||
topics[topic] = nil
|
||||
if next(topics) == nil then
|
||||
memberOf[src] = nil
|
||||
end
|
||||
end
|
||||
count = count + 1
|
||||
end
|
||||
|
||||
subs[topic] = nil
|
||||
ownerOf[topic] = nil
|
||||
return count
|
||||
end
|
||||
|
||||
-- Remove a leaving player from every topic. serverIds are reused by the server,
|
||||
-- so skipping this would leak a topic's data to whoever inherits the slot.
|
||||
AddEventHandler('playerDropped', function()
|
||||
local src = source
|
||||
local topics = memberOf[src]
|
||||
if not topics then
|
||||
return
|
||||
end
|
||||
|
||||
for topic in pairs(topics) do
|
||||
local set = subs[topic]
|
||||
if set then
|
||||
set[src] = nil
|
||||
if next(set) == nil then
|
||||
subs[topic] = nil
|
||||
ownerOf[topic] = nil
|
||||
end
|
||||
end
|
||||
end
|
||||
memberOf[src] = nil
|
||||
end)
|
||||
|
||||
-- Purge the topics a stopped producer owned, so a producer restart never leaves
|
||||
-- orphan subscriptions that keep receiving pushes.
|
||||
AddEventHandler('onResourceStop', function(stopped)
|
||||
if stopped == resourceName then
|
||||
return
|
||||
end
|
||||
|
||||
local orphaned = {}
|
||||
for topic, owner in pairs(ownerOf) do
|
||||
if owner == stopped then
|
||||
orphaned[#orphaned + 1] = topic
|
||||
end
|
||||
end
|
||||
|
||||
for i = 1, #orphaned do
|
||||
xLib.pubsub_clear(orphaned[i])
|
||||
end
|
||||
end)
|
||||
@@ -0,0 +1,21 @@
|
||||
-- Triggers an event for one or more clients. For an array of players the payload
|
||||
-- is packed once instead of being re-serialised per client.
|
||||
|
||||
---@diagnostic disable: duplicate-set-field
|
||||
|
||||
local pack = msgpack.pack_args
|
||||
|
||||
---@param eventName string
|
||||
---@param targets number | number[] a single serverId, or an array of them
|
||||
---@param ... any
|
||||
function xLib.triggerClientEvent(eventName, targets, ...)
|
||||
if type(targets) == 'number' then
|
||||
return TriggerClientEvent(eventName, targets, ...)
|
||||
end
|
||||
|
||||
local payload = pack(...)
|
||||
local length = #payload
|
||||
for i = 1, #targets do
|
||||
TriggerClientEventInternal(eventName, targets[i], payload, length)
|
||||
end
|
||||
end
|
||||
Reference in New Issue
Block a user