From e5f655482fd4cf896030c95e6f8c1b4d0e447cf0 Mon Sep 17 00:00:00 2001 From: Richard Date: Wed, 25 Sep 2024 22:10:19 -0700 Subject: [PATCH] refactor: separate code for preserving history --- src/sync/server.luau | 234 +++++++++++++++++++++++++++---------------- 1 file changed, 148 insertions(+), 86 deletions(-) diff --git a/src/sync/server.luau b/src/sync/server.luau index e13a4dc..a395abc 100644 --- a/src/sync/server.luau +++ b/src/sync/server.luau @@ -62,69 +62,103 @@ type ServerSyncer = { local function server(options: ServerOptions): ServerSyncer local self = {} :: ServerSyncer local sync: (player: Player, payload: SyncPayload) -> () - local snapshots: { { [string]: any } } = {} - local changed = false - -- Start the interval to send state patches over the network. - function self:connect(callback) - local cleanups = {} - sync = callback + local function createSnapshot() + local snapshot: { [string]: any } = {} - local function getSnapshot() - local snapshot: { [string]: any } = {} - for key, atom in next, options.atoms do - snapshot[key] = atom() - end - return snapshot - end - - local function pushSnapshot(key: string, current: any, previous: any) - if not options.preserveHistory then - if snapshots[2] then - snapshots[2][key] = current - else - snapshots[2] = getSnapshot() - end - return - end - - local lastSnapshot = snapshots[#snapshots] - local previousSnapshot = snapshots[#snapshots - 1] - - -- Optimize snapshots by updating the most recent snapshot if the - -- previous and current values are the same, since this allows us - -- to group multiple changes into a single snapshot. - if previousSnapshot and previousSnapshot[key] == previous and lastSnapshot[key] == previous then - lastSnapshot[key] = current - else - local nextSnapshot = table.clone(lastSnapshot) - nextSnapshot[key] = current - table.insert(snapshots, nextSnapshot) - end - end - - -- Populate the initial state and snapshot for each atom. - -- Subscribe to each atom and update the state when it changes. for key, atom in next, options.atoms do - cleanups[key] = subscribe(atom, function(current, previous) - pushSnapshot(key, current, previous) - changed = true - end) + snapshot[key] = atom() end - snapshots[1] = getSnapshot() + return snapshot + end + + if options.preserveHistory then + -- Preserve a history of every state change to allow for more granular + -- syncing with the client. + local snapshots: { { [string]: any } } = {} + local changed = false + + function self:connect(callback) + local subscriptions = {} + + sync = callback + + local function pushSnapshot(key: string, current: any, previous: any) + local lastSnapshot = snapshots[#snapshots] + local previousSnapshot = snapshots[#snapshots - 1] + + -- Optimize snapshots by updating the most recent snapshot if the + -- previous and current values are the same, since this allows us + -- to group multiple changes into a single snapshot. + if previousSnapshot and previousSnapshot[key] == previous and lastSnapshot[key] == previous then + lastSnapshot[key] = current + else + local nextSnapshot = table.clone(lastSnapshot) + nextSnapshot[key] = current + table.insert(snapshots, nextSnapshot) + end + end + + -- Populate the snapshot with the initial state of each atom. + -- Subscribe to each atom and update the state when it changes. + for key, atom in next, options.atoms do + subscriptions[key] = subscribe(atom, function(current, previous) + pushSnapshot(key, current, previous) + changed = true + end) + end + + snapshots[1] = createSnapshot() + + local disconnect = setInterval(function() + if not changed then + return + end + + local payloads: { SyncPayload } = {} + local lastSnapshot + + for index, snapshot in next, snapshots do + lastSnapshot = snapshot + + if index == 1 then + continue + end + + table.insert(payloads, { + type = "patch", + data = patch.diff(snapshots[index - 1], snapshot), + }) + end + + snapshots = { lastSnapshot } + changed = false + + for _, player in next, Players:GetPlayers() do + callback(player, unpack(payloads)) + end + end, options.interval or 0) + + return function() + disconnect() + + for _, unsubscribe in next, subscriptions do + unsubscribe() + end + end + end + + function self:_sendPatch(player: Player) + assert(sync, "connect() must be called before _sendPatch()") - local disconnect = setInterval(function() if not changed then return end local payloads: { SyncPayload } = {} - local lastSnapshot for index, snapshot in next, snapshots do - lastSnapshot = snapshot - if index == 1 then continue end @@ -135,19 +169,73 @@ local function server(options: ServerOptions): ServerSyncer }) end - snapshots = { lastSnapshot } + snapshots = { snapshots[#snapshots] } changed = false - for _, player in next, Players:GetPlayers() do - callback(player, unpack(payloads)) - end - end, options.interval or 0) + sync(player, unpack(payloads)) + end + else + -- Only send the most recent state to the client. + local previousSnapshot = createSnapshot() + local currentSnapshot = table.clone(previousSnapshot) + local changed = false - return function() - disconnect() - for _, cleanup in next, cleanups do - cleanup() + function self:connect(callback) + local subscriptions = {} + + sync = callback + + -- Subscribe to each atom and update the state when one changes. + for key, atom in next, options.atoms do + subscriptions[key] = subscribe(atom, function(state) + currentSnapshot[key] = state + changed = true + end) end + + local disconnect = setInterval(function() + if not changed then + return + end + + local payload: SyncPayload = { + type = "patch", + data = patch.diff(previousSnapshot, currentSnapshot), + } + + previousSnapshot = table.clone(currentSnapshot) + changed = false + + for _, player in next, Players:GetPlayers() do + callback(player, payload) + end + end, options.interval or 0) + + return function() + disconnect() + + for _, unsubscribe in next, subscriptions do + unsubscribe() + end + end + end + + function self:_sendPatch(player: Player) + assert(sync, "connect() must be called before _sendPatch()") + + if not changed then + return + end + + local payload: SyncPayload = { + type = "patch", + data = patch.diff(previousSnapshot, currentSnapshot), + } + + previousSnapshot = table.clone(currentSnapshot) + changed = false + + sync(player, payload) end end @@ -157,36 +245,10 @@ local function server(options: ServerOptions): ServerSyncer sync(player, { type = "init", - data = snapshots[#snapshots], + data = createSnapshot(), }) end - function self:_sendPatch(player: Player) - assert(sync, "connect() must be called before _sendPatch()") - - if not changed then - return - end - - local payloads: { SyncPayload } = {} - - for index, snapshot in next, snapshots do - if index == 1 then - continue - end - - table.insert(payloads, { - type = "patch", - data = patch.diff(snapshots[index - 1], snapshot), - }) - end - - snapshots = { snapshots[#snapshots] } - changed = false - - sync(player, unpack(payloads)) - end - return self end