hakowork/charm
3
0
Fork 0
mirror of https://github.com/littensy/charm.git synced 2026-10-20 17:51:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: update charm sync api

Browse source
This commit is contained in:
littensy 2025-10-31 21:56:02 -07:00
parent 1e0b10aef6
commit 32788b19e1
13 changed files with 359 additions and 335 deletions
Download patch file Download diff file Expand all files Collapse all files

54
packages/charm-sync/src/__tests__/client.spec.luau
View file

@ -2,8 +2,6 @@ local Charm = require("../../Charm")
local client = require("../client")
local suite = require("../../../../test/suite")
local atom = Charm.atom
local subscribe = Charm.subscribe
local test = suite.test
local afterEach = suite.afterEach
@ -11,29 +9,29 @@ afterEach(function()
client.reset()
end)
test("should sync state payload", function()
local a = atom(1)
test("should hydrate atom", function()
local a = Charm.atom(1)
client.syncAtoms({ a = a })
client.applyPayloads({
client.register({ a = a })
client.hydrate({
{ type = "init", data = { a = 2 } },
})
assert(a() == 2, "should set a to the initial value")
end)
test("should sync state patch", function()
local a = atom({ b = 1, c = 2 })
test("should merge patch", function()
local a = Charm.atom({ b = 1, c = 2 })
local previous = a()
client.syncAtoms({ a = a })
client.applyPayloads({
client.register({ a = a })
client.hydrate({
{ type = "init", data = { a = a() } },
})
assert(a() == previous, "should set a to the initial value")
client.applyPayloads({
client.hydrate({
{ type = "patch", data = { a = { b = 3 } } },
})
@ -43,17 +41,17 @@ test("should sync state patch", function()
end)
test("should fix sparse array patches", function()
local a = atom({ 1 })
local a = Charm.atom({ 1 })
client.syncAtoms({ a = a })
client.applyPayloads({
client.register({ a = a })
client.hydrate({
{ type = "init", data = { a = a() } },
})
-- This patch would be the result of a diff between { 1 } and { 1, 2 } that
-- was JSON-serialized and deserialized. The client should be able to catch
-- this and convert the string key back to a number.
client.applyPayloads({
client.hydrate({
{ type = "patch", data = { a = { ["2"] = 2 } } },
})
@ -61,21 +59,21 @@ test("should fix sparse array patches", function()
assert(a()[2] == 2, "should convert serialized arrays to use numbers")
end)
test("should apply multiple patches", function()
local a = atom(1)
test("should merge multiple patches", function()
local a = Charm.atom(1)
local calls = {}
client.syncAtoms({ a = a })
client.register({ a = a })
subscribe(a, function(value)
Charm.subscribe(a, function(value)
table.insert(calls, value)
end)
client.applyPayloads({
client.hydrate({
{ type = "init", data = { a = a() } },
})
client.applyPayloads({
client.hydrate({
{ type = "patch", data = { a = 2 } },
{ type = "patch", data = { a = 3 } },
})
@ -85,25 +83,25 @@ test("should apply multiple patches", function()
end)
test("should batch patches", function()
local a = atom(0)
local b = atom(0)
local a = Charm.atom(0)
local b = Charm.atom(0)
local states = {}
client.syncAtoms({ a = a, b = b })
client.register({ a = a, b = b })
subscribe(a, function()
Charm.subscribe(a, function()
states.b = b()
end)
subscribe(b, function()
Charm.subscribe(b, function()
states.a = a()
end)
client.applyPayloads({
client.hydrate({
{ type = "init", data = { a = a(), b = b() } },
})
client.applyPayloads({
client.hydrate({
{ type = "patch", data = { a = 1, b = 2 } },
})

16
packages/charm-sync/src/__tests__/e2e.spec.luau
View file

@ -20,11 +20,11 @@ test("should replicate server state to client", function()
local clientA = Charm.atom({ count = -1 })
local clientB = Charm.atom({ count = -1 })
client.syncAtoms({ a = clientA, b = clientB })
server.syncAtomsToClient(target, { a = serverA, b = serverB })
server.startSync(function(syncTarget, payloads)
client.register({ a = clientA, b = clientB })
server.subscribeClient(target, { a = serverA, b = serverB })
server.connect(function(syncTarget, payloads)
if syncTarget == target then
client.applyPayloads(payloads)
client.hydrate(payloads)
end
end)
@ -54,11 +54,11 @@ test("should replicate historical server state to client", function()
config.preserveHistory = true
client.syncAtoms({ a = clientA, b = clientB })
server.syncAtomsToClient(target, { a = serverA, b = serverB })
server.startSync(function(syncTarget, payloads)
client.register({ a = clientA, b = clientB })
server.subscribeClient(target, { a = serverA, b = serverB })
server.connect(function(syncTarget, payloads)
if syncTarget == target then
client.applyPayloads(payloads)
client.hydrate(payloads)
end
end)

50
packages/charm-sync/src/__tests__/server.spec.luau
View file

@ -20,8 +20,8 @@ test("should replicate state changes", function()
local c = Charm.atom(0)
local target, payloads
server.syncAtomsToClient(client, { a = a, b = b, c = c })
server.startSync(function(...)
server.subscribeClient(client, { a = a, b = b, c = c })
server.connect(function(...)
target, payloads = ...
end)
@ -47,8 +47,8 @@ test("should replicate partial state patches", function()
local b = Charm.atom(0 :: number?)
local target, payloads
server.syncAtomsToClient(client, { a = a, b = b })
server.startSync(function(...)
server.subscribeClient(client, { a = a, b = b })
server.connect(function(...)
target, payloads = ...
end)
@ -75,8 +75,8 @@ test("should init new clients", function()
local a = Charm.atom(0)
local payloadsA, payloadsB
server.syncAtomsToClient(clientA, { a = a })
server.startSync(function(target, payloads)
server.subscribeClient(clientA, { a = a })
server.connect(function(target, payloads)
if target == clientA then
payloadsA = payloads
elseif target == clientB then
@ -89,7 +89,7 @@ test("should init new clients", function()
assert(payloadsA[1].type == "init", "Expected first payload to be init for client A")
assert(payloadsA[1].data.a == a(), "Expected init payload to contain atom 'a' for client A")
server.syncAtomsToClient(clientB, { a = a })
server.subscribeClient(clientB, { a = a })
a(2)
server.flush()
assert(payloadsA[1].type == "patch", "Expected first payload to be patch for client A")
@ -104,13 +104,13 @@ test("should init new atoms", function()
local b = Charm.atom(0)
local target, payloads
server.syncAtomsToClient(client, { a = a })
server.startSync(function(...)
server.subscribeClient(client, { a = a })
server.connect(function(...)
target, payloads = ...
end)
server.flush()
server.syncAtomsToClient(client, { b = b })
server.subscribeClient(client, { b = b })
a(1)
b(1)
server.flush()
@ -129,9 +129,9 @@ test("should only replicate to synced clients", function()
local b = Charm.atom(0)
local payloadsA, payloadsB
server.syncAtomsToClient(clientA, { a = a })
server.syncAtomsToClient(clientB, { b = b })
server.startSync(function(target, payloads)
server.subscribeClient(clientA, { a = a })
server.subscribeClient(clientB, { b = b })
server.connect(function(target, payloads)
if target == clientA then
payloadsA = payloads
elseif target == clientB then
@ -170,8 +170,8 @@ test("should validate state", function()
local client = {}
local a = Charm.atom({})
server.syncAtomsToClient(client, { a = a })
server.startSync(function() end)
server.subscribeClient(client, { a = a })
server.connect(function() end)
a({ [0.5] = true })
assert(not pcall(server.flush), "Expected server.flush to fail")
@ -185,8 +185,8 @@ test("should serialize state", function()
local a = Charm.atom({ 1, 2, 3, 4, 5 })
local payloads
server.syncAtomsToClient(client, { a = a })
server.startSync(function(_, p)
server.subscribeClient(client, { a = a })
server.connect(function(_, p)
payloads = p
end)
server.flush()
@ -213,8 +213,8 @@ describe("historical", function()
local c = Charm.atom(0)
local target, payloads
server.syncAtomsToClient(client, { a = a, b = b, c = c })
server.startSync(function(...)
server.subscribeClient(client, { a = a, b = b, c = c })
server.connect(function(...)
target, payloads = ...
end)
@ -242,8 +242,8 @@ describe("historical", function()
local a = Charm.atom(0)
local payloadsA, payloadsB
server.syncAtomsToClient(clientA, { a = a })
server.startSync(function(target, payloads)
server.subscribeClient(clientA, { a = a })
server.connect(function(target, payloads)
if target == clientA then
payloadsA = payloads
elseif target == clientB then
@ -259,7 +259,7 @@ describe("historical", function()
assert(payloadsA[2].type == "patch", "Expected second payload to be patch for client A")
assert(payloadsA[2].data.a == 2, "Expected second patch payload to contain atom 'a' with value 1 for client A")
server.syncAtomsToClient(clientB, { a = a })
server.subscribeClient(clientB, { a = a })
a(3)
a(4)
server.flush()
@ -278,8 +278,8 @@ describe("historical", function()
local b = Charm.atom(0)
local payloads
server.syncAtomsToClient(client, { a = a })
server.startSync(function(_, ...)
server.subscribeClient(client, { a = a })
server.connect(function(_, ...)
payloads = ...
end)
@ -287,7 +287,7 @@ describe("historical", function()
assert(payloads[1].type == "init", "Expected first payload to be init")
assert(payloads[1].data.a == 0, "Expected init payload to contain atom 'a'")
server.syncAtomsToClient(client, { b = b })
server.subscribeClient(client, { b = b })
a(1)
a(2)
b(1)

4
packages/charm-sync/src/__tests__/spec.luau
View file

@ -8,8 +8,8 @@ suite.describe("e2e", function()
require("./e2e.spec")
end)
suite.describe("flatten", function()
require("./flatten.spec")
suite.describe("utils", function()
require("./utils.spec")
end)
suite.describe("patch", function()

7
packages/charm-sync/src/__tests__/flatten.spec.luau → packages/charm-sync/src/__tests__/utils.spec.luau
View file

@ -1,17 +1,18 @@
local Charm = require("../../Charm")
local flatten = require("../flatten")
local suite = require("../../../../test/suite")
local utils = require("../utils")
local atom = Charm.atom
local test = suite.test
test("should flatten atoms", function()
local getF, setF = Charm.signal(0)
local input = {
a = atom(0),
b = { c = atom(0) },
d = { e = { f = atom(0) } },
d = { e = { f = utils.signalToAtom(getF, setF) } },
}
local result = flatten(input)
local result = utils.flattenState(input)
assert(result.a == input.a, "should flatten the first level")
assert(result["b/c"] == input.b.c, "should flatten the second level")

60
packages/charm-sync/src/client.luau
View file

@ -2,46 +2,43 @@ local Charm = require("../Charm")
local patch = require("./patch")
local types = require("./types")
type Atom<T = any> = Charm.Atom<T>
type SyncPayload = types.SyncPayload
local atomRegister: { [string]: Charm.Atom } = {}
local trackedAtoms: { [string]: Atom } = {}
--- Registers atoms to be kept in sync with the server. Atoms with matching
--- keys on the server will be synced to these atoms.
--- Registers a map of atoms to the state with the corresponding keys on the
--- server. When the server sends updates, these atoms will be hydrated with
--- the new values.
---
--- @param atoms An object mapping string keys to atoms to be synced.
local function syncAtoms(atoms: { [string]: Charm.Atom })
--- If syncing signals instead of atoms, you can use `signalToAtom` to sync
--- signal getters and setters.
---
--- @param atoms A map of atoms to sync with the server.
local function register(atoms: { [string]: Atom })
for key, atom in atoms do
if atomRegister[key] and atomRegister[key] ~= atom then
warn("[Charm] Attempted to sync different atoms with the same key:", key)
continue
end
atomRegister[key] = atom
trackedAtoms[key] = atom
end
end
--- Unregisters one or more atoms from being kept in sync with the server.
--- The atoms will retain their current values, but will no longer receive
--- updates from the server.
---
--- @param keys The keys of the atoms to stop syncing.
local function unsyncAtoms(...: string)
--- Unregisters from server state updates for the given keys. The atoms
--- will retain their current values, but will no longer receive updates
--- from the server.
local function unregister(...: string)
for index = 1, select("#", ...) do
local key = select(index, ...)
atomRegister[key] = nil
trackedAtoms[key] = nil
end
end
--- Unregisters all atoms from being kept in sync with the server. The
--- atoms will retain their current values, but will no longer receive
--- updates from the server.
--- Unsubscribes from all server state updates.
local function reset()
table.clear(atomRegister)
table.clear(trackedAtoms)
end
local function applyInitPayload(data: { [string]: any })
for key, value in data do
local atom = atomRegister[key]
local atom = trackedAtoms[key]
if atom then
atom(value)
end
@ -51,23 +48,22 @@ end
local function applyPatchPayload(data: { [string | number]: any })
local target = {}
for key, atom in atomRegister do
for key, atom in trackedAtoms do
target[key] = atom()
end
target = patch.apply(target, data)
for key, atom in atomRegister do
for key, atom in trackedAtoms do
atom(target[key])
end
end
--- Applies one or more payloads received from the server to the synced
--- atoms. Payloads are state patches or initial state updates sent by the
--- server, and are usually received via a remote event.
--- Merges the server's state patches into the client's state, either
--- initializing it with the full state or merging a partial patch.
---
--- @param payloads The payloads to apply.
local function applyPayloads(payloads: { SyncPayload })
--- @param payloads The state updates received from the server.
local function hydrate(payloads: { SyncPayload })
for _, payload in payloads do
if payload.type == "init" then
Charm.batched(applyInitPayload, payload.data)
@ -78,8 +74,8 @@ local function applyPayloads(payloads: { SyncPayload })
end
return {
syncAtoms = syncAtoms,
unsyncAtoms = unsyncAtoms,
register = register,
unregister = unregister,
reset = reset,
applyPayloads = applyPayloads,
hydrate = hydrate,
}

30
packages/charm-sync/src/flatten.luau
View file

@ -1,30 +0,0 @@
type NestedAtomMap = {
[string]: NestedAtomMap | () -> any,
}
--- Flattens a nested atom map into a single object with slash-separated keys.
--- Useful for recursively collecting atoms returned by modules.
---
--- @param atoms The nested atom map to flatten.
--- @return A flattened atom map.
local function flatten(atoms: NestedAtomMap): { [string]: () -> any }
local result: { [string]: () -> any } = {}
local function visit(node: NestedAtomMap, path: string)
for key, value in node do
local location = if path == "" then key else path .. "/" .. key
if type(value) == "table" then
visit(value, location)
else
result[location] = value
end
end
end
visit(atoms, "")
return result
end
return flatten

130
packages/charm-sync/src/index.d.ts vendored
View file

@ -33,29 +33,29 @@ type DataType = DataTypes[keyof DataTypes];
* If a value was removed, it is replaced with `None`. This can be checked
* using the `isNone` function.
*/
export type SyncPatch<State, Serialize extends boolean = true> = MaybeNone<State> | State extends
export type SyncPatch<State, FixArrays extends boolean = true> = MaybeNone<State> | State extends
| ReadonlyMap<infer K, infer V>
| Map<infer K, infer V>
? ReadonlyMap<K, SyncPatch<V, Serialize> | None>
? ReadonlyMap<K, SyncPatch<V, FixArrays> | None>
: State extends Set<infer T> | ReadonlySet<infer T>
? ReadonlyMap<T, true | None>
: State extends readonly (infer T)[]
? Serialize extends true
? ReadonlyMap<string | number, SyncPatch<T, Serialize> | None>
: readonly (SyncPatch<T, Serialize> | None | undefined)[]
? FixArrays extends true
? ReadonlyMap<string | number, SyncPatch<T, FixArrays> | None>
: readonly (SyncPatch<T, FixArrays> | None | undefined)[]
: State extends DataType
? State
: State extends object
? { readonly [P in keyof State]?: SyncPatch<State[P], Serialize> }
? { readonly [P in keyof State]?: SyncPatch<State[P], FixArrays> }
: State;
/**
* A payload that can be sent from the server to the client to synchronize
* state between the two.
*/
export type SyncPayload<Selectors extends SelectorMap = SelectorMap, Serialize extends boolean = true> =
export type SyncPayload<Selectors extends SelectorMap = SelectorMap, FixArrays extends boolean = true> =
| { type: "init"; data: StateOfMap<Selectors> }
| { type: "patch"; data: SyncPatch<StateOfMap<Selectors>, Serialize> };
| { type: "patch"; data: SyncPatch<StateOfMap<Selectors>, FixArrays> };
/**
* Global configuration options that affect the behavior of Charm Sync.
@ -102,38 +102,37 @@ export const config: {
*/
export namespace client {
/**
* Registers atoms to be kept in sync with the server. Atoms with matching
* keys on the server will be synced to these atoms.
* Registers a map of atoms to the state with the corresponding keys on the
* server. When the server sends updates, these atoms will be hydrated with
* the new values.
*
* @param atoms An object mapping string keys to atoms to be synced.
* If syncing signals instead of atoms, you can use `signalToAtom` to sync
* signal getters and setters as atoms.
*
* @param atoms A map of atoms to sync with the server.
*/
export function syncAtoms(atoms: { [key: string]: Atom<any> }): void;
export function register(atoms: { [key: string]: Atom<any> }): void;
/**
* Unregisters one or more atoms from being kept in sync with the server.
* The atoms will retain their current values, but will no longer receive
* updates from the server.
*
* @param keys The keys of the atoms to stop syncing.
* Unregisters from server state updates for the given keys. The atoms
* will retain their current values, but will no longer receive updates
* from the server.
*/
export function unsyncAtoms(...keys: string[]): void;
export function unregister(...keys: string[]): void;
/**
* Unregisters all atoms from being kept in sync with the server. The
* atoms will retain their current values, but will no longer receive
* updates from the server.
* Unsubscribes from all server state updates.
*/
export function reset(): void;
/**
* Applies one or more payloads received from the server to the synced
* atoms. Payloads are state patches or initial state updates sent by the
* server, and are usually received via a remote event.
* Merges the server's state patches into the client's state, either
* initializing it with the full state or merging a partial patch.
*
* @param payloads The payloads to apply.
* @param payloads The state updates received from the server.
*/
export function applyPayloads<Atoms extends { [key: string]: Atom<any> }, Serialize extends boolean = true>(
payloads: SyncPayload<Atoms, Serialize>[],
export function hydrate<Atoms extends { [key: string]: Atom<any> }, FixArrays extends boolean = true>(
payloads: SyncPayload<Atoms, FixArrays>[],
): void;
}
@ -142,61 +141,62 @@ export namespace client {
*/
export namespace server {
/**
* Registers atoms to be kept in sync with a specific client. Should be
* called when a player joins the game to start syncing atoms to them. Call
* `unsyncAtomsFromClient` or `unsyncClient` to stop syncing.
* Subscribes a client to the given atoms. When an update occurs, the client
* will receive a partial state patch to merge into their local state. May be
* called multiple times to subscribe to additional keys.
*
* Each atom should be assigned a unique key, allowing the server to re-use
* patches between multiple clients.
* Atoms, signals converted to atoms, and signal getter functions are allowed.
* Note that the client still requires some way to set the state, so
* registering signals on the client should be done through `signalToAtom()`.
*
* @param client The player to sync atoms to.
* @param atoms An object mapping string keys to atoms to be synced.
* @param client The client receiving state updates.
* @param atoms A map of atoms to sync with the client.
*/
export function syncAtomsToClient(client: Player, atoms: { [key: string]: Atom<any> }): void;
export function subscribeClient(client: Player, atoms: { [key: string]: Atom<any> }): void;
/**
* Unregisters one or more atoms from being kept in sync with a specific
* client.
* Unsubscribes a client from receiving all state updates. To only unsubscribe
* from specific keys, use `unsubscribeClientFrom` instead.
*
* @param client The player to stop syncing atoms to.
* @param keys The keys of the atoms to stop syncing.
* @param client The client receiving state updates.
*/
export function unsyncAtomsFromClient(client: Player, ...keys: string[]): void;
export function unsubscribeClient(client: Player): void;
/**
* Unregisters all atoms from being kept in sync with a specific client.
* Should be called when a player leaves the game.
* Unsubscribes a client from receiving updates for the given keys.
*
* @param client The player to stop syncing atoms to.
* @param client The client receiving state updates.
* @param keys The keys of the state to unsubscribe from.
*/
export function unsyncClient(client: Player): void;
export function unsubscribeClientFrom(client: Player, ...keys: string[]): void;
/**
* Starts automatically syncing registered atoms to clients. Patches will
* be sent to clients whenever the atoms are updated.
* Uses the callback to send state updates to clients that are subscribed
* to state changes. Starts automatically sending updates at the interval
* specified in `config.interval`.
*
* @param onSync The function to call when patches can be sent to a client.
* @param onSync Called when sending patches to a client.
*/
export function startSync<Atoms extends { [key: string]: Atom<any> }, Serialize extends boolean = true>(
onSync: (client: Player, payloads: SyncPayload<Atoms, Serialize>[]) => void,
export function connect<Atoms extends { [key: string]: Atom<any> }, FixArrays extends boolean = true>(
onSync: (client: Player, payloads: SyncPayload<Atoms, FixArrays>[]) => void,
): void;
/**
* Stops automatically syncing registered atoms to clients. Patches will
* no longer be sent to clients until `startSync` is called again.
* Stops syncing state updates to clients at the automatic interval.
* Flushing can still be performed manually via `flush`.
*/
export function stopSync(): void;
export function disconnect(): void;
/**
* Immediately sends pending updates and initial states to all clients.
* Normally, updates are sent at the interval specified in `config.interval`.
* Calling this function forces all pending updates to be sent immediately.
* Immediately sends pending updates to all clients. Normally, updates are
* sent at the interval specified in `config.interval`, but this function
* can be used to force an immediate sync.
*/
export function flush(): void;
/**
* Unregisters all atoms from being kept in sync with all clients, and
* stops automatic syncing.
* Unsubscribes all clients, stops tracking all state, and disconnects
* from the automatic sync interval.
*/
export function reset(): void;
}
@ -204,8 +204,8 @@ export namespace server {
export namespace patch {
/**
* Converts `nil` values to the special `None` symbol. This is useful when
* preparing data for serialization, since `nil` can represent both the absence
* of a value and the removal of a value in patches.
* preparing data for serialization, since `nil` can represent both the
* absence of a value and the removal of a value in patches.
*/
export function nilToNone<T>(value: T | undefined): T | None;
@ -217,10 +217,10 @@ export namespace patch {
* @param nextState The next state.
* @returns A patch that can be applied to the current state.
*/
export function diff<State, Serialize extends boolean = true>(
export function diff<State, FixArrays extends boolean = true>(
currentState: State,
nextState: State,
): SyncPatch<State, Serialize>;
): SyncPatch<State, FixArrays>;
/**
* Applies a patch to a state and returns the resulting value.
@ -229,9 +229,9 @@ export namespace patch {
* @param statePatch The patches to apply.
* @returns The new state with the patch applied.
*/
export function apply<State, Serialize extends boolean = true>(
export function apply<State, FixArrays extends boolean = true>(
currentState: State,
statePatch: SyncPatch<State, Serialize>,
statePatch: SyncPatch<State, FixArrays>,
): State;
}
@ -247,8 +247,6 @@ type AppendPath<Path extends Key | undefined, Name extends Key> = undefined exte
: never
: never;
type UnionToIntersection<U> = (U extends any ? (k: U) => void : never) extends (k: infer I) => void ? I : never;
type IntersectValues<T> = UnionToIntersection<{ readonly [K in keyof T]: T[K] }[keyof T]>;
export type FlattenNestedAtoms<
@ -269,7 +267,7 @@ export type FlattenNestedAtoms<
* @param atoms The nested atom map to flatten.
* @returns A flattened atom map.
*/
export function flatten<Atoms extends NestedAtomMap>(atoms: Atoms): FlattenNestedAtoms<Atoms>;
export function flattenAtoms<Atoms extends NestedAtomMap>(atoms: Atoms): FlattenNestedAtoms<Atoms>;
/**
* Checks whether a value is `None`. If `true`, the value is scheduled to be

5
packages/charm-sync/src/init.luau
View file

@ -1,9 +1,9 @@
local client = require("@self/client")
local config = require("@self/config")
local flatten = require("@self/flatten")
local patch = require("@self/patch")
local server = require("@self/server")
local types = require("@self/types")
local utils = require("@self/utils")
export type None = types.None
export type SyncPayload = types.SyncPayload
@ -14,5 +14,6 @@ return {
server = server,
patch = patch,
isNone = patch.isNone,
flatten = flatten,
flattenState = utils.flattenState,
signalToAtom = utils.signalToAtom,
}

273
packages/charm-sync/src/server.luau
View file

@ -7,85 +7,95 @@ type SyncPayload = types.SyncPayload
local globals = Charm.globals
type RegisteredAtom<T = any> = {
callback: () -> T,
type SyncStatus = number
type ClientAtoms = {
[string]: SyncStatus,
}
type TrackedState<T = any> = {
getter: () -> T,
currentValue: T,
previousValue: T,
dispose: () -> (),
}
type SyncStatus = number
type StateUpdate = {
key: string,
currentValue: any,
previousValue: any,
}
local PENDING_INITIAL_STATE: SyncStatus = 0
local LISTENING_FOR_CHANGES: SyncStatus = 1
local atomRegister: { [string]: RegisteredAtom } = {}
local clientRegister: { [any]: { [string]: SyncStatus } } = {}
local globalUpdates: { { key: string, currentValue: any, previousValue: any } } = {}
local trackedAtoms: { [string]: TrackedState } = {}
local globalUpdates: { StateUpdate } = {}
local clients: { [any]: ClientAtoms } = {}
local connection: RBXScriptConnection?
local onSyncCallback: (client: any, payloads: { SyncPayload }) -> ()?
local syncCallback: (client: any, payloads: { SyncPayload }) -> ()?
local function syncGlobalAtom<T>(key: string, callback: () -> T)
if atomRegister[key] then
if atomRegister[key].callback ~= callback then
warn("[Charm] Attempted to sync different atoms with the same key:", key)
local function startTracking<T>(key: string, getter: () -> T)
if trackedAtoms[key] then
-- Ignore duplicates, but warn against name collisions
if trackedAtoms[key].getter ~= getter then
warn("[Charm] Attempted to sync different states with the same key:", key)
end
return
end
local registeredAtom: RegisteredAtom<T> = {
callback = callback,
currentValue = callback(),
previousValue = callback(),
dispose = function() end,
local value = getter()
local trackedState = {
getter = getter,
currentValue = value,
previousValue = value,
}
registeredAtom.dispose = Charm.effect(function()
local newValue = callback()
trackedState.dispose = Charm.effect(function()
local newValue = getter()
-- When preserving history, we need to keep track of previous values
-- so we can create granular patches for each change.
if config.preserveHistory and registeredAtom.currentValue ~= newValue then
if config.preserveHistory and trackedState.currentValue ~= newValue then
table.insert(globalUpdates, {
key = key,
currentValue = newValue,
previousValue = registeredAtom.currentValue,
previousValue = trackedState.previousValue,
})
end
registeredAtom.currentValue = newValue
trackedState.currentValue = newValue
end)
atomRegister[key] = registeredAtom
trackedAtoms[key] = trackedState
end
local function unsyncGlobalAtom(key: string)
local registeredAtom = atomRegister[key]
if registeredAtom then
registeredAtom.dispose()
atomRegister[key] = nil
local function stopTracking(key: string)
local tracker = trackedAtoms[key]
if tracker then
tracker.dispose()
trackedAtoms[key] = nil
end
end
local function unsyncStaleGlobalAtoms()
local function purgeUntracked()
local active: { [string]: boolean } = {}
for _, clientAtoms in clientRegister do
for _, clientAtoms in clients do
for key in clientAtoms do
active[key] = true
end
end
for key in atomRegister do
for key in trackedAtoms do
if not active[key] then
unsyncGlobalAtom(key)
stopTracking(key)
end
end
end
local function createGlobalPatchHistory(): { any }
local function diffGlobalUpdateBuffer(): { any }
local patchCount = #globalUpdates
local patches = table.create(patchCount)
@ -96,10 +106,10 @@ local function createGlobalPatchHistory(): { any }
return patches
end
local function createPayloadHistoryForClient(client: any, patches: { any }): { SyncPayload }
local syncedAtoms = clientRegister[client]
local function selectFromGlobalPatchList(client: any, globalPatches: { any }): { SyncPayload }
local clientAtoms = clients[client]
if not syncedAtoms then
if not clientAtoms then
return {}
end
@ -108,14 +118,14 @@ local function createPayloadHistoryForClient(client: any, patches: { any }): { S
-- Iterate over all global updates and see if the client is interested in
-- any of them. If so, we create a patch payload for the update.
for index, update in globalUpdates do
local patchData = patches[index]
local patchData = globalPatches[index]
if not patchData then
continue
end
local key = update.key
local status = syncedAtoms[key]
local status = clientAtoms[key]
if status == LISTENING_FOR_CHANGES then
table.insert(payloads, {
@ -127,179 +137,180 @@ local function createPayloadHistoryForClient(client: any, patches: { any }): { S
type = "init",
data = { [key] = update.currentValue },
})
syncedAtoms[key] = LISTENING_FOR_CHANGES
clientAtoms[key] = LISTENING_FOR_CHANGES
end
end
-- If a pending atom was never updated, we still need to send its initial
-- state. This can happen if the atom was registered after its last change.
local initPayload: { [string]: any } = {}
local initData: { [string]: any } = {}
for key, status in syncedAtoms do
local registeredAtom = atomRegister[key]
for key, status in clientAtoms do
local trackedState = trackedAtoms[key]
if status == PENDING_INITIAL_STATE and registeredAtom then
initPayload[key] = registeredAtom.currentValue
syncedAtoms[key] = LISTENING_FOR_CHANGES
if status == PENDING_INITIAL_STATE and trackedState then
initData[key] = trackedState.currentValue
clientAtoms[key] = LISTENING_FOR_CHANGES
end
end
if next(initPayload) ~= nil then
table.insert(payloads, 1, { type = "init" :: "init", data = initPayload })
if next(initData) ~= nil then
table.insert(payloads, 1, { type = "init" :: "init", data = initData })
end
return payloads
end
local function createGlobalPatch(): any
local current: { [string]: any } = {}
local previous: { [string]: any } = {}
local function diffGlobalState(): any
local currentGlobalState: { [string]: any } = {}
local previousGlobalState: { [string]: any } = {}
for key, registeredAtom in atomRegister do
current[key] = registeredAtom.currentValue
previous[key] = registeredAtom.previousValue
registeredAtom.previousValue = registeredAtom.currentValue
for key, tracker in trackedAtoms do
currentGlobalState[key] = tracker.currentValue
previousGlobalState[key] = tracker.previousValue
tracker.previousValue = tracker.currentValue
end
return patch.diff(previous, current)
return patch.diff(previousGlobalState, currentGlobalState)
end
local function createPayloadForClient(client: any, patchData: any): { SyncPayload }
local syncedAtoms = clientRegister[client]
local payloads: { SyncPayload } = {}
local function selectFromGlobalPatch(client: any, globalPatch: any): { SyncPayload }
local clientAtoms = clients[client]
if not syncedAtoms then
return payloads
if not clientAtoms then
return {}
end
local initPayload: { [string]: any } = {}
local patchPayload: { [string]: any } = {}
local payloads: { SyncPayload } = {}
local initData: { [string]: any } = {}
local patchData: { [string]: any } = {}
-- Create patches for initialized atoms that changed, and add atoms that
-- are still pending initial state to the init payload.
for key, status in syncedAtoms do
local registeredAtom = atomRegister[key]
for key, status in clientAtoms do
local trackedState = trackedAtoms[key]
if status == LISTENING_FOR_CHANGES then
patchPayload[key] = patchData[key]
patchData[key] = globalPatch[key]
elseif status == PENDING_INITIAL_STATE then
initPayload[key] = registeredAtom.currentValue
syncedAtoms[key] = LISTENING_FOR_CHANGES
initData[key] = trackedState.currentValue
clientAtoms[key] = LISTENING_FOR_CHANGES
end
end
if next(initPayload) ~= nil then
table.insert(payloads, { type = "init", data = initPayload })
if next(initData) ~= nil then
table.insert(payloads, { type = "init", data = initData })
end
if next(patchPayload) ~= nil then
table.insert(payloads, { type = "patch", data = patchPayload })
if next(patchData) ~= nil then
table.insert(payloads, { type = "patch", data = patchData })
end
return payloads
end
--- Registers atoms to be kept in sync with a specific client. Should be
--- called when a player joins the game to start syncing atoms to them. Call
--- `unsyncAtomsFromClient` or `unsyncClient` to stop syncing.
--- Subscribes a client to the given atoms. When an update occurs, the client
--- will receive a partial state patch to merge into their local state. May be
--- called multiple times to subscribe to additional keys.
---
--- Each atom should be assigned a unique key, allowing the server to re-use
--- patches between multiple clients.
--- Atoms, signals converted to atoms, and signal getter functions are allowed.
--- Note that the client still requires some way to set the state, so
--- registering signals on the client should be done through `signalToAtom()`.
---
--- @param client The player to sync atoms to.
--- @param atoms An object mapping string keys to atoms to be synced.
local function syncAtomsToClient(client: any, atoms: { [string]: () -> any })
if not clientRegister[client] then
clientRegister[client] = {}
--- @param client The client receiving state updates.
--- @param atoms A map of atoms to sync with the client.
local function subscribeClient(client: any, atoms: { [string]: () -> any })
if not clients[client] then
clients[client] = {}
end
local syncedAtoms = clientRegister[client]
local clientAtoms = clients[client]
for key, atom in atoms do
syncGlobalAtom(key, atom)
for key, state in atoms do
startTracking(key, if type(state) == "function" then state else state.get)
if not syncedAtoms[key] then
if not clientAtoms[key] then
-- Mark the atom as pending initial state so we send its full value
-- the next time we sync.
syncedAtoms[key] = PENDING_INITIAL_STATE
clientAtoms[key] = PENDING_INITIAL_STATE
end
end
end
--- Unregisters all atoms from being kept in sync with a specific client.
--- Should be called when a player leaves the game.
--- Unsubscribes a client from receiving all state updates. To only unsubscribe
--- from specific keys, use `unsubscribeClientFrom` instead.
---
--- @param client The player to stop syncing atoms to.
local function unsyncClient(client: any)
clientRegister[client] = nil
--- @param client The client receiving state updates.
local function unsubscribeClient(client: any)
clients[client] = nil
end
--- Unregisters one or more atoms from being kept in sync with a specific
--- client.
--- Unsubscribes a client from receiving updates for the given keys.
---
--- @param client The player to stop syncing atoms to.
--- @param keys The keys of the atoms to stop syncing.
local function unsyncAtomsFromClient(client: any, ...: string)
local syncedAtoms = clientRegister[client]
--- @param client The client receiving state updates.
--- @param keys The keys of the state to unsubscribe from.
local function unsubscribeClientFrom(client: any, ...: string)
local clientAtoms = clients[client]
if not syncedAtoms then
if not clientAtoms then
return
end
for index = 1, select("#", ...) do
local key = select(index, ...)
syncedAtoms[key] = nil
clientAtoms[key] = nil
end
if next(syncedAtoms) == nil then
clientRegister[client] = nil
if next(clientAtoms) == nil then
clients[client] = nil
end
end
--- Immediately sends pending updates and initial states to all clients.
--- Normally, updates are sent at the interval specified in `config.interval`.
--- Calling this function forces all pending updates to be sent immediately.
--- Immediately sends pending updates to all clients. Normally, updates are
--- sent at the interval specified in `config.interval`, but this function
--- can be used to force an immediate sync.
local function flush()
if not onSyncCallback then
if not syncCallback then
return
end
local patches = if config.preserveHistory then createGlobalPatchHistory() else createGlobalPatch()
local globalPatch = if config.preserveHistory then diffGlobalUpdateBuffer() else diffGlobalState()
-- While payload objects are technically mutable as they are unique to
-- each client, the data inside them is not safe to mutate.
if globals.frozen then
Charm.deepFreeze(patches)
Charm.deepFreeze(globalPatch)
end
for client in clientRegister do
for client in clients do
local payloads: { SyncPayload } = if config.preserveHistory
then createPayloadHistoryForClient(client, patches)
else createPayloadForClient(client, patches)
then selectFromGlobalPatchList(client, globalPatch)
else selectFromGlobalPatch(client, globalPatch)
if payloads[1] then
if task then
task.spawn(onSyncCallback, client, payloads)
task.spawn(syncCallback, client, payloads)
else
onSyncCallback(client, payloads)
syncCallback(client, payloads)
end
end
end
unsyncStaleGlobalAtoms()
purgeUntracked()
table.clear(globalUpdates)
end
--- Starts automatically syncing registered atoms to clients. Patches will
--- be sent to clients whenever the atoms are updated.
--- Uses the callback to send state updates to clients that are subscribed
--- to state changes. Starts automatically sending updates at the interval
--- specified in `config.interval`.
---
--- @param onSync The function to call when patches can be sent to a client.
local function startSync(onSync: (client: any, payloads: { SyncPayload }) -> ())
--- @param onSync Called when sending patches to a client.
local function connect(onSync: (client: any, payloads: { SyncPayload }) -> ())
if connection then
connection:Disconnect()
end
onSyncCallback = onSync
syncCallback = onSync
if not game then
return
@ -323,30 +334,30 @@ local function startSync(onSync: (client: any, payloads: { SyncPayload }) -> ())
end)
end
--- Stops automatically syncing registered atoms to clients. Patches will
--- no longer be sent to clients until `startSync` is called again.
local function stopSync()
--- Stops syncing state updates to clients at the automatic interval.
--- Flushing can still be performed manually via `flush`.
local function disconnect()
if connection then
connection:Disconnect()
connection = nil
end
end
--- Unregisters all atoms from being kept in sync with all clients, and
--- stops automatic syncing.
--- Unsubscribes all clients, stops tracking all state, and disconnects
--- from the automatic sync interval.
local function reset()
table.clear(atomRegister)
table.clear(clientRegister)
table.clear(trackedAtoms)
table.clear(clients)
table.clear(globalUpdates)
stopSync()
disconnect()
end
return {
syncAtomsToClient = syncAtomsToClient,
unsyncClient = unsyncClient,
unsyncAtomsFromClient = unsyncAtomsFromClient,
startSync = startSync,
stopSync = stopSync,
subscribeClient = subscribeClient,
unsubscribeClient = unsubscribeClient,
unsubscribeClientFrom = unsubscribeClientFrom,
connect = connect,
disconnect = disconnect,
flush = flush,
reset = reset,
}

52
packages/charm-sync/src/utils.luau Normal file
View file

@ -0,0 +1,52 @@
local Charm = require("../Charm")
type Atom<T> = Charm.Atom<T>
type NestedState = {
[string]: NestedState | () -> any,
}
type FlattenedState = {
[string]: () -> any,
}
local function isAtomOrSignal(value: any): boolean
return type(value) ~= "table" or (type(value.get) == "function" and type(value.set) == "function")
end
--- Flattens a nested map of atoms or signals into a single object with
--- slash-separated keys. Useful for recursively collecting atoms returned by
--- modules.
---
--- @param states A nested table of atoms or signals.
--- @return A flattened map of atoms or signals.
local function flattenState(states: NestedState): FlattenedState
local result: FlattenedState = {}
local function visit(node: NestedState, path: string)
for key, value in node do
local location = if path == "" then key else path .. "/" .. key
if isAtomOrSignal(value) then
result[location] = value :: () -> any
else
visit(value :: NestedState, location)
end
end
end
visit(states, "")
return result
end
local function signalToAtom<T>(getter: () -> T, setter: (value: ((T) -> T) | T) -> T): Atom<T>
return function(...)
return if select("#", ...) == 0 then getter() else setter((...))
end
end
return {
flattenState = flattenState,
signalToAtom = signalToAtom,
}

11
packages/charm/src/core.luau
View file

@ -312,7 +312,7 @@ end
--- @see https://github.com/littensy/charm?tab=readme-ov-file#atomstate-equals
local function atom<T>(initialValue: T, equals: ((current: T, incoming: T) -> boolean)?): (...((T) -> T) | T) -> T
local getter, setter = signal(initialValue, equals)
return function(...: ((T) -> T) | T): T
return function(...)
return if select("#", ...) == 0 then getter() else setter((...))
end
end
@ -321,9 +321,6 @@ end
--- The returned atom is read-only, and its value is automatically updated when
--- any atoms accessed from within the getter function change.
---
--- If an initial value is provided, the getter will receive this value as its
--- first argument on the initial run, instead of `undefined`.
---
--- @example
--- ```ts
--- const count = atom(1);
@ -332,8 +329,7 @@ end
--- ```
---
--- @param getter A function that produces the next value.
--- @param value An optional initial value for the selector.
--- @see https://github.com/littensy/charm?tab=readme-ov-file#computedcallback-options
--- @see https://github.com/littensy/charm?tab=readme-ov-file#computedgetter
local function computed<T>(getter: (previousValue: T?) -> T): () -> T
return computedOper({
flags = bit32.bor(MUTABLE, DIRTY),
@ -432,6 +428,7 @@ end
--- @param callback The callback function.
--- @param args Arguments to pass to the callback.
--- @return The return value of the callback.
--- @see https://github.com/littensy/charm?tab=readme-ov-file#untrackedcallback
local function untracked<Args..., Results...>(callback: (Args...) -> Results..., ...: Args...): Results...
local prevSub = setActiveSub(nil)
local results: { any } = { pcall(callback, ...) }
@ -450,6 +447,7 @@ end
--- @param callback The callback function.
--- @param args Arguments to pass to the callback.
--- @return The return value of the callback.
--- @see https://github.com/littensy/charm?tab=readme-ov-file#peekcallback
local function peek<Args..., Results...>(callback: (Args...) -> Results..., ...: Args...): Results...
if
activeSub
@ -474,6 +472,7 @@ end
--- @param callback A function that performs multiple atom updates.
--- @param args Arguments to pass to the callback.
--- @return The return value of the callback.
--- @see https://github.com/littensy/charm?tab=readme-ov-file#batchedcallback
local function batched<Args..., Results...>(callback: (Args...) -> Results..., ...: Args...): Results...
startBatch()
local results: { any } = { (if globals.strict then noYield else pcall)(callback, ...) }

2
packages/charm/src/index.d.ts vendored
View file

@ -83,11 +83,9 @@ export function atom<T>(): Atom<T | undefined>;
* ```
*
* @param getter A function that produces the next value.
* @param value An optional initial value for the selector.
* @see https://github.com/littensy/charm?tab=readme-ov-file#computedcallback-options
*/
export function computed<T>(getter: (previousValue?: T) => T): () => T;
export function computed<T>(getter: (previousValue: T) => T, value: T): () => T;
/**
* Creates an effect that runs a callback in response to atom state changes.