PubSubLib 2.0.3
dotnet add package PubSubLib --version 2.0.3
NuGet\Install-Package PubSubLib -Version 2.0.3
<PackageReference Include="PubSubLib" Version="2.0.3" />
<PackageVersion Include="PubSubLib" Version="2.0.3" />
<PackageReference Include="PubSubLib" />
paket add PubSubLib --version 2.0.3
#r "nuget: PubSubLib, 2.0.3"
#:package PubSubLib@2.0.3
#addin nuget:?package=PubSubLib&version=2.0.3
#tool nuget:?package=PubSubLib&version=2.0.3
Server — PubSubLib
Package chứa toàn bộ logic không gian (spatial grid), quản lý Unit/Watcher, và cầu nối NATS (Natify). Đây là thành phần authoritative — sở hữu dữ liệu chính thống về vị trí, trạng thái của mọi unit.
Mục lục
Cài đặt
Package multi-target netstandard2.1 (Unity IL2CPP) và net9.0 (standalone server):
<PackageReference Include="PubSubLib" Version="2.0.1" />
Hoặc project reference:
<ProjectReference Include="..\PubSubLib\PubSubLib.csproj" />
Dependencies:
System.Threading.Channels— hàng đợi worker thread (bundled cho netstandard2.1)Natify(v1.2.0) — optional, chỉ cần cho chế độ networkedGoogle.Protobuf(v3.34.1) — serialize binary message cho networkedGcoder.Collections(v1.3.0) — ITimedCollection cho player expiration
Khái niệm
| Khái niệm | Định nghĩa |
|---|---|
| Unit | Thực thể game có Id, Type, Position (2D), Data (binary), và strong reference (IAlive) tới object của bạn. |
| Watcher | Người quan sát tại một vị trí với Radius. Nhận event cho mọi unit có vị trí nằm trong bán kính. |
| Cell | Lưới không gian được chia thành các ô vuông (kích thước GridSize). Truy vấn không gian O(1). |
| BatchEnter | Thông báo fire-and-forget: unit này vừa vào tầm quan sát của bạn. |
| BatchLeave | Thông báo fire-and-forget: unit này vừa rời tầm quan sát của bạn. |
| SyncEnter | Thông báo bulk: danh sách toàn bộ unit hiện có trong tầm. Dùng cho lần đầu sync và ping reconciliation. |
| SyncLeave | Thông báo bulk: những unit key này không còn trong tầm. |
| UnitEvent | Sự kiện tùy ý do unit phát ra (vd: "attack", "pickup") → forward tới mọi watcher đang quan sát unit đó. |
Cấu hình (PubSubConfig)
public class PubSubConfig
{
public float GridSize = 100f; // kích thước mỗi cell vuông (đơn vị world)
public int WatcherTimeoutSeconds = 5; // watcher hết hạn nếu không ping trong khoảng này
public int WatcherCleanupIntervalSeconds = 2; // tần suất kiểm tra watcher hết hạn (mỗi N giây)
}
- GridSize: Giá trị nhỏ → nhiều cell hơn, ít unit/cell hơn → query nhanh hơn nhưng tốn bộ nhớ hơn. Giá trị lớn → ít cell hơn, nhiều unit/cell hơn. Điều chỉnh theo scale world và mật độ unit.
- WatcherTimeoutSeconds: Nếu
WatcherPingUnitskhông được gọi cho watcher trong khoảng này, watcher tự động bị xóa. Mặc định 5 giây. - WatcherCleanupIntervalSeconds: Worker thread kiểm tra và xóa watcher hết hạn ở tần suất này trong pha idle. Mặc định 2 giây.
API Reference
IPubSub
public interface IPubSub : IDisposable
{
// ── Factory ──
static IPubSub Create(PubSubConfig config);
// ── Unit lifecycle ──
void CreateUnit<T>(long id, string type, Vector2 position, T target,
Action<IUnit> onCreated, byte[]? data = null) where T : class, IAlive;
Task<IUnit> CreateUnitAsync<T>(long id, string type, Vector2 position,
T target, byte[]? data = null) where T : class, IAlive;
Task FlushAsync();
// ── Watcher lifecycle ──
void AddWatcher(long watcherId, Vector2 position, float radius);
void RemoveWatcher(long watcherId);
void MoveWatcher(long watcherId, Vector2 position, float radius);
// ── State reconciliation ──
void WatcherPingUnits(long watcherId,
Dictionary<string, Dictionary<long, int>> typeVersions);
// ── Natify integration ──
void AddNatify(NatifyClientFast client);
void AddNatify(NatifyClient client);
// ── Batch callbacks (enter/leave tức thời) ──
void OnUnitEnter(Action<(List<long> notyWatchIds, IUnit units)> callBack);
void OnUnitLeave(Action<(List<long> notyWatchIds, IUnit units)> callBack);
// ── Sync callbacks (trạng thái ban đầu / reconciliation) ──
void OnUnitEnter(Action<(long notyWatchId, List<IUnit> units)> callBack);
void OnUnitLeave(Action<(long notyWatchId, List<UnitKey> unitKeys)> callBack);
// ── Event callback ──
void OnUnitEvent(Action<(List<long> notyWatchId, IUnit units, string eventName, object data, bool reliable)> callBack);
}
Lưu ý về generic:
CreateUnit<T>vàCreateUnitAsync<T>yêu cầuT : class, IAliveđể type-safety cho tham sốtarget. Các callback và interfaceIUnitkhông còn generic —Targettrả vềobject?.
Các phương thức chi tiết
| Phương thức | Mô tả |
|---|---|
CreateUnit(id, type, pos, target, onCreated, data) |
Tạo unit mới. onCreated callback chạy trên worker thread với IUnit đã được đăng ký trong grid. |
CreateUnitAsync(id, type, pos, target, data) |
Tạo unit async. Trả về Task<IUnit> hoàn thành khi unit đã được đăng ký. |
FlushAsync() |
Enqueue no-op, trả về Task hoàn thành sau khi mọi action trước đó đã được xử lý. Dùng để đồng bộ trong test hoặc cleanup. |
AddWatcher(id, pos, radius) |
Thêm watcher vào grid. Tính toán tất cả cell trong bán kính, đăng ký watcher vào các cell đó, fire SyncEnter với mọi unit hiện có. |
RemoveWatcher(id) |
Xóa watcher. Không fire SyncLeave. |
MoveWatcher(id, pos, radius) |
Di chuyển watcher. Tính diff cell (thêm/xóa), fire SyncEnter/SyncLeave tương ứng. |
WatcherPingUnits(id, typeVersions) |
Ping giữ watcher sống + đồng bộ trạng thái. typeVersions map từ "unitType" → {unitId: version}. Server so sánh version: thiếu/khớp sai → SyncEnter, thừa key → SyncLeave. Watcher lưu _knownTypes — các type đã từng ping. Khi ping rỗng vẫn process các type đã biết (unitVersions rỗng → sync toàn bộ unit trong cell). |
AddNatify(client) |
Gắn Natify để publish event qua NATS và subscribe command từ client. |
Callback Overloads
Hai overload cho OnUnitEnter và OnUnitLeave:
| Overload | Dạng tuple | Dùng cho |
|---|---|---|
OnUnitEnter |
(List<long> watcherIds, IUnit unit) |
Batch: unit vừa vào tầm → thông báo các watcher bị ảnh hưởng |
OnUnitEnter |
(long watcherId, List<IUnit> units) |
Sync: thêm watcher mới hoặc re-sync → danh sách toàn bộ unit trong tầm |
OnUnitLeave |
(List<long> watcherIds, IUnit unit) |
Batch: unit rời tầm → thông báo các watcher bị ảnh hưởng |
OnUnitLeave |
(long watcherId, List<UnitKey> unitKeys) |
Sync: watcher di chuyển hoặc ping phát hiện key cũ → danh sách key cần xóa |
IUnit
public interface IUnit
{
long Id { get; } // ID duy nhất
string Type { get; } // loại unit ("hero", "mob", "item")
Vector2 Position { get; set; } // vị trí 2D — set trigger cập nhật cell + tăng version
bool IsAlive { get; } // false nếu target có IAlive.IsAlive == false
object? Target { get; } // target object (null nếu bị collect)
int Version { get; } // tăng khi Position hoặc Data thay đổi
byte[]? Data { get; set; } // binary payload — set trigger tăng version
void PublishEvent(string eventName, object? data, bool reliable = true); // phát event tới mọi watcher đang quan sát. reliable=true → TCP, reliable=false → UDP
void Destroy(); // xóa unit khỏi grid, fire leave event
}
Các hành vi chính:
- Set
Position: kiểm tra cell thay đổi. Nếu cell đổi → fireBatchLeavecho watcher mất tầm nhìn,BatchEntercho watcher có tầm nhìn mới, version++. Nếu cell không đổi → không làm gì. - Set
Positioncùng giá trị: no-op (không tăng version, không event). Destroy(): xóa unit khỏi grid, fireBatchLeavetới mọi watcher hiện tại đang quan sát.PublishEvent(): fireUnitEventtới mọi watcher có cell chứa unit.IsAlive: delegate quaIAlive.IsAlive— nếu target setIsAlive = false, unit coi là dead và sẽ bị lazy cleanup.
UnitKey
public readonly struct UnitKey : IEquatable<UnitKey>
{
public long Id { get; }
public string Type { get; }
// So sánh bằng cả Id và Type
}
Dùng trong SyncLeave callback và nội bộ WatcherPingUnits để định danh unit theo composite key (Id + Type). Unit khác type có thể dùng chung numeric Id.
IRegionModule
public interface IRegionModule : IAsyncDisposable
{
static IRegionModule Create(RegionConfig config);
Task<T> CreateUnitAsync<T, TR>(long id, Vector2 position, TR target)
where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
void CreateUnit<T, TR>(long id, Vector2 position, TR target, Action<T> callback)
where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
Task<T> CreateUnitAsync<T, TR>(long id, Vector2 position, Func<TR> target)
where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
void CreateUnit<T, TR>(long id, Vector2 position, Func<TR> target, Action<T> callback)
where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
T? GetUnit<T, TR>(long id) where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
bool TryGetUnit<T, TR>(long id, out T unit) where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
IList<T> GetUnits<T, TR>() where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
void DestroyUnit<T, TR>(long id) where T : class, IRegionUnit<TR>, new() where TR : class, IAlive;
void Tick();
}
| Phương thức | Mô tả |
|---|---|
Create(config) |
Static factory — tạo instance từ RegionConfig |
CreateUnitAsync<T, TR>(id, pos, target) |
Tạo unit với generated mirror wrapper. Trả về Task<T> — unit được enqueue vào _unitEventQueue, thêm vào _units sau khi Tick() |
CreateUnit<T, TR>(id, pos, target, callback) |
Version async — callback chạy sau khi unit đã được thêm vào _units |
GetUnit<T, TR>(id) |
Lấy unit theo id. Yêu cầu Tick() đã được gọi — nếu không, trả về null vì unit chưa được thêm vào _units |
TryGetUnit<T, TR>(id, out unit) |
Version bool của GetUnit. Yêu cầu Tick() đã được gọi |
GetUnits<T, TR>() |
Lấy toàn bộ unit theo type. Yêu cầu Tick() đã được gọi |
DestroyUnit<T, TR>(id) |
Xóa unit — enqueue vào _unitEventQueue, thực sự xóa sau khi Tick() |
Tick() |
Bắt buộc gọi định kỳ (mỗi frame hoặc mỗi N ms). Drain _unitEventQueue — xử lý tất cả thao tác thêm/xóa unit vào _units |
Ràng buộc quan trọng:
GetUnit,TryGetUnit,GetUnitsđọc trực tiếp từ_unitsdictionary. NhưngCreateUnitAsyncvàDestroyUnitchỉ enqueue thay đổi — chúng chưa thực sự thêm/xóa_unitscho đến khiTick()được gọi. Do đó, bắt buộc gọiTick()trước khi truy vấn unit.
// Ví dụ: tạo unit rồi query
var unit = await regionModule.CreateUnitAsync<MyUnit, MyTarget>(42, position, target);
regionModule.Tick(); // BẮT BUỘC — enqueue add vào _units
var found = regionModule.GetUnit<MyUnit, MyTarget>(42); // OK — đã có trong _units
// Sai — sẽ trả về null:
var found2 = regionModule.GetUnit<MyUnit, MyTarget>(42); // null nếu chưa Tick()
// Ví dụ: destroy unit rồi query
regionModule.DestroyUnit<MyUnit, MyTarget>(42);
regionModule.Tick(); // BẮT BUỘC — enqueue remove khỏi _units
var found = regionModule.GetUnit<MyUnit, MyTarget>(42); // null — đã bị xóa
IPlayerData
Contract cho dữ liệu player. Mỗi type implement IPlayerData được quản lý riêng biệt bởi IPlayerSpeaksManager, keyed by (DataName, PlayerId).
public interface IPlayerData
{
long PlayerId { get; } // ID của player sở hữu dữ liệu
bool IsOnLine { get; } // true nếu player đang online (được set bởi manager)
string DataName { get; } // Tên duy nhất của loại dữ liệu (vd: "PlayerProfile")
void OnChange(Action<byte[], string> handler); // Đăng ký callback khi dữ liệu thay đổi (data, commit)
void OnMessage(Action<string, byte[]> handler); // Đăng ký callback khi nhận message từ server (subject, data)
void Commit(string commit); // Gửi commit thay đổi dữ liệu đến client
void DoneInit(); // Kích hoạt OnCreate + OnClientConnect (gọi sau khi tạo data)
}
| Member | Mô tả |
|---|---|
PlayerId |
ID của player sở hữu dữ liệu này |
IsOnLine |
true khi player đang kết nối đến router |
DataName |
String key phân biệt các loại dữ liệu khác nhau của cùng 1 player |
OnChange(handler) |
Đăng ký handler nhận (byte[] data, string commit) mỗi khi dữ liệu thay đổi |
OnMessage(handler) |
Đăng ký handler nhận (string subject, byte[] data) khi server gửi message đến player |
Commit(commit) |
Gửi commit message đến client — client nhận qua IPlayerMirrorClient.ApplyUpdate() |
DoneInit() |
Gọi sau khi tạo data. Kiểm tra IsInitDone, fire IOnCreate.OnCreate(), nếu online thì fire IOnClientConnect.OnClientConnect() |
IPlayerDataInternal (internal) extends IPlayerData với các phương thức chỉ dùng bởi manager:
IsInitDone— true nếuDoneInit()đã được gọiSetOnline(bool)— cập nhật trạng thái onlineSetPlayerId(long)— gán playerId (khi tạo lần đầu)PrepareMessageDispatch(subject, data)— chuẩn bị dispatch message đến handlerOnMessage
PlayerSpeakerConfig
public class PlayerSpeakerConfig
{
public int PlayerTimeoutSeconds = 5; // Timeout xóa dữ liệu player khi offline (ITimedCollection)
public int PlayerCleanupIntervalSeconds = 2; // Tần suất kiểm tra timeout (không dùng nữa, giữ để backward compat)
public NatifyClientFast? ClientFast = null; // Natify client (fast path)
public NatifyClient? Client = null; // Natify client (standard path)
}
| Field | Default | Mô tả |
|---|---|---|
PlayerTimeoutSeconds |
5 | Sau khi player offline, dữ liệu tồn tại thêm N giây rồi tự động bị xóa bởi ITimedCollection expiration |
PlayerCleanupIntervalSeconds |
2 | Không dùng nữa (cleanup qua ITimedCollection.Tick()), giữ để backward compat |
ClientFast / Client |
null | Natify client để giao tiếp NATS — cần ít nhất 1 trong 2 để networked mode hoạt động |
IPlayerSpeaksManager
Quản lý dữ liệu cho từng player (đối tượng IPlayerData) gắn với playerId. Dữ liệu tồn tại xuyên suốt phiên — được tạo từ trước khi client connect, và không bị xóa khi client disconnect.
public interface IPlayerSpeaksManager : IAsyncDisposable
{
static IPlayerSpeaksManager Create(PlayerSpeakerConfig config);
// ── Data lifecycle ──
T CreateData<T>(long playerId) where T : class, IPlayerData, new();
T? GetData<T>(long playerId) where T : class, IPlayerData, new();
bool Remove(long playerId);
// ── Main loop ──
void Tick();
// ── Lifecycle hooks ──
void OnDefault<T>(Func<T, Task<Action>> callback) where T : class, IPlayerData, new();
}
| Phương thức | Mô tả |
|---|---|
Create(config) |
Static factory — tạo instance từ PlayerSpeakerConfig |
CreateData<T>(playerId) |
Tạo dữ liệu cho player. Gọi OnDefault<T> nếu là lần đầu tạo. Caller phải gọi data.DoneInit() sau khi setup xong. |
Remove(playerId) |
Xóa toàn bộ dữ liệu của player. Trả về true nếu xóa thành công, false nếu player đang online hoặc không có dữ liệu. |
GetData<T>(playerId) |
Lấy dữ liệu theo type. Trả về null nếu không có. |
Tick() |
Gọi định kỳ (vd: mỗi frame). Xử lý message queue từ NATS, chạy ITimedCollection.Tick() kiểm tra player timeout. |
OnDefault<T>(callback) |
Đăng ký callback chạy khi player lần đầu connect. Func<T, Task<Action>> — Task chạy async, Action chạy trên main thread sau khi hoàn thành. |
Ràng buộc:
Removechỉ thành công khi player không online (IsOnline == false) — bảo vệ dữ liệu đang activeCreateDatakhông gọiOnCreate()— caller phải gọidata.DoneInit()sau khi setup- Player online status được theo dõi qua
_playerMetaData: Dictionary<long, PlayerMetaData>+ITimedCollectionexpiration - Key nội bộ:
Dictionary<long, Dictionary<string, IPlayerData>>— outer key = playerId, inner key = DataName
Kiến trúc nội bộ:
// PlayerSpeaksManager quản lý:
private readonly Dictionary<long, Dictionary<string, IPlayerData>> _data; // dữ liệu player
private readonly Dictionary<string, IDefaultFactory> _defaults; // handler OnDefault<T>
private readonly ITimedCollection<long, PlayerMetaData> _playerTimers; // timeout expiration
private readonly Dictionary<long, PlayerMetaData> _playerMetaData; // LastPingTicks tracking
private readonly ConcurrentQueue<Action> _mainThreadActions; // main thread action queue
Ví dụ:
var config = new PlayerSpeakerConfig
{
ClientFast = natifyClientFast, // Natify client để giao tiếp NATS
PlayerTimeoutSeconds = 5
};
var manager = IPlayerSpeaksManager.Create(config);
// Đăng ký handler cho dữ liệu mặc định
manager.OnDefault<PlayerProfile>(async profile =>
{
profile.Level = 1;
profile.Gold = 100;
return () => { /* Action chạy trên main thread sau async */ };
});
// Tạo dữ liệu
var profile = manager.CreateData<PlayerProfile>(playerId: 42);
profile.DoneInit(); // Bắt buộc gọi sau khi setup xong
// Gọi Tick() định kỳ (mỗi frame hoặc mỗi giây)
manager.Tick();
// Khi muốn xóa dữ liệu (player phải offline)
var removed = manager.Remove(42);
// removed == true → dữ liệu đã bị xóa
// Kiểm tra
var check = manager.GetData<PlayerProfile>(42); // null
PubSubLog
Static logger class tập trung — mọi exception trong worker thread, callback, và network đều được log qua đây. Hỗ trợ try-catch hardening trên toàn bộ hệ thống.
public static class PubSubLog
{
public static Action<string>? Info { get; set; }
public static Action<string>? Warning { get; set; }
public static Action<string>? Error { get; set; }
}
| Member | Mô tả |
|---|---|
Info |
Set action để nhận log info (kết nối, trạng thái) |
Warning |
Set action để nhận warning (timeout, retry) |
Error |
Set action để nhận error (exception trong callback, network lỗi) |
// Setup logging
PubSubLog.Info = msg => Console.WriteLine($"[INFO] {msg}");
PubSubLog.Warning = msg => Console.WriteLine($"[WARN] {msg}");
PubSubLog.Error = msg => Console.WriteLine($"[ERROR] {msg}");
3 chế độ sử dụng
1. Standalone (không Natify)
Toàn bộ logic chạy trong 1 process. Phù hợp cho:
- Local test: kiểm tra luồng event không cần network
- Server monolithic: game server xử lý mọi thứ nội bộ
using PubSubLib;
var pubSub = IPubSub.Create(new PubSubConfig { GridSize = 100f });
// Đăng ký callbacks
pubSub.OnUnitEnter(tuple => {
Console.WriteLine($"[BatchEnter] Watchers {string.Join(",", tuple.Item1)} thấy unit {tuple.Item2.Id}");
});
pubSub.OnUnitLeave(tuple => {
Console.WriteLine($"[SyncLeave] Watcher {tuple.Item1} mất {tuple.Item2.Count} unit");
});
pubSub.OnUnitEvent(tuple => {
Console.WriteLine($"[Event] {tuple.Item2.Id} phát '{tuple.Item3}' reliable={tuple.Item5}");
});
// Thêm watcher
pubSub.AddWatcher(1, new Vector2 { x = 0, y = 0 }, 200f);
// Tạo unit
var myObj = new Player();
var unit = await pubSub.CreateUnitAsync<Player>(42, "hero", new Vector2 { x = 50, y = 50 }, myObj);
// Di chuyển
unit.Position = new Vector2 { x = 150, y = 150 };
// Event
unit.PublishEvent("attack", new { damage = 50 }); // TCP (reliable=true)
unit.PublishEvent("bullet_fx", data, reliable: false); // UDP (best-effort)
// Đồng bộ
pubSub.WatcherPingUnits(1, new Dictionary<string, Dictionary<long, int>>
{
["hero"] = new() { [42] = unit.Version }
});
// Dọn dẹp
unit.Destroy();
await pubSub.FlushAsync();
pubSub.Dispose();
Luồng sự kiện standalone:
CreateUnit (bất kỳ thread nào)
│ enqueue Action
▼
Worker Thread (PubSubLib.EventChannel)
├─ Tạo Unit(id, type, pos, target) với IAlive
├─ _units[key] = unit
├─ Cell = GetGridCellByPosition(pos)
├─ cell.AddUnit(key)
├─ Lấy watcher trong cell → FireBatchEnter(watcherIds, unit)
│ ├─ User callback: OnUnitEnterBatch(tuple)
│ └─ (nếu có Natify) After* → serialize → NATS
│
AddWatcher (bất kỳ thread nào)
│ enqueue Action
▼
Worker Thread
├─ GetAllGridCellsInRange(pos, radius) → danh sách cell
├─ Đăng ký watcher trong tất cả cell đó
├─ Thu thập mọi unit trong các cell
└─ FireSyncEnter(watcherId, [IUnit...])
├─ User callback: OnUnitEnterSync(tuple)
└─ (nếu có Natify) After* → serialize → NATS
2. Networked (có Natify)
Khi gắn Natify, mọi event được tự động publish lên NATS topic PubSub.Evt, và server subscribe PubSub.Cmd để nhận command từ client.
using Natify;
using PubSubLib;
// Tạo PubSub
var pubSub = IPubSub.Create(new PubSubConfig { GridSize = 100f });
// Gắn Natify
var natifyClient = await INatifyClient.CreateFast(
"nats://localhost:4222",
"PubSubServer", // tên server
"ServerGroup", // group
"VN", // region
"Router" // router name để nhận command
);
pubSub.AddNatify(natifyClient);
// Đăng ký callbacks như bình thường
pubSub.OnUnitEnter(tuple => { /* xử lý game logic */ });
pubSub.OnUnitEvent(tuple => { /* xử lý game logic */ });
// Mọi thứ khác giống standalone — event tự động publish + command tự động nhận
Cách PubSubNatifySync hoạt động:
┌─────────────────────────────────┐
│ PubSubNatifySync │
│ │
NATS PubSub.Cmd ──────▶── Subscribe<PubSubCommand> ──▶── OnCommand()
│ ├─ AddWatcher → pubSub.AddWatcher()
│ ├─ RemoveWatcher → pubSub.RemoveWatcher()
│ ├─ MoveWatcher → pubSub.MoveWatcher()
│ ├─ PingUnits → HandlePingUnits()
│ │ └─ Gom TypeGroup → Dictionary<string,Dictionary<long,int>>
│ │ └─ pubSub.WatcherPingUnits(watcherId, typeVersions)
│ └─ PublishEvent → pubSub.HandleNatifyPublishEvent()
│ │
│ After* callbacks │
│ ├─ AfterBatchEnter → serialize → Publish("PubSub.Evt", evt)
│ ├─ AfterBatchLeave → serialize → Publish("PubSub.Evt", evt)
│ ├─ AfterSyncEnter → serialize → Publish("PubSub.Evt", evt)
│ ├─ AfterSyncLeave → serialize → Publish("PubSub.Evt", evt)
│ └─ AfterUnitEvent → serialize → Publish("PubSub.Evt", evt)
│ │
└─────────────────────────────────┘
- Inbound: Subscribe
PubSub.Cmd→ dispatch theo loại command → gọi phương thức tương ứng trênpubSub - Outbound:
EventChannelcó cặp callbackOn*(user) vàAfter*(Natify).After*serialize thành protobuf và publish lênPubSub.Evt HandlePingUnitsgom tất cảTypeGrouptrongPingUnitsCmdthành 1Dictionary<string, Dictionary<long, int>>, gọiWatcherPingUnitsđúng 1 lần
Luồng xử lý nội bộ
Thread Model
Code của bạn (bất kỳ thread) Worker Thread ("PubSubLib.EventChannel")
│ │
│── pubSub.AddWatcher(...) │
│── pubSub.CreateUnitAsync(...) │
│── unit.Position = ... │
│ (enqueue Action vào Channel) │
│ │
│ ═══ Channel<Action> ═════════════════▶ │
│ │── Đọc action từ channel
│ │── Thực thi action
│ │── Gọi FireBatchEnter / FireSyncEnter / ...
│ │── Gọi user callback (try/catch)
│ │── Gọi After* callback (→ Natify publish)
│ │── OnIdleCheck (→ dọn watcher hết hạn)
│ │
│◀─ FlushAsync hoàn thành ────────────────── │
Chi tiết triển khai:
Channel.CreateUnbounded<Action>()— không lock, thread-safe- 1 background thread duy nhất tên
"PubSubLib.EventChannel"— không cần lock trên_units,_cells,_watchers FlushAsync()enqueueTaskCompletionSource.SetResult()— await nó đảm bảo mọi action trước đó đã xử lý xong- Callback được gọi trong try/catch — 1 subscriber lỗi không crash worker thread
- Idle check chạy mỗi vòng lặp: kiểm tra và xóa watcher hết hạn
Worker loop (EventChannel.cs:60-78):
private void WorkerLoop()
{
while (!_cts.IsCancellationRequested)
{
// Đợi có action (timeout 1s để không block cancel)
_reader.WaitToReadAsync(_cts.Token).AsTask().Wait(1000);
// Xử lý tất cả action đang có
while (_reader.TryRead(out var action))
{
try { action(); } catch { }
}
// Idle: kiểm tra watcher hết hạn
try { _onIdleCheck(); } catch { }
}
}
Spatial Grid
Thế giới được chia thành các cell vuông cạnh GridSize (mặc định 100):
Cell key: "{cellX}:{cellY}"
cellX = floor(x / gridSize)
cellY = floor(y / gridSize)
Mỗi cell chứa:
HashSet<UnitKey>— các unit trong cellHashSet<long>— các watcher đang quan sát cell
Các thao tác không gian:
| Thao tác | Logic |
|---|---|
AddUnit(pos) |
Xác định cell → thêm unit vào cell → thông báo watcher trong cell |
AddWatcher(pos, radius) |
GetAllGridCellsInRange() → đánh dấu watcher trong tất cả cell → thu thập unit → SyncEnter |
MoveWatcher(pos, radius) |
Tính cell diff (thêm/xóa) → SyncEnter/SyncLeave |
Unit.Position = newPos |
Nếu cell đổi → xóa unit khỏi cell cũ, thêm vào cell mới → BatchEnter/BatchLeave |
Event Model
BatchEnter / BatchLeave
Khi nào fire:
CreateUnit/CreateUnitAsync— unit xuất hiện trong tầm watcherunit.Position = ...— unit di chuyển vào/ra tầm watcherunit.Destroy()— unit bị xóa khỏi grid
Thứ tự xử lý:
- Xác định watcher bị ảnh hưởng (cell membership)
- Gọi user callback
OnUnitEnter/OnUnitLeave(batch overload) - Gọi
AfterBatchEnter/AfterBatchLeave(→ Natify publish nếu có)
SyncEnter / SyncLeave
Khi nào fire:
AddWatcher— danh sách toàn bộ unit trong tầmMoveWatcher— unit mới vào tầm (SyncEnter), unit rời tầm (SyncLeave)WatcherPingUnits— reconciliation: version khớp sai →SyncEnter, key thừa →SyncLeave
Lưu ý: RemoveWatcher không fire SyncLeave (watcher đã biến mất, không cần thông báo).
UnitEvent
Khi nào fire:
unit.PublishEvent("eventName", data, reliable)— forward tới mọi watcher trong cell chứa unit
Tham số reliable:
| Giá trị | Transport | Ý nghĩa |
|---------|-----------|---------|
| true (mặc định) | TCP | Đảm bảo delivery, dùng cho logic game quan trọng (attack, damage, state change) |
| false | UDP | Best-effort, dùng cho hiệu ứng không quan trọng (FX, animation trigger, positional update nhanh) |
Version Tracking
Mỗi unit có Version (int), tăng khi:
Positionthay đổi (giá trị khác, cell có thể đổi hoặc không)Datasetter được gọi
Cách dùng trong ping reconciliation:
Client ping: { "hero": { 42: 5 } } → client nghĩ unit 42 version 5
Server check: unit 42 version = 7 → version khác → SyncEnter với version 7 mới
Server check: unit 42 không tồn tại → SyncLeave với key [42]
WatcherPingUnits nhận Dictionary<string, Dictionary<long, int>> — map type → (unitId → version). Server:
- Refresh watcher expiration timestamp
- Với mỗi type trong
watcher.KnownTypes(các type đã từng ping):- Nếu type có trong ping → lấy
unitVersionstừ ping - Nếu type không có trong ping →
unitVersionsrỗng → sync toàn bộ unit trong cell
- Nếu type có trong ping → lấy
- So sánh version: thiếu/khớp sai →
FireSyncEnter, thừa key →FireSyncLeave
Watcher Expiration
Watcher phải ping định kỳ để duy trì sự sống. Nếu không ping trong WatcherTimeoutSeconds (mặc định 5), watcher tự động bị xóa.
Cơ chế:
DictionaryScore<long, long>— cấu trúc skip list + hash map (giống Redis ZSET), lưuwatcherId → expirationTimestampAddWatcher/WatcherPingUnits→ refresh timestampRemoveWatcher→ xóa timestamp- Worker idle check (
CheckIdle): mỗiWatcherCleanupIntervalSeconds(mặc định 2), gọiRangeByScore(0, nowTicks)— O(log n + k) — lấy tất cả watcher hết hạn
// Giữ watcher sống: ping mỗi 3 giây
while (running)
{
pubSub.WatcherPingUnits(1, new Dictionary<string, Dictionary<long, int>>());
await Task.Delay(3000);
}
// Watcher 1 sống. Watcher nào không ping trong 5s → tự động bị xóa.
SyncLeavekhông fire khi watcher hết hạn (giốngRemoveWatcher)- WatcherId bị xóa có thể tái sử dụng ngay
Memory Management
- Unit giữ strong reference tới target object (vd:
Player,Monster) yêu cầu implementIAlive IsAlivetrả vềfalsekhi target setIsAlive = falseTryResolveAlive()gặp unit dead trong bất kỳ thao tác nào (ping, cell query) → tự động xóa unit khỏi dictionary nội bộ + fire cleanup event- Best practice: Gọi
unit.Destroy()tường minh khi xóa entity. SetIsAlive = falsetrước khi bỏ reference. ListPool<T>(ConcurrentBag-based) tái sử dụngList<T>instance cho callback để giảm allocation.
| Product | Versions Compatible and additional computed target framework versions. |
|---|---|
| .NET | net5.0 was computed. net5.0-windows was computed. net6.0 was computed. net6.0-android was computed. net6.0-ios was computed. net6.0-maccatalyst was computed. net6.0-macos was computed. net6.0-tvos was computed. net6.0-windows was computed. net7.0 was computed. net7.0-android was computed. net7.0-ios was computed. net7.0-maccatalyst was computed. net7.0-macos was computed. net7.0-tvos was computed. net7.0-windows was computed. net8.0 was computed. net8.0-android was computed. net8.0-browser was computed. net8.0-ios was computed. net8.0-maccatalyst was computed. net8.0-macos was computed. net8.0-tvos was computed. net8.0-windows was computed. net9.0 is compatible. net9.0-android was computed. net9.0-browser was computed. net9.0-ios was computed. net9.0-maccatalyst was computed. net9.0-macos was computed. net9.0-tvos was computed. net9.0-windows was computed. net10.0 was computed. net10.0-android was computed. net10.0-browser was computed. net10.0-ios was computed. net10.0-maccatalyst was computed. net10.0-macos was computed. net10.0-tvos was computed. net10.0-windows was computed. |
| .NET Core | netcoreapp3.0 was computed. netcoreapp3.1 was computed. |
| .NET Standard | netstandard2.1 is compatible. |
| MonoAndroid | monoandroid was computed. |
| MonoMac | monomac was computed. |
| MonoTouch | monotouch was computed. |
| Tizen | tizen60 was computed. |
| Xamarin.iOS | xamarinios was computed. |
| Xamarin.Mac | xamarinmac was computed. |
| Xamarin.TVOS | xamarintvos was computed. |
| Xamarin.WatchOS | xamarinwatchos was computed. |
-
.NETStandard 2.1
- Gcoder.Collections (>= 1.3.0)
- Google.Protobuf (>= 3.34.1)
- Natify (>= 1.2.0)
- PubSubLib.Contracts (>= 2.0.1)
-
net9.0
- Gcoder.Collections (>= 1.3.0)
- Google.Protobuf (>= 3.34.1)
- Natify (>= 1.2.0)
- PubSubLib.Contracts (>= 2.0.1)
NuGet packages (1)
Showing the top 1 NuGet packages that depend on PubSubLib:
| Package | Downloads |
|---|---|
|
PubSubLib.Router
Package Description |
GitHub repositories
This package is not used by any popular GitHub repositories.
| Version | Downloads | Last Updated |
|---|---|---|
| 2.0.3 | 19 | 7/17/2026 |
| 2.0.2 | 35 | 7/17/2026 |
| 2.0.1 | 43 | 7/16/2026 |
| 2.0.0 | 43 | 7/15/2026 |
| 1.51.12 | 46 | 7/13/2026 |
| 1.51.11 | 50 | 7/13/2026 |
| 1.51.10 | 57 | 7/6/2026 |
| 1.51.8 | 97 | 7/6/2026 |
| 1.51.7 | 95 | 7/6/2026 |
| 1.51.6 | 96 | 7/6/2026 |
| 1.51.5 | 59 | 6/24/2026 |
| 1.51.4 | 65 | 6/24/2026 |
| 1.51.2 | 54 | 6/23/2026 |
| 1.51.1 | 99 | 6/23/2026 |
| 1.51.0 | 99 | 6/21/2026 |
| 1.50.0 | 58 | 6/21/2026 |
| 1.4.0 | 63 | 6/19/2026 |
| 1.3.0 | 110 | 6/19/2026 |
| 1.2.0 | 105 | 6/19/2026 |
| 1.1.0 | 115 | 6/18/2026 |