diff --git a/docs/_AI_INDEX.md b/docs/_AI_INDEX.md index 0b48eac90d2..6e5fbddaf12 100644 --- a/docs/_AI_INDEX.md +++ b/docs/_AI_INDEX.md @@ -40,7 +40,7 @@ This file is an index for AI agents. The `_` prefix keeps it out of Docusaurus. | [cli.md](cli.md) | `nevermore` CLI command reference: every command and flag (`init`, `install`, `login`, `test`, `deploy`, `batch`, `tools`), global options, command tree | | [deploy.md](deploy.md) | `nevermore deploy`: login, `deploy init`, `deploy run`, config schema, flag reference, common workflows | | [architecture/](architecture/index.md) | Architecture: workspace layout, design philosophy, ServiceBag, dependency injection | -| [architecture/patterns.md](architecture/patterns.md) | Core patterns: Maid, BaseObject, Binder, Rx, Brio, Blend, AdorneeData, TieDefinition; Brio pipeline pitfalls | +| [architecture/patterns.md](architecture/patterns.md) | Core patterns: Maid, BaseObject, Binder, Rx, Brio, Blend, AdorneeData, TieDefinition, Remoting observables; Brio pipeline pitfalls | | [build.md](build.md) | Contributing: local setup, tools, versioning, custom Rojo | | [testing/](testing/index.md) | Testing: Jest3, deploy config, CLI commands, credentials, CI | | [testing/integration-testing.md](testing/integration-testing.md) | Integration testing: full-game tests, base place merging, deploy pipeline | diff --git a/docs/architecture/patterns.md b/docs/architecture/patterns.md index 5f60c2e3ff5..fa8e0a8a54a 100644 --- a/docs/architecture/patterns.md +++ b/docs/architecture/patterns.md @@ -37,6 +37,27 @@ The three adders return different things, which matters when you want to keep us **When to use:** Any time you create connections, spawn threads, or instantiate objects that need cleanup. Almost every class uses one. +### Register with the parent maid before attaching `Finally` + +A common shape is a short-lived maid, scoped to one async operation, parked on a long-lived one: + +```lua +self._maid[opMaid] = opMaid +opMaid:GiveTask(function() + self._maid[opMaid] = nil +end) + +promise:Finally(function() + self._maid[opMaid] = nil -- must come after the registration above +end) +``` + +Written the other way round it leaks, silently and only sometimes. Our promises settle *synchronously* when the work is already done — an `:Then` on a resolved promise runs inline — so a `Finally` attached first fires immediately, removes an entry that isn't there yet, and then the registration lands with nothing left to clear it. The entry survives for the lifetime of the owner. It looks correct in review and behaves correctly whenever the operation happens to be genuinely async, which is what makes it easy to miss. + +`Maid:GivePromise` already does this correctly and is worth reading as the reference — it registers before attaching `Finally`, and early-outs entirely when the promise is already settled. Reach for it when the thing you're tracking *is* the promise. Hand-roll only when it isn't (tracking a sub-maid that owns the intermediate work, say), and then copy the ordering exactly. + +The invariant is worth asserting directly: bind, release, and check the owner's task count is back where it started. Repeating the cycle ten times turns a slow leak into an obvious one. + ## BaseObject A lightweight base class that gives you a `_maid` and optional `_obj` reference for free. Nearly all Nevermore classes inherit from it. @@ -275,6 +296,36 @@ end **When to use:** When you need optional or pluggable interfaces — particularly across client/server boundaries or for plugin systems where the implementer shouldn't need to know about the consumer. +## Remoting observables + +[Remoting](/api/Remoting) wraps RemoteEvents and RemoteFunctions so a service can declare its network surface as members rather than hand-managed Instances. Alongside the event and method members it can carry observable streams: the server binds a factory, the client subscribes, and values flow for as long as the client stays subscribed. + +```lua +-- Server +remoting:BindObservable("Health", function(player, entityId) + return observeHealth(player, entityId) +end) + +-- Client +maid:GiveTask(remoting:Observe("Health", entityId):Subscribe(print)) + +-- Or through member syntax, which reads better at the call site +remoting.Health:BindObservable(function(player, entityId) ... end) +maid:GiveTask(remoting.Health:Observe(entityId):Subscribe(print)) +``` + +The factory runs once per subscription, so it can vary the stream by player and by the arguments the client passed. The server tears the stream down when the client unsubscribes, when the source completes or fails, when the player leaves, or when the remoting is destroyed — the client's subscription completes in that last case rather than hanging. + +Under the hood each observable member reserves one extra remote event named `__Observe`. A RemoteEvent is full duplex, so that single instance carries subscribe and unsubscribe up and emissions down. + +**Two ways this differs from a local observable:** + +**It cannot emit synchronously on subscribe.** Most Nevermore observables fire an initial value during `:Subscribe()`; this one can't, because the first value is at least a round trip away. Anything that assumes a synchronous first emission — a Blend binding, a `Rx.combineLatest` that mixes local and remote sources — will sit empty until the value lands. Give it a starting value with `Rx.defaultsTo` when the consumer can't tolerate that gap. + +**It is cold, and each subscription costs a stream.** Two `:Subscribe()` calls on the same observable open two server-side subscriptions and produce two streams of packets. That is correct Rx semantics but a real network bill, so pipe through `Rx.share()` when several consumers want the same values. + +**When to use:** When the client needs a live view of server state rather than a one-shot answer. For a single value, `PromiseInvokeServer` is cheaper and simpler. + ## How the patterns fit together These patterns compose naturally: @@ -286,3 +337,4 @@ These patterns compose naturally: 5. **Blend + Rx** — Blend properties accept observables directly, making UI reactive. 6. **ServiceBag + Binder** — Services create and manage binders; binders receive ServiceBag for dependency injection. 7. **AdorneeData + Binder** — Binder creates a class per tagged Instance; AdorneeData reads/observes configuration attributes on that Instance. +8. **Remoting + Rx** — Remoting carries observables across the client/server boundary, so a server-side Rx pipeline can drive client UI directly. diff --git a/docs/conventions/luau.md b/docs/conventions/luau.md index 330e20330e6..4a46aeed748 100644 --- a/docs/conventions/luau.md +++ b/docs/conventions/luau.md @@ -151,3 +151,11 @@ These are the types you'll use most often: - `Signal.new() :: any` — when the signal type would be too complex to annotate inline **Prefer fixing upstream types** over casting. If a type is wrong, fix it in the source package. + +## Type system limits worth knowing + +Two limits show up often enough to cost real time if you rediscover them. + +**Singleton types are strings and booleans only.** `"server" :: "server"` is how the enum packages pin a literal, so `OPCODE_FIRE = 3 :: 3` looks like it should work too. It doesn't — numeric singletons don't exist, and the failure is a *parse* error rather than a type error, so stylua and selene report a cascade of confusing syntax complaints starting at the enclosing call rather than the offending line. Write the number plain. + +**A class stored behind a table index type loses its generic methods.** Our classes are `typeof(setmetatable(...))` types whose methods are generic (`Maid.Add` is `(self, T) -> T`). Put one in a field or index type — `_subscriptions: { [string]: Maid.Maid }` — and calling a method on a value read back out fails with the memorable `Expected this to be 'Maid', but got 'Maid'`, or refuses `==` between two of them because "they do not have the same metatable". Type the stored value `any` and note why; the annotation was never buying much, since it stops checking at exactly the point you'd use it. diff --git a/docs/gotchas/troubleshooting.md b/docs/gotchas/troubleshooting.md index 4cdf497e0ce..55a468ee243 100644 --- a/docs/gotchas/troubleshooting.md +++ b/docs/gotchas/troubleshooting.md @@ -82,6 +82,34 @@ instrumentation that still times out tells you nothing. Two distinct causes to r step detached (`task.spawn`) so the test body returns, and have a later test in the same suite sample and print the detached thread's `coroutine.status`/`debug.traceback`. +### A dual-realm spec hangs on a tie/binder that resolves fine on the other realm + +Symptom: one realm's `TieDefinition:Promise(...)` never resolves (jest reports a 5s timeout) while the +identical call on the other realm passes, and the only warning is a downstream "failed to find X for +player". The cause is usually **not** the package under test — it's a stale `node_modules`. + +`node_modules` is untracked, so `git stash -u` and a clean checkout both leave it stale; the failure +looks pre-existing and reproducible when it is purely local. When a workspace dependency was added to +some transitive package's `package.json` since your last install, the symlink is missing and the +loader throws `[Loader] - "SomeModule" is not available` from inside that package's `Init`. ServiceBag +does not abort the boot on a failed service `Init`, so the bag comes up **half-registered**: services +the aborted `Init` had not reached yet are never added. Anything that later calls +`serviceBag:GetService(...)` for one of them (a binder constructor, for instance) throws +`Cannot initialize service "..." after start`, the bind silently fails, and the tie for that realm +never gets an implementation. + +Fix: run `pnpm install` from the repo root. To confirm before you start editing source, run the boot +with `--logs` and read the *first* error in the log, not the last — the loader failure appears many +lines above the symptom. + +### `--script-text` only runs the first line on Windows + +Multi-line strings passed to `nevermore test --cloud --script-text` are truncated at the first newline +when the CLI is invoked through PowerShell or Git Bash, and the run reports `Tests passed!` having +executed one line. Join the script into a single line before passing it +(`(Get-Content diag.lua) -join ' '`), and drop any `--` comments and the `--!strict`/`--!nonstrict` +header first — the header is otherwise parsed as a CLI flag. + ### Test fails with `loader is not a valid member of ModuleScript "..."` The standard file header `require(script.Parent.loader).load(script)` deliberately does not diff --git a/docs/testing/testing.md b/docs/testing/testing.md index a6d3f84b71f..056a4e46a23 100644 --- a/docs/testing/testing.md +++ b/docs/testing/testing.md @@ -284,6 +284,28 @@ Known limits — these are gaps in `player-mock`'s contract to fix there, not pa - Per-player data with no lookup domain yet: add a domain to `LOOKUPS` rather than stubbing per call site. +### Both realms share a thread, so replication no longer hides creation races + +In a live game a server-created remote reaches the client a network step after the server finished +wiring it up. Dummy mode collapses that: the server's `Instance.new` and the client's reaction to it +run on one thread, so a client watching for a remote to appear can act on it *while the server is +still mid-setup* — after the instance was parented, before the handler was attached. A message sent +in that window goes nowhere and is never retried, which reads as a stream that silently never starts. + +The fix is to make the instance discoverable only once it is fully wired: create it detached, attach +the handler, then parent it. `Remoting._getOrCreateRemoteEvent` and `_getOrCreateRemoteFunction` take +an `attachHandler` callback for exactly this, so `Connect` and `Bind` are live the instant anything +can see them. Parenting last is worth doing on reflex whenever another realm — or another observer — +keys off an instance appearing. + +Resist the urge to paper over it with `task.defer` on the reacting side. It passes the test for the +same reason production passes without it, which is precisely why it hides the defect rather than +fixing it: the window is still there, just harder to hit. + +The general shape: when a spec covers one realm reacting to another realm creating an Instance, ask +whether the reaction can observe a half-built state. If it can, the bug is real — production just has +a network hop papering over it — and the ordering fix belongs in the package, not the spec. + ### jest.config.lua Every testable package needs a `jest.config.lua` in its `src/` directory. This tells the test runner to discover `.spec` files: diff --git a/src/remoting/README.md b/src/remoting/README.md index f80ab0e681d..3f47b7d559a 100644 --- a/src/remoting/README.md +++ b/src/remoting/README.md @@ -23,4 +23,5 @@ npm install @quenty/remoting --save ## Features * Promise implementation available +* Observable streams across the client/server boundary via `BindObservable`/`Observe` * Always safe-to-use remotes if acquired \ No newline at end of file diff --git a/src/remoting/src/Shared/Interface/Remoting.Lifetime.spec.lua b/src/remoting/src/Shared/Interface/Remoting.Lifetime.spec.lua new file mode 100644 index 00000000000..73fbce90f1b --- /dev/null +++ b/src/remoting/src/Shared/Interface/Remoting.Lifetime.spec.lua @@ -0,0 +1,538 @@ +--!nonstrict +--[[ + @class Remoting.Lifetime.spec.lua +]] + +local require = require(script.Parent.loader).load(script) + +local Workspace = game:GetService("Workspace") + +local Jest = require("Jest") +local Observable = require("Observable") +local PlayerMock = require("PlayerMock") +local PromiseTestUtils = require("PromiseTestUtils") +local Remoting = require("Remoting") + +local describe = Jest.Globals.describe +local expect = Jest.Globals.expect +local it = Jest.Globals.it + +local function countTasks(maid) + local count = 0 + for _ in maid._tasks do + count += 1 + end + + return count +end + +local function setup() + local instance = Instance.new("Folder") + + local server = Remoting.Server.new(instance, "Lifetime") + server._useDummyObject = true + + local client = Remoting.Client.new(instance, "Lifetime") + client._useDummyObject = true + + local playerMock = PlayerMock.new({ UserId = 12345 }) + playerMock.Parent = Workspace + PlayerMock.setMockedLocalPlayer(playerMock) + + local serverDestroyed = false + local clientDestroyed = false + + local controller = {} + + controller.instance = instance + controller.server = server + controller.client = client + controller.player = playerMock + + function controller.destroyServer() + serverDestroyed = true + server:Destroy() + end + + function controller.destroyClient() + clientDestroyed = true + client:Destroy() + end + + function controller.serverTaskCount() + return countTasks(server._maid) + end + + function controller.clientTaskCount() + return countTasks(client._maid) + end + + function controller.container() + return instance:FindFirstChildOfClass("Configuration") + end + + function controller.settle(promise) + return PromiseTestUtils.awaitSettled(promise) + end + + function controller.destroy() + if not clientDestroyed then + client:Destroy() + end + if not serverDestroyed then + server:Destroy() + end + PlayerMock.setMockedLocalPlayer(nil) + playerMock:Destroy() + instance:Destroy() + end + + return controller +end + +describe("Remoting.Connect lifetime", function() + it("registers a task on the remoting and releases it when the connection is cleaned", function() + local controller = setup() + + controller.server:Connect("Ping", function() end):DoCleaning() + + local before = controller.serverTaskCount() + local connectMaid = controller.server:Connect("Ping", function() end) + + expect(controller.serverTaskCount()).toEqual(before + 1) + + connectMaid:DoCleaning() + + expect(controller.serverTaskCount()).toEqual(before) + + controller.destroy() + end) + + it("stops delivering to the callback once the connection is cleaned", function() + local controller = setup() + + local calls = 0 + local connectMaid = controller.server:Connect("Ping", function() + calls += 1 + end) + + controller.client:FireServer("Ping") + expect(PromiseTestUtils.awaitValue(function() + return calls == 1 + end)).toEqual(true) + + connectMaid:DoCleaning() + controller.client:FireServer("Ping") + + expect(PromiseTestUtils.awaitValue(function() + return calls == 1 + end)).toEqual(true) + + controller.destroy() + end) + + it("does not accumulate tasks across repeated connect and release cycles", function() + local controller = setup() + + controller.server:Connect("Ping", function() end):DoCleaning() + + local before = controller.serverTaskCount() + for _ = 1, 10 do + local connectMaid = controller.server:Connect("Ping", function() end) + connectMaid:DoCleaning() + end + + expect(controller.serverTaskCount()).toEqual(before) + + controller.destroy() + end) + + it("tolerates cleaning the same connection twice", function() + local controller = setup() + + controller.server:Connect("Ping", function() end):DoCleaning() + + local before = controller.serverTaskCount() + local connectMaid = controller.server:Connect("Ping", function() end) + connectMaid:DoCleaning() + connectMaid:DoCleaning() + + expect(controller.serverTaskCount()).toEqual(before) + + controller.destroy() + end) +end) + +describe("Remoting.Bind lifetime", function() + it("registers a task on the remoting and releases it when the bind is cleaned", function() + local controller = setup() + + controller.server + :Bind("Ask", function() + return true + end) + :DoCleaning() + + local before = controller.serverTaskCount() + local bindMaid = controller.server:Bind("Ask", function() + return true + end) + + expect(controller.serverTaskCount()).toEqual(before + 1) + + bindMaid:DoCleaning() + + expect(controller.serverTaskCount()).toEqual(before) + + controller.destroy() + end) + + it("marks the translated callback dead once the bind is released", function() + local controller = setup() + + local bindMaid = controller.server:Bind("Ask", function() + return true + end) + + expect(controller.settle(controller.client:PromiseInvokeServer("Ask"))).toEqual(true) + + -- Asserted against the translated callback rather than through an invoke: routing the + -- disconnect error through a BindableFunction gets it logged by the engine even when + -- the caller pcalls it, which the runner counts as a failed run + local translated = controller.server:_translateCallback(bindMaid, "Ask", function() + return true + end) + + expect(translated()).toEqual(true) + + bindMaid:DoCleaning() + + expect(function() + translated() + end).toThrow() + + controller.destroy() + end) + + it("allows the member to be bound again after release", function() + local controller = setup() + + local bindMaid = controller.server:Bind("Ask", function() + return "first" + end) + bindMaid:DoCleaning() + + controller.server:Bind("Ask", function() + return "second" + end) + + local promise = controller.client:PromiseInvokeServer("Ask") + expect(controller.settle(promise)).toEqual(true) + + local _isFulfilled, value = promise:Yield() + expect(value).toEqual("second") + + controller.destroy() + end) +end) + +describe("Remoting promise lifetime", function() + it("releases the fire task once PromiseFireServer settles", function() + local controller = setup() + + controller.server:Connect("Ping", function() end) + + local before = controller.clientTaskCount() + local promise = controller.client:PromiseFireServer("Ping") + + expect(controller.settle(promise)).toEqual(true) + expect(controller.clientTaskCount()).toEqual(before) + + controller.destroy() + end) + + it("releases the invoke task once PromiseInvokeServer settles", function() + local controller = setup() + + controller.server:Bind("Ask", function() + return true + end) + + local before = controller.clientTaskCount() + local promise = controller.client:PromiseInvokeServer("Ask") + + expect(controller.settle(promise)).toEqual(true) + expect(controller.clientTaskCount()).toEqual(before) + + controller.destroy() + end) + + it("releases the invoke task once PromiseInvokeClient settles", function() + local controller = setup() + + controller.client:Bind("AskClient", function() + return true + end) + + expect(controller.settle(controller.server:PromiseInvokeClient("AskClient", controller.player))).toEqual(true) + + local before = controller.serverTaskCount() + local promise = controller.server:PromiseInvokeClient("AskClient", controller.player) + + expect(controller.settle(promise)).toEqual(true) + expect(controller.serverTaskCount()).toEqual(before) + + controller.destroy() + end) + + it("does not accumulate tasks across repeated invokes", function() + local controller = setup() + + controller.server:Bind("Ask", function() + return true + end) + + local before = controller.clientTaskCount() + for _ = 1, 10 do + expect(controller.settle(controller.client:PromiseInvokeServer("Ask"))).toEqual(true) + end + + expect(controller.clientTaskCount()).toEqual(before) + + controller.destroy() + end) +end) + +describe("Remoting.BindObservable lifetime", function() + it("registers a task on the remoting and releases it when the bind is cleaned", function() + local controller = setup() + + controller.server + :BindObservable("Stream", function() + return Observable.new(function() + return nil + end) + end) + :DoCleaning() + + local before = controller.serverTaskCount() + local bindMaid = controller.server:BindObservable("Stream", function() + return Observable.new(function() + return nil + end) + end) + + expect(controller.serverTaskCount() > before).toEqual(true) + + bindMaid:DoCleaning() + + expect(controller.serverTaskCount()).toEqual(before) + + controller.destroy() + end) + + it("allows the member to be bound again after release", function() + local controller = setup() + + local bindMaid = controller.server:BindObservable("Stream", function() + return Observable.new(function(sub) + sub:Fire("first") + return nil + end) + end) + bindMaid:DoCleaning() + + controller.server:BindObservable("Stream", function() + return Observable.new(function(sub) + sub:Fire("second") + return nil + end) + end) + + local received = {} + local subscription = controller.client:Observe("Stream"):Subscribe(function(value) + table.insert(received, value) + end) + + expect(PromiseTestUtils.awaitValue(function() + return #received >= 1 + end)).toEqual(true) + expect(received[#received]).toEqual("second") + + subscription:Destroy() + controller.destroy() + end) + + it("keeps one relay per member across repeated observes", function() + local controller = setup() + + controller.server:BindObservable("Stream", function() + return Observable.new(function() + return nil + end) + end) + + local first = controller.client:Observe("Stream") + local relay = controller.client._observableRelays["Stream"] + local second = controller.client:Observe("Stream") + + expect(controller.client._observableRelays["Stream"]).toBe(relay) + expect(first).never.toBe(second) + + controller.destroy() + end) + + it("forgets a client subscription once it is unsubscribed", function() + local controller = setup() + + controller.server:BindObservable("Stream", function() + return Observable.new(function(sub) + sub:Fire(1) + return nil + end) + end) + + local received = {} + local subscription = controller.client:Observe("Stream"):Subscribe(function(value) + table.insert(received, value) + end) + + expect(PromiseTestUtils.awaitValue(function() + return #received >= 1 + end)).toEqual(true) + + subscription:Destroy() + + local relay = controller.client._observableRelays["Stream"] + expect(next(relay._subscriptions)).toEqual(nil) + + controller.destroy() + end) +end) + +describe("Remoting.Destroy lifetime", function() + it("removes every instance it created from the data model", function() + local controller = setup() + + controller.server:Connect("Ping", function() end) + controller.server:Bind("Ask", function() + return true + end) + + expect(controller.container()).never.toEqual(nil) + + controller.destroyServer() + + expect(controller.container()).toEqual(nil) + + controller.destroy() + end) + + it("runs the cleanup of a connection that is still live", function() + local controller = setup() + + local cleaned = false + local connectMaid = controller.server:Connect("Ping", function() end) + connectMaid:GiveTask(function() + cleaned = true + end) + + controller.destroyServer() + + expect(cleaned).toEqual(true) + + controller.destroy() + end) + + it("runs the cleanup of a bind that is still live", function() + local controller = setup() + + local cleaned = false + local bindMaid = controller.server:Bind("Ask", function() + return true + end) + bindMaid:GiveTask(function() + cleaned = true + end) + + controller.destroyServer() + + expect(cleaned).toEqual(true) + + controller.destroy() + end) + + it("tears down the source of a live observable subscription", function() + local controller = setup() + + local cleaned = false + controller.server:BindObservable("Stream", function() + return Observable.new(function(sub) + sub:Fire(1) + return function() + cleaned = true + end + end) + end) + + local received = {} + local subscription = controller.client:Observe("Stream"):Subscribe(function(value) + table.insert(received, value) + end) + + expect(PromiseTestUtils.awaitValue(function() + return #received >= 1 + end)).toEqual(true) + + controller.destroyServer() + + expect(cleaned).toEqual(true) + + subscription:Destroy() + controller.destroy() + end) + + it("completes a live client subscription when the client is destroyed", function() + local controller = setup() + + controller.server:BindObservable("Stream", function() + return Observable.new(function(sub) + sub:Fire(1) + return nil + end) + end) + + local completed = false + local subscription = controller.client:Observe("Stream"):Subscribe(nil, nil, function() + completed = true + end) + + controller.destroyClient() + + expect(completed).toEqual(true) + + subscription:Destroy() + controller.destroy() + end) + + it("empties the remoting maid", function() + local controller = setup() + + controller.server:Connect("Ping", function() end) + controller.server:Bind("Ask", function() + return true + end) + controller.server:BindObservable("Stream", function() + return Observable.new(function() + return nil + end) + end) + + local maid = controller.server._maid + controller.destroyServer() + + expect(countTasks(maid)).toEqual(0) + + controller.destroy() + end) +end) diff --git a/src/remoting/src/Shared/Interface/Remoting.lua b/src/remoting/src/Shared/Interface/Remoting.lua index c25a04730cb..f8c9e56047c 100644 --- a/src/remoting/src/Shared/Interface/Remoting.lua +++ b/src/remoting/src/Shared/Interface/Remoting.lua @@ -19,6 +19,8 @@ local Promise = require("Promise") local PromiseUtils = require("PromiseUtils") local RemoteFunctionUtils = require("RemoteFunctionUtils") local RemotingMember = require("RemotingMember") +local RemotingObservableClientRelay = require("RemotingObservableClientRelay") +local RemotingObservableServerRelay = require("RemotingObservableServerRelay") local RemotingRealmUtils = require("RemotingRealmUtils") local RemotingRealms = require("RemotingRealms") local RxBrioUtils = require("RxBrioUtils") @@ -65,12 +67,20 @@ export type Remoting = typeof(setmetatable( _remotingRealm: RemotingRealms.RemotingRealm, _useDummyObject: boolean, _remoteFolderName: string, + _observableRelays: { [string]: any }, + _boundMemberKinds: { [string]: string }, -- Public methods DeclareEvent: (self: Remoting, memberName: string) -> (), DeclareMethod: (self: Remoting, memberName: string) -> (), Connect: (self: Remoting, memberName: string, callback: (...any) -> ()) -> Maid.Maid, Bind: (self: Remoting, memberName: string, callback: (...any) -> ()) -> Maid.Maid, + BindObservable: ( + self: Remoting, + memberName: string, + factory: RemotingObservableServerRelay.ObservableFactory + ) -> Maid.Maid, + Observe: (self: Remoting, memberName: string, ...any) -> Observable.Observable<...any>, FireClient: (self: Remoting, memberName: string, player: Player, ...any) -> (), FireAllClients: (self: Remoting, memberName: string, ...any) -> (), FireAllClientsExcept: (self: Remoting, memberName: string, excludePlayer: Player, ...any) -> (), @@ -97,10 +107,23 @@ export type Remoting = typeof(setmetatable( self: Remoting, memberName: string ) -> Observable.Observable>, + _observeUpstreamRemoteEventBrio: ( + self: Remoting, + memberName: string + ) -> Observable.Observable>, + _fireUpstreamRemoteEvent: (self: Remoting, remoteEvent: RemoteEvent | BindableEvent, ...any) -> (), _promiseContainer: (self: Remoting, maid: Maid.Maid) -> Promise.Promise, _promiseRemoteEvent: (self: Remoting, maid: Maid.Maid, memberName: string) -> Promise.Promise, - _getOrCreateRemoteEvent: (self: Remoting, memberName: string) -> RemoteEvent | BindableEvent, - _getOrCreateRemoteFunction: (self: Remoting, memberName: string) -> RemoteFunction | BindableFunction, + _getOrCreateRemoteEvent: ( + self: Remoting, + memberName: string, + attachHandler: ((RemoteEvent | BindableEvent) -> ())? + ) -> RemoteEvent | BindableEvent, + _getOrCreateRemoteFunction: ( + self: Remoting, + memberName: string, + attachHandler: ((RemoteFunction | BindableFunction) -> ())? + ) -> RemoteFunction | BindableFunction, _promiseRemoteFunction: ( self: Remoting, maid: Maid.Maid, @@ -143,6 +166,8 @@ function Remoting.new(instance: Instance, name: string, remotingRealm: RemotingR self._remoteFolderName = string.format("%sRemotes", self._name) self._remoteObjects = {} + self._observableRelays = {} + self._boundMemberKinds = {} return self end @@ -172,15 +197,14 @@ function Remoting.Connect(self: Remoting, memberName: string, callback: (...any) if self._remotingRealm == RemotingRealms.SERVER then if self._useDummyObject then - self:DeclareEvent(memberName) - self:_getOrCreateRemoteEvent(self:_getDummyMemberName(memberName, "OnClientEvent")) - local bindableEvent: BindableEvent = - self:_getOrCreateRemoteEvent(self:_getDummyMemberName(memberName, "OnServerEvent")) :: any - connectMaid:GiveTask(bindableEvent.Event:Connect(callback)) + self:_getOrCreateRemoteEvent(self:_getDummyMemberName(memberName, "OnServerEvent"), function(event) + connectMaid:GiveTask((event :: BindableEvent).Event:Connect(callback)) + end) else - local remoteEvent: RemoteEvent = self:_getOrCreateRemoteEvent(memberName) :: any - connectMaid:GiveTask(remoteEvent.OnServerEvent:Connect(callback)) + self:_getOrCreateRemoteEvent(memberName, function(event) + connectMaid:GiveTask((event :: RemoteEvent).OnServerEvent:Connect(callback)) + end) end -- TODO: Cleanup if nothing else is expecting this @@ -242,19 +266,25 @@ end function Remoting.Bind(self: Remoting, memberName: string, callback: (...any) -> ...any): Maid.Maid assert(type(memberName) == "string", "Bad memberName") assert(type(callback) == "function", "Bad callback") + assert( + self._boundMemberKinds[memberName] ~= "Observable", + string.format("[Remoting.Bind] - %q is already bound as an observable", self:_getDebugMemberName(memberName)) + ) + + self._boundMemberKinds[memberName] = "Method" local bindMaid: Maid.Maid = Maid.new() if self._remotingRealm == RemotingRealms.SERVER then if self._useDummyObject then - self:DeclareMethod(memberName) - - local bindableFunction: BindableFunction = - self:_getOrCreateRemoteFunction(self:_getDummyMemberName(memberName, "OnServerInvoke")) :: any - bindableFunction.OnInvoke = self:_translateCallback(bindMaid, memberName, callback) + self:_getOrCreateRemoteFunction(self:_getDummyMemberName(memberName, "OnServerInvoke"), function(func) + (func :: BindableFunction).OnInvoke = self:_translateCallback(bindMaid, memberName, callback) + end) + self:_getOrCreateRemoteFunction(self:_getDummyMemberName(memberName, "OnClientInvoke")) else - local remoteFunction: RemoteFunction = self:_getOrCreateRemoteFunction(memberName) :: any - remoteFunction.OnServerInvoke = self:_translateCallback(bindMaid, memberName, callback) + self:_getOrCreateRemoteFunction(memberName, function(func) + (func :: RemoteFunction).OnServerInvoke = self:_translateCallback(bindMaid, memberName, callback) + end) end -- TODO: Cleanup if nothing else is expecting this @@ -308,6 +338,104 @@ function Remoting.Bind(self: Remoting, memberName: string, callback: (...any) -> return bindMaid end +--[=[ + Binds an observable factory to the member. The factory is invoked once per client + subscription, so it can vary the stream by player and by the arguments the client + passed to [Remoting.Observe]. + + ```lua + remoting:BindObservable("Health", function(player, entityId) + return observeHealth(player, entityId) + end) + ``` + + The stream lives for as long as the client stays subscribed. It is torn down when the + client unsubscribes, when the source completes or fails, when the player leaves, or + when this remoting is destroyed. + + @server + @param memberName string + @param factory (player: Player, ...any) -> Observable + @return MaidTask +]=] +function Remoting.BindObservable( + self: Remoting, + memberName: string, + factory: RemotingObservableServerRelay.ObservableFactory +): Maid.Maid + assert(type(memberName) == "string", "Bad memberName") + assert(type(factory) == "function", "Bad factory") + assert(self._remotingRealm == RemotingRealms.SERVER, "BindObservable must be called on server") + assert( + self._boundMemberKinds[memberName] ~= "Method", + string.format( + "[Remoting.BindObservable] - %q is already bound as a method", + self:_getDebugMemberName(memberName) + ) + ) + assert( + not self._observableRelays[memberName], + string.format( + "[Remoting.BindObservable] - %q is already bound as an observable", + self:_getDebugMemberName(memberName) + ) + ) + + self._boundMemberKinds[memberName] = "Observable" + + local relay = RemotingObservableServerRelay.new(self, memberName, factory) + self._observableRelays[memberName] = relay + + local bindMaid = Maid.new() + bindMaid:GiveTask(function() + -- Destroy already drained the table, so it owns the teardown in that case + if self._observableRelays[memberName] == relay then + self._observableRelays[memberName] = nil + self._boundMemberKinds[memberName] = nil + relay:Destroy() + end + end) + + self._maid[bindMaid] = bindMaid + bindMaid:GiveTask(function() + self._maid[bindMaid] = nil + end) + + return bindMaid +end + +--[=[ + Observes the member bound on the server with [Remoting.BindObservable]. + + ```lua + maid:GiveTask(remoting:Observe("Health", entityId):Subscribe(print)) + ``` + + The observable is cold. Every subscription opens its own stream on the server, so pipe + it through [Rx.share] if more than one consumer needs the same values. + + Unlike most observables in Nevermore this cannot emit synchronously on subscribe. The + first value is at least one round trip away, and combinators like [Rx.combineLatest] + will not emit until it lands. + + @client + @param memberName string + @param ... any + @return Observable<...any> +]=] +function Remoting.Observe(self: Remoting, memberName: string, ...): Observable.Observable<...any> + assert(type(memberName) == "string", "Bad memberName") + assert(self._remotingRealm == RemotingRealms.CLIENT, "Observe must be called on client") + + local relay = self._observableRelays[memberName] + if not relay then + relay = RemotingObservableClientRelay.new(self, memberName) + self._observableRelays[memberName] = relay + end + + return relay:Observe(...) +end + --[=[ Forward declares an event on the remoting object @@ -572,14 +700,18 @@ function Remoting.PromiseFireServer(self: Remoting, memberName: string, ...) end) end - promise:Finally(function() - self._maid[fireMaid] = nil - end) + -- Registered before Finally is attached: the promise settles synchronously whenever the + -- remote already exists, and a Finally attached first would fire before there was + -- anything to remove, stranding the maid for the lifetime of the remoting self._maid[fireMaid] = fireMaid fireMaid:GiveTask(function() self._maid[fireMaid] = nil end) + promise:Finally(function() + self._maid[fireMaid] = nil + end) + -- TODO: Warn if remote event doesn't exist return promise @@ -633,14 +765,16 @@ function Remoting.PromiseInvokeServer(self: Remoting, memberName: string, ...): end) end - promise:Finally(function() - self._maid[invokeMaid] = nil - end) + -- Registered before Finally is attached -- see PromiseFireServer self._maid[invokeMaid] = invokeMaid invokeMaid:GiveTask(function() self._maid[invokeMaid] = nil end) + promise:Finally(function() + self._maid[invokeMaid] = nil + end) + -- TODO: Warn if remote function doesn't exist return promise @@ -671,12 +805,13 @@ function Remoting.PromiseInvokeClient(self: Remoting, memberName: string, player promise = invokeMaid:GivePromise(RemoteFunctionUtils.promiseInvokeClient(remoteFunction, player, ...)) end - promise:Finally(function() + -- Registered before Finally is attached -- see PromiseFireServer + self._maid[invokeMaid] = invokeMaid + invokeMaid:GiveTask(function() self._maid[invokeMaid] = nil end) - self._maid[invokeMaid] = invokeMaid - invokeMaid:GiveTask(function() + promise:Finally(function() self._maid[invokeMaid] = nil end) @@ -737,6 +872,31 @@ function Remoting._observeRemoteEventBrio(self: Remoting, memberName: string) }) end +--[[ + Resolves the raw event the client sends relay messages over. Sends bypass + PromiseFireServer so ordering is preserved -- see RemotingObservableClientRelay. +]] +function Remoting._observeUpstreamRemoteEventBrio(self: Remoting, memberName: string) + assert(type(memberName) == "string", "Bad memberName") + assert(self._remotingRealm == RemotingRealms.CLIENT, "Upstream events are only observed on the client") + + if self._useDummyObject then + return self:_observeRemoteEventBrio(self:_getDummyMemberName(memberName, "OnServerEvent")) + else + return self:_observeRemoteEventBrio(memberName) + end +end + +function Remoting._fireUpstreamRemoteEvent(self: Remoting, remoteEvent: RemoteEvent | BindableEvent, ...) + if self._useDummyObject then + local bindableEvent = remoteEvent :: BindableEvent + bindableEvent:Fire(Players.LocalPlayer or PlayerMock.getMockedLocalPlayer(), ...) + else + local realRemoteEvent = remoteEvent :: RemoteEvent + realRemoteEvent:FireServer(...) + end +end + function Remoting._promiseContainer(self: Remoting, maid: Maid.Maid): Promise.Promise return maid:GivePromise(promiseChild(self._instance, self._remoteFolderName, 5)) end @@ -769,13 +929,29 @@ function Remoting._observeFolderBrio(self: Remoting): Observable.Observable ())? +): RemoteFunction | BindableFunction assert(type(memberName) == "string", "Bad memberName") local remoteFunctionName = self:_getMemberName(memberName, REMOTE_FUNCTION_SUFFIX) - if self._remoteObjects[remoteFunctionName] then - return self._remoteObjects[remoteFunctionName] :: any + local existing = self._remoteObjects[remoteFunctionName] + if existing then + if attachHandler then + attachHandler(existing :: any) + end + + return existing :: any end local container = self:_ensureContainer() @@ -789,21 +965,42 @@ function Remoting._getOrCreateRemoteFunction(self: Remoting, memberName: string) remoteFunction.Name = remoteFunctionName remoteFunction.Archivable = false - remoteFunction.Parent = container self._remoteObjects[remoteFunctionName] = remoteFunction :: any self._maid[remoteFunction] = remoteFunction + if attachHandler then + attachHandler(remoteFunction :: any) + end + + remoteFunction.Parent = container + return remoteFunction :: any end -function Remoting._getOrCreateRemoteEvent(self: Remoting, memberName: string): RemoteEvent | BindableEvent +--[[ + Creates the remote event if it does not exist yet, otherwise returns the existing one. + + attachHandler runs against the event before it is parented, so a listener watching for the + event to appear can never see one that exists but is not yet handling anything. Existing + events are already parented, so it runs immediately for them. +]] +function Remoting._getOrCreateRemoteEvent( + self: Remoting, + memberName: string, + attachHandler: ((RemoteEvent | BindableEvent) -> ())? +): RemoteEvent | BindableEvent assert(type(memberName) == "string", "Bad memberName") local remoteEventName = self:_getMemberName(memberName, REMOTE_EVENT_SUFFIX) - if self._remoteObjects[remoteEventName] then - return self._remoteObjects[remoteEventName] :: any + local existing = self._remoteObjects[remoteEventName] + if existing then + if attachHandler then + attachHandler(existing :: any) + end + + return existing :: any end local container = self:_ensureContainer() @@ -817,11 +1014,16 @@ function Remoting._getOrCreateRemoteEvent(self: Remoting, memberName: string): R remoteEvent.Name = remoteEventName remoteEvent.Archivable = false - remoteEvent.Parent = container self._maid[remoteEvent] = remoteEvent self._remoteObjects[remoteEventName] = remoteEvent :: any + if attachHandler then + attachHandler(remoteEvent :: any) + end + + remoteEvent.Parent = container + return remoteEvent :: any end @@ -843,6 +1045,14 @@ end Cleans up the remoting object ]=] function Remoting.Destroy(self: Remoting) + -- Drained before the maid so relays can end their streams while the remotes still exist + local relays = self._observableRelays + self._observableRelays = {} + + for _, relay in relays do + relay:Destroy() + end + self._maid:DoCleaning() setmetatable(self :: any, nil) end diff --git a/src/remoting/src/Shared/Interface/RemotingMember.lua b/src/remoting/src/Shared/Interface/RemotingMember.lua index 66a5f18eb49..b7c3278065d 100644 --- a/src/remoting/src/Shared/Interface/RemotingMember.lua +++ b/src/remoting/src/Shared/Interface/RemotingMember.lua @@ -9,8 +9,10 @@ local require = require(script.Parent.loader).load(script) local Maid = require("Maid") +local Observable = require("Observable") local PlayerMock = require("PlayerMock") local Promise = require("Promise") +local RemotingObservableServerRelay = require("RemotingObservableServerRelay") local RemotingRealms = require("RemotingRealms") local RemotingMember = {} @@ -63,6 +65,44 @@ function RemotingMember.Bind(self: RemotingMember, callback: (...any) -> ...any) return self._remoting:Bind(self._memberName, callback) end +--[=[ + Binds an observable factory to the member. + + See [Remoting.BindObservable]. + + @server + @param factory (player: Player, ...any) -> Observable + @return MaidTask +]=] +function RemotingMember.BindObservable( + self: RemotingMember, + factory: RemotingObservableServerRelay.ObservableFactory +): Maid.Maid + assert(type(factory) == "function", "Bad factory") + assert(self._remotingRealm == RemotingRealms.SERVER, "BindObservable must be called on server") + + return self._remoting:BindObservable(self._memberName, factory) +end + +--[=[ + Observes the member bound on the server with [RemotingMember.BindObservable]. + + ```lua + maid:GiveTask(remoting.Health:Observe(entityId):Subscribe(print)) + ``` + + See [Remoting.Observe]. + + @client + @param ... any + @return Observable<...any> +]=] +function RemotingMember.Observe(self: RemotingMember, ...): Observable.Observable<...any> + assert(self._remotingRealm == RemotingRealms.CLIENT, "Observe must be called on client") + + return self._remoting:Observe(self._memberName, ...) +end + --[=[ Connects to the equivalent of a RemoteEvent for this member. diff --git a/src/remoting/src/Shared/Observables/Remoting.Observable.spec.lua b/src/remoting/src/Shared/Observables/Remoting.Observable.spec.lua new file mode 100644 index 00000000000..aa57265e265 --- /dev/null +++ b/src/remoting/src/Shared/Observables/Remoting.Observable.spec.lua @@ -0,0 +1,430 @@ +--!nonstrict +--[[ + @class RemotingObservable.spec.lua +]] + +local require = require(script.Parent.loader).load(script) + +local Workspace = game:GetService("Workspace") + +local Jest = require("Jest") +local Observable = require("Observable") +local PlayerMock = require("PlayerMock") +local PromiseTestUtils = require("PromiseTestUtils") +local Remoting = require("Remoting") + +local describe = Jest.Globals.describe +local expect = Jest.Globals.expect +local it = Jest.Globals.it + +local function setup() + local instance = Instance.new("Folder") + + local server = Remoting.Server.new(instance, "ObservableTest") + server._useDummyObject = true + + local client = Remoting.Client.new(instance, "ObservableTest") + client._useDummyObject = true + + local playerMock = PlayerMock.new({ UserId = 12345 }) + playerMock.Parent = Workspace + PlayerMock.setMockedLocalPlayer(playerMock) + + local serverDestroyed = false + + local controller = {} + + controller.server = server + controller.client = client + controller.player = playerMock + + function controller.destroyServer() + serverDestroyed = true + server:Destroy() + end + + function controller.awaitCount(getCount, target) + return PromiseTestUtils.awaitValue(function() + return getCount() >= target + end) + end + + function controller.destroy() + client:Destroy() + if not serverDestroyed then + server:Destroy() + end + PlayerMock.setMockedLocalPlayer(nil) + playerMock:Destroy() + instance:Destroy() + end + + return controller +end + +describe("Remoting.BindObservable", function() + it("delivers emissions from the server source to the client", function() + local controller = setup() + + controller.server:BindObservable("Health", function() + return Observable.new(function(sub) + sub:Fire(10) + sub:Fire(20) + return nil + end) + end) + + local received = {} + local subscription = controller.client:Observe("Health"):Subscribe(function(value) + table.insert(received, value) + end) + + expect(controller.awaitCount(function() + return #received + end, 2)).toEqual(true) + expect(received).toEqual({ 10, 20 }) + + subscription:Destroy() + controller.destroy() + end) + + it("passes the requesting player and the client arguments to the factory", function() + local controller = setup() + + local receivedPlayer, receivedArgs + controller.server:BindObservable("Health", function(player, ...) + receivedPlayer = player + receivedArgs = table.pack(...) + + return Observable.new(function(sub) + sub:Fire(true) + return nil + end) + end) + + local received = {} + local subscription = controller.client:Observe("Health", "entity", 7):Subscribe(function(value) + table.insert(received, value) + end) + + expect(controller.awaitCount(function() + return #received + end, 1)).toEqual(true) + expect(receivedPlayer).toBe(controller.player) + expect(receivedArgs.n).toEqual(2) + expect(receivedArgs[1]).toEqual("entity") + expect(receivedArgs[2]).toEqual(7) + + subscription:Destroy() + controller.destroy() + end) + + it("subscribes once the member is bound after the client is already observing", function() + local controller = setup() + + local received = {} + local subscription = controller.client:Observe("Late"):Subscribe(function(value) + table.insert(received, value) + end) + + controller.server:BindObservable("Late", function() + return Observable.new(function(sub) + sub:Fire("ready") + return nil + end) + end) + + expect(controller.awaitCount(function() + return #received + end, 1)).toEqual(true) + expect(received).toEqual({ "ready" }) + + subscription:Destroy() + controller.destroy() + end) + + it("completes the client subscription when the server source completes", function() + local controller = setup() + + controller.server:BindObservable("Health", function() + return Observable.new(function(sub) + sub:Fire(1) + sub:Complete() + return nil + end) + end) + + local completed = false + local subscription = controller.client:Observe("Health"):Subscribe(nil, nil, function() + completed = true + end) + + expect(PromiseTestUtils.awaitValue(function() + return completed + end)).toEqual(true) + + subscription:Destroy() + controller.destroy() + end) + + it("fails the client subscription when the server source fails", function() + local controller = setup() + + controller.server:BindObservable("Health", function() + return Observable.new(function(sub) + sub:Fail("bad news") + return nil + end) + end) + + local failure + local subscription = controller.client:Observe("Health"):Subscribe(nil, function(err) + failure = err + end) + + expect(PromiseTestUtils.awaitValue(function() + return failure ~= nil + end)).toEqual(true) + expect(failure).toEqual("bad news") + + subscription:Destroy() + controller.destroy() + end) + + it("fails the client subscription when the factory does not return an observable", function() + local controller = setup() + + controller.server:BindObservable("Health", function() + return nil + end) + + local failure + local subscription = controller.client:Observe("Health"):Subscribe(nil, function(err) + failure = err + end) + + expect(PromiseTestUtils.awaitValue(function() + return failure ~= nil + end)).toEqual(true) + + subscription:Destroy() + controller.destroy() + end) + + it("tears down the server source when the client unsubscribes", function() + local controller = setup() + + local cleanedUp = false + controller.server:BindObservable("Health", function() + return Observable.new(function(sub) + sub:Fire(1) + return function() + cleanedUp = true + end + end) + end) + + local received = {} + local subscription = controller.client:Observe("Health"):Subscribe(function(value) + table.insert(received, value) + end) + + expect(controller.awaitCount(function() + return #received + end, 1)).toEqual(true) + + subscription:Destroy() + + expect(PromiseTestUtils.awaitValue(function() + return cleanedUp + end)).toEqual(true) + + controller.destroy() + end) + + it("opens a separate server stream for each subscription", function() + local controller = setup() + + local factoryCalls = 0 + controller.server:BindObservable("Health", function() + factoryCalls += 1 + + return Observable.new(function(sub) + sub:Fire(factoryCalls) + return nil + end) + end) + + local observable = controller.client:Observe("Health") + + local firstReceived = {} + local first = observable:Subscribe(function(value) + table.insert(firstReceived, value) + end) + + local secondReceived = {} + local second = observable:Subscribe(function(value) + table.insert(secondReceived, value) + end) + + expect(controller.awaitCount(function() + return #firstReceived + end, 1)).toEqual(true) + expect(controller.awaitCount(function() + return #secondReceived + end, 1)).toEqual(true) + expect(factoryCalls).toEqual(2) + + first:Destroy() + second:Destroy() + controller.destroy() + end) + + it("completes live client subscriptions when the server remoting is destroyed", function() + local controller = setup() + + controller.server:BindObservable("Health", function() + return Observable.new(function(sub) + sub:Fire(1) + return nil + end) + end) + + local received = {} + local completed = false + local subscription = controller.client:Observe("Health"):Subscribe( + function(value) + table.insert(received, value) + end, + nil, + function() + completed = true + end + ) + + expect(controller.awaitCount(function() + return #received + end, 1)).toEqual(true) + + controller.destroyServer() + + expect(PromiseTestUtils.awaitValue(function() + return completed + end)).toEqual(true) + + subscription:Destroy() + controller.destroy() + end) + + it("drops the server subscription when the bind is released", function() + local controller = setup() + + local cleanedUp = false + local bindMaid = controller.server:BindObservable("Health", function() + return Observable.new(function(sub) + sub:Fire(1) + return function() + cleanedUp = true + end + end) + end) + + local received = {} + local subscription = controller.client:Observe("Health"):Subscribe(function(value) + table.insert(received, value) + end) + + expect(controller.awaitCount(function() + return #received + end, 1)).toEqual(true) + + bindMaid:DoCleaning() + + expect(PromiseTestUtils.awaitValue(function() + return cleanedUp + end)).toEqual(true) + + subscription:Destroy() + controller.destroy() + end) +end) + +describe("Remoting observable guards", function() + it("rejects Observe on the server", function() + local controller = setup() + + expect(function() + controller.server:Observe("Health") + end).toThrow() + + controller.destroy() + end) + + it("rejects BindObservable on the client", function() + local controller = setup() + + expect(function() + controller.client:BindObservable("Health", function() + return Observable.new(function() + return nil + end) + end) + end).toThrow() + + controller.destroy() + end) + + it("rejects binding a member as both a method and an observable", function() + local controller = setup() + + controller.server:Bind("Mixed", function() + return true + end) + + expect(function() + controller.server:BindObservable("Mixed", function() + return Observable.new(function() + return nil + end) + end) + end).toThrow() + + controller.destroy() + end) + + it("rejects binding a member as both an observable and a method", function() + local controller = setup() + + controller.server:BindObservable("Mixed", function() + return Observable.new(function() + return nil + end) + end) + + expect(function() + controller.server:Bind("Mixed", function() + return true + end) + end).toThrow() + + controller.destroy() + end) + + it("rejects binding the same observable member twice", function() + local controller = setup() + + local factory = function() + return Observable.new(function() + return nil + end) + end + + controller.server:BindObservable("Health", factory) + + expect(function() + controller.server:BindObservable("Health", factory) + end).toThrow() + + controller.destroy() + end) +end) diff --git a/src/remoting/src/Shared/Observables/RemotingObservableClientRelay.lua b/src/remoting/src/Shared/Observables/RemotingObservableClientRelay.lua new file mode 100644 index 00000000000..2dd2f0c992d --- /dev/null +++ b/src/remoting/src/Shared/Observables/RemotingObservableClientRelay.lua @@ -0,0 +1,200 @@ +--!strict +--[=[ + Client half of the observable relay. Hands out cold observables for a single member and + routes emissions coming down the reserved remote event back to the subscription that + asked for them. + + Outbound messages go through the raw remote event rather than + [Remoting.PromiseFireServer] on purpose. That path resolves synchronously once the + remote exists but asynchronously before it does, so a subscribe issued while waiting + could be overtaken by the unsubscribe that follows it and strand a live subscription on + the server. Here the live subscription table is the queue: nothing is sent until the + remote resolves, and whatever is still subscribed at that point is sent in one pass. + + @class RemotingObservableClientRelay + @private +]=] + +local require = require(script.Parent.loader).load(script) + +local Maid = require("Maid") +local Observable = require("Observable") +local RemotingObservableConstants = require("RemotingObservableConstants") + +local SUBSCRIPTION_KEY_OWNER = "c" + +local RemotingObservableClientRelay = {} +RemotingObservableClientRelay.ClassName = "RemotingObservableClientRelay" +RemotingObservableClientRelay.__index = RemotingObservableClientRelay + +-- The subscription is held as any: a Subscription stored behind a table index type loses +-- its generic methods and stops type checking at the call site +type PendingSubscription = { + subscription: any, + args: { n: number, [number]: any }, +} + +export type RemotingObservableClientRelay = typeof(setmetatable( + {} :: { + _maid: Maid.Maid, + _remoting: any, + _memberName: string, + _reservedMemberName: string, + _subscriptions: { [string]: PendingSubscription }, + _nextSubscriptionId: number, + _remoteEvent: (RemoteEvent | BindableEvent)?, + }, + {} :: typeof({ __index = RemotingObservableClientRelay }) +)) + +--[=[ + Constructs a new client relay for the member and starts listening for emissions. + + @param remoting Remoting + @param memberName string + @return RemotingObservableClientRelay +]=] +function RemotingObservableClientRelay.new(remoting: any, memberName: string): RemotingObservableClientRelay + local self: RemotingObservableClientRelay = setmetatable({} :: any, RemotingObservableClientRelay) + + self._maid = Maid.new() + self._remoting = assert(remoting, "No remoting") + self._memberName = assert(memberName, "No memberName") + self._reservedMemberName = memberName .. RemotingObservableConstants.RESERVED_MEMBER_SUFFIX + self._subscriptions = {} + self._nextSubscriptionId = 0 + + self._maid:GiveTask(self._remoting:Connect(self._reservedMemberName, function(opcode, subscriptionKey, ...) + self:_handleMessage(opcode, subscriptionKey, ...) + end)) + + self._maid:GiveTask( + self._remoting:_observeUpstreamRemoteEventBrio(self._reservedMemberName):Subscribe(function(brio) + if brio:IsDead() then + return + end + + local maid, remoteEvent = brio:ToMaidAndValue() + + self._remoteEvent = remoteEvent + self:_sendLiveSubscriptions() + + maid:GiveTask(function() + if self._remoteEvent == remoteEvent then + self._remoteEvent = nil + end + end) + end) + ) + + return self +end + +--[=[ + Returns a cold observable. Each subscription opens its own stream on the server, so + share the result if more than one consumer needs it. + + @param ... any + @return Observable<...any> +]=] +function RemotingObservableClientRelay.Observe(self: RemotingObservableClientRelay, ...): Observable.Observable<...any> + local args = table.pack(...) + + return Observable.new(function(subscription) + local subscriptionKey = self:_nextSubscriptionKey() + local maid = Maid.new() + + self._subscriptions[subscriptionKey] = { + subscription = subscription :: any, + args = args, + } + + self:_send(RemotingObservableConstants.OPCODE_SUBSCRIBE, subscriptionKey, table.unpack(args, 1, args.n)) + + maid:GiveTask(function() + -- Already cleared when the server ended the stream, so there is nothing to cancel + if self._subscriptions[subscriptionKey] then + self._subscriptions[subscriptionKey] = nil + self:_send(RemotingObservableConstants.OPCODE_UNSUBSCRIBE, subscriptionKey) + end + end) + + return maid + end) :: any +end + +function RemotingObservableClientRelay._handleMessage( + self: RemotingObservableClientRelay, + opcode: any, + subscriptionKey: any, + ... +) + if type(subscriptionKey) ~= "string" then + return + end + + local entry = self._subscriptions[subscriptionKey] + if not entry then + return + end + + if opcode == RemotingObservableConstants.OPCODE_FIRE then + if entry.subscription:IsPending() then + entry.subscription:Fire(...) + end + elseif opcode == RemotingObservableConstants.OPCODE_COMPLETE then + self._subscriptions[subscriptionKey] = nil + if entry.subscription:IsPending() then + entry.subscription:Complete() + end + elseif opcode == RemotingObservableConstants.OPCODE_FAIL then + self._subscriptions[subscriptionKey] = nil + if entry.subscription:IsPending() then + entry.subscription:Fail(...) + end + end +end + +function RemotingObservableClientRelay._sendLiveSubscriptions(self: RemotingObservableClientRelay) + for subscriptionKey, entry in self._subscriptions do + self:_send( + RemotingObservableConstants.OPCODE_SUBSCRIBE, + subscriptionKey, + table.unpack(entry.args, 1, entry.args.n) + ) + end +end + +function RemotingObservableClientRelay._send(self: RemotingObservableClientRelay, opcode: number, ...) + local remoteEvent = self._remoteEvent + if not remoteEvent then + return + end + + self._remoting:_fireUpstreamRemoteEvent(remoteEvent, opcode, ...) +end + +function RemotingObservableClientRelay._nextSubscriptionKey(self: RemotingObservableClientRelay): string + self._nextSubscriptionId += 1 + + return string.format("%s%d", SUBSCRIPTION_KEY_OWNER, self._nextSubscriptionId) +end + +--[=[ + Completes every live subscription and stops listening for emissions. +]=] +function RemotingObservableClientRelay.Destroy(self: RemotingObservableClientRelay) + local subscriptions = self._subscriptions + self._subscriptions = {} + + for _, entry in subscriptions do + if entry.subscription:IsPending() then + entry.subscription:Complete() + end + end + + self._maid:DoCleaning() + setmetatable(self :: any, nil) +end + +return RemotingObservableClientRelay diff --git a/src/remoting/src/Shared/Observables/RemotingObservableConstants.lua b/src/remoting/src/Shared/Observables/RemotingObservableConstants.lua new file mode 100644 index 00000000000..66689514b4d --- /dev/null +++ b/src/remoting/src/Shared/Observables/RemotingObservableConstants.lua @@ -0,0 +1,31 @@ +--!strict +--[=[ + Wire constants for the observable relay used by [Remoting]. + + @class RemotingObservableConstants + @private +]=] + +local require = require(script.Parent.loader).load(script) + +local Table = require("Table") + +return Table.readonly({ + --[[ + Appended to a member name to get the reserved member the relay talks over. A single + RemoteEvent is full duplex, so one reserved member carries both directions. + ]] + RESERVED_MEMBER_SUFFIX = "__Observe" :: "__Observe", + + -- Client to server + OPCODE_SUBSCRIBE = 1, + OPCODE_UNSUBSCRIBE = 2, + + -- Server to client + OPCODE_FIRE = 3, + OPCODE_COMPLETE = 4, + OPCODE_FAIL = 5, + + MAX_SUBSCRIPTION_KEY_LENGTH = 64, + MAX_SUBSCRIPTIONS_PER_PLAYER = 64, +}) diff --git a/src/remoting/src/Shared/Observables/RemotingObservableServerRelay.lua b/src/remoting/src/Shared/Observables/RemotingObservableServerRelay.lua new file mode 100644 index 00000000000..a29fe57a1c0 --- /dev/null +++ b/src/remoting/src/Shared/Observables/RemotingObservableServerRelay.lua @@ -0,0 +1,240 @@ +--!strict +--[=[ + Server half of the observable relay. Owns the bound observable factory for a single + member, turns client subscribe requests into live subscriptions, and forwards each + emission back down the reserved remote event. + + Requests arrive from untrusted clients, so every field off the wire is validated here + and each player is capped to + [RemotingObservableConstants.MAX_SUBSCRIPTIONS_PER_PLAYER] concurrent subscriptions. + + @class RemotingObservableServerRelay + @private +]=] + +local require = require(script.Parent.loader).load(script) + +local Players = game:GetService("Players") + +local Maid = require("Maid") +local Observable = require("Observable") +local RemotingObservableConstants = require("RemotingObservableConstants") + +local RemotingObservableServerRelay = {} +RemotingObservableServerRelay.ClassName = "RemotingObservableServerRelay" +RemotingObservableServerRelay.__index = RemotingObservableServerRelay + +export type ObservableFactory = (player: Player, ...any) -> Observable.Observable<...any> + +export type RemotingObservableServerRelay = typeof(setmetatable( + {} :: { + _maid: Maid.Maid, + _remoting: any, + _memberName: string, + _reservedMemberName: string, + _factory: ObservableFactory, + -- Maid values are held as any: a Maid stored behind a table index type loses its + -- generic methods and stops type checking at the call site + _subscriptions: { [Player]: { [string]: any } }, + _subscriptionCount: { [Player]: number }, + }, + {} :: typeof({ __index = RemotingObservableServerRelay }) +)) + +--[=[ + Constructs a new server relay for the member and starts listening for requests. + + @param remoting Remoting + @param memberName string + @param factory (player: Player, ...any) -> Observable + @return RemotingObservableServerRelay +]=] +function RemotingObservableServerRelay.new( + remoting: any, + memberName: string, + factory: ObservableFactory +): RemotingObservableServerRelay + local self: RemotingObservableServerRelay = setmetatable({} :: any, RemotingObservableServerRelay) + + self._maid = Maid.new() + self._remoting = assert(remoting, "No remoting") + self._memberName = assert(memberName, "No memberName") + self._reservedMemberName = memberName .. RemotingObservableConstants.RESERVED_MEMBER_SUFFIX + self._factory = assert(factory, "No factory") + self._subscriptions = {} + self._subscriptionCount = {} + + self._maid:GiveTask(self._remoting:Connect(self._reservedMemberName, function(player, opcode, subscriptionKey, ...) + self:_handleRequest(player, opcode, subscriptionKey, ...) + end)) + + self._maid:GiveTask(Players.PlayerRemoving:Connect(function(player) + self:_cleanupPlayer(player) + end)) + + return self +end + +function RemotingObservableServerRelay._handleRequest( + self: RemotingObservableServerRelay, + player: Player, + opcode: any, + subscriptionKey: any, + ... +) + if typeof(player) ~= "Instance" then + return + end + + if type(subscriptionKey) ~= "string" then + return + end + + if #subscriptionKey == 0 or #subscriptionKey > RemotingObservableConstants.MAX_SUBSCRIPTION_KEY_LENGTH then + return + end + + if opcode == RemotingObservableConstants.OPCODE_SUBSCRIBE then + self:_handleSubscribe(player, subscriptionKey, ...) + elseif opcode == RemotingObservableConstants.OPCODE_UNSUBSCRIBE then + self:_cleanupSubscription(player, subscriptionKey) + end +end + +function RemotingObservableServerRelay._handleSubscribe( + self: RemotingObservableServerRelay, + player: Player, + subscriptionKey: string, + ... +) + local playerSubscriptions = self._subscriptions[player] + if not playerSubscriptions then + playerSubscriptions = {} + self._subscriptions[player] = playerSubscriptions + self._subscriptionCount[player] = 0 + end + + if playerSubscriptions[subscriptionKey] then + return + end + + if self._subscriptionCount[player] >= RemotingObservableConstants.MAX_SUBSCRIPTIONS_PER_PLAYER then + self:_send(player, RemotingObservableConstants.OPCODE_FAIL, subscriptionKey, "Subscription limit reached") + return + end + + local ok, observable = pcall(self._factory, player, ...) + if not ok then + warn( + string.format( + "[RemotingObservableServerRelay] - Observable factory for %q errored: %s", + self._memberName, + tostring(observable) + ) + ) + self:_send(player, RemotingObservableConstants.OPCODE_FAIL, subscriptionKey, "Failed to construct observable") + return + end + + if not Observable.isObservable(observable) then + warn( + string.format( + "[RemotingObservableServerRelay] - Observable factory for %q returned a non-observable", + self._memberName + ) + ) + self:_send(player, RemotingObservableConstants.OPCODE_FAIL, subscriptionKey, "Failed to construct observable") + return + end + + local subscriptionMaid = Maid.new() + playerSubscriptions[subscriptionKey] = subscriptionMaid + self._subscriptionCount[player] += 1 + + subscriptionMaid:GiveTask(function() + local subscriptions = self._subscriptions[player] + if subscriptions and subscriptions[subscriptionKey] then + subscriptions[subscriptionKey] = nil + self._subscriptionCount[player] -= 1 + end + end) + + local subscription = (observable :: any):Subscribe(function(...) + self:_send(player, RemotingObservableConstants.OPCODE_FIRE, subscriptionKey, ...) + end, function(...) + self:_send(player, RemotingObservableConstants.OPCODE_FAIL, subscriptionKey, ...) + self:_cleanupSubscription(player, subscriptionKey) + end, function() + self:_send(player, RemotingObservableConstants.OPCODE_COMPLETE, subscriptionKey) + self:_cleanupSubscription(player, subscriptionKey) + end) + + -- A source that completes synchronously has already torn the entry down by now + if playerSubscriptions[subscriptionKey] then + subscriptionMaid:GiveTask(subscription) + else + subscription:Destroy() + end +end + +function RemotingObservableServerRelay._send( + self: RemotingObservableServerRelay, + player: Player, + opcode: number, + subscriptionKey: string, + ... +) + self._remoting:FireClient(self._reservedMemberName, player, opcode, subscriptionKey, ...) +end + +function RemotingObservableServerRelay._cleanupSubscription( + self: RemotingObservableServerRelay, + player: Player, + subscriptionKey: string +) + local playerSubscriptions = self._subscriptions[player] + if not playerSubscriptions then + return + end + + local subscriptionMaid = playerSubscriptions[subscriptionKey] + if not subscriptionMaid then + return + end + + subscriptionMaid:DoCleaning() +end + +function RemotingObservableServerRelay._cleanupPlayer(self: RemotingObservableServerRelay, player: Player) + local playerSubscriptions = self._subscriptions[player] + if not playerSubscriptions then + return + end + + self._subscriptions[player] = nil + self._subscriptionCount[player] = nil + + for _, subscriptionMaid in playerSubscriptions do + subscriptionMaid:DoCleaning() + end +end + +--[=[ + Completes every live subscription and stops listening for requests. +]=] +function RemotingObservableServerRelay.Destroy(self: RemotingObservableServerRelay) + for player, playerSubscriptions in self._subscriptions do + for subscriptionKey, subscriptionMaid in playerSubscriptions do + self:_send(player, RemotingObservableConstants.OPCODE_COMPLETE, subscriptionKey) + subscriptionMaid:DoCleaning() + end + end + + table.clear(self._subscriptions) + table.clear(self._subscriptionCount) + + self._maid:DoCleaning() + setmetatable(self :: any, nil) +end + +return RemotingObservableServerRelay diff --git a/src/remoting/src/Shared/Observables/RemotingObservableServerRelay.spec.lua b/src/remoting/src/Shared/Observables/RemotingObservableServerRelay.spec.lua new file mode 100644 index 00000000000..bb6400decd1 --- /dev/null +++ b/src/remoting/src/Shared/Observables/RemotingObservableServerRelay.spec.lua @@ -0,0 +1,510 @@ +--!nonstrict +--[[ + @class RemotingObservableServerRelay.spec.lua +]] + +local require = require(script.Parent.loader).load(script) + +local Workspace = game:GetService("Workspace") + +local Jest = require("Jest") +local Maid = require("Maid") +local Observable = require("Observable") +local PlayerMock = require("PlayerMock") +local RemotingObservableConstants = require("RemotingObservableConstants") +local RemotingObservableServerRelay = require("RemotingObservableServerRelay") + +local describe = Jest.Globals.describe +local expect = Jest.Globals.expect +local it = Jest.Globals.it + +local SUBSCRIBE = RemotingObservableConstants.OPCODE_SUBSCRIBE +local UNSUBSCRIBE = RemotingObservableConstants.OPCODE_UNSUBSCRIBE +local FIRE = RemotingObservableConstants.OPCODE_FIRE +local COMPLETE = RemotingObservableConstants.OPCODE_COMPLETE +local FAIL = RemotingObservableConstants.OPCODE_FAIL + +local function setup(factory) + local maid = Maid.new() + + local sent = {} + local cleanups = 0 + + local fakeRemoting = {} + + function fakeRemoting.Connect(_self, memberName, callback) + fakeRemoting.connectedMember = memberName + fakeRemoting.handler = callback + + return maid:Add(Maid.new()) + end + + function fakeRemoting.FireClient(_self, memberName, player, opcode, subscriptionKey, ...) + table.insert(sent, { + memberName = memberName, + player = player, + opcode = opcode, + subscriptionKey = subscriptionKey, + values = table.pack(...), + }) + end + + local defaultFactory = function(_player, ...) + local args = table.pack(...) + + return Observable.new(function(sub) + sub:Fire(table.unpack(args, 1, args.n)) + + return function() + cleanups += 1 + end + end) + end + + local relay = RemotingObservableServerRelay.new(fakeRemoting, "Health", factory or defaultFactory) + + local controller = {} + + controller.relay = relay + controller.sent = sent + + function controller.newPlayer(userId) + local playerMock = PlayerMock.new({ UserId = userId }) + playerMock.Parent = Workspace + maid:GiveTask(playerMock) + + return playerMock + end + + function controller.request(player, opcode, subscriptionKey, ...) + fakeRemoting.handler(player, opcode, subscriptionKey, ...) + end + + function controller.cleanupCount() + return cleanups + end + + function controller.destroy() + if getmetatable(relay) then + relay:Destroy() + end + maid:DoCleaning() + end + + return controller +end + +describe("RemotingObservableServerRelay.new", function() + it("listens on the reserved member for its own member name", function() + local controller = setup() + + expect(controller.relay._reservedMemberName).toEqual( + "Health" .. RemotingObservableConstants.RESERVED_MEMBER_SUFFIX + ) + + controller.destroy() + end) +end) + +describe("RemotingObservableServerRelay subscribe", function() + it("relays each emission with the requesting subscription key", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1", "hello") + + expect(#controller.sent).toEqual(1) + expect(controller.sent[1].opcode).toEqual(FIRE) + expect(controller.sent[1].subscriptionKey).toEqual("c1") + expect(controller.sent[1].player).toBe(player) + expect(controller.sent[1].values[1]).toEqual("hello") + + controller.destroy() + end) + + it("relays every value of a multi-value emission", function() + local controller = setup(function() + return Observable.new(function(sub) + sub:Fire(1, nil, "three") + return nil + end) + end) + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + + expect(controller.sent[1].values.n).toEqual(3) + expect(controller.sent[1].values[1]).toEqual(1) + expect(controller.sent[1].values[2]).toEqual(nil) + expect(controller.sent[1].values[3]).toEqual("three") + + controller.destroy() + end) + + it("routes emissions only to the player that subscribed", function() + local controller = setup() + local playerA = controller.newPlayer(1) + local playerB = controller.newPlayer(2) + + controller.request(playerA, SUBSCRIBE, "c1", "for-a") + controller.request(playerB, SUBSCRIBE, "c1", "for-b") + + expect(controller.sent[1].player).toBe(playerA) + expect(controller.sent[1].values[1]).toEqual("for-a") + expect(controller.sent[2].player).toBe(playerB) + expect(controller.sent[2].values[1]).toEqual("for-b") + + controller.destroy() + end) + + it("sends complete and forgets the subscription when the source completes", function() + local controller = setup(function() + return Observable.new(function(sub) + sub:Complete() + return nil + end) + end) + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + + expect(controller.sent[1].opcode).toEqual(COMPLETE) + expect(controller.relay._subscriptions[player]["c1"]).toEqual(nil) + expect(controller.relay._subscriptionCount[player]).toEqual(0) + + controller.destroy() + end) + + it("sends fail with the source error and forgets the subscription", function() + local controller = setup(function() + return Observable.new(function(sub) + sub:Fail("nope") + return nil + end) + end) + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + + expect(controller.sent[1].opcode).toEqual(FAIL) + expect(controller.sent[1].values[1]).toEqual("nope") + expect(controller.relay._subscriptions[player]["c1"]).toEqual(nil) + + controller.destroy() + end) + + it("fails the subscription when the factory errors", function() + local controller = setup(function() + error("factory blew up") + end) + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + + expect(controller.sent[1].opcode).toEqual(FAIL) + expect(controller.relay._subscriptions[player]["c1"]).toEqual(nil) + + controller.destroy() + end) + + it("fails the subscription when the factory returns a non-observable", function() + local controller = setup(function() + return { not_an = "observable" } + end) + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + + expect(controller.sent[1].opcode).toEqual(FAIL) + expect(controller.relay._subscriptions[player]["c1"]).toEqual(nil) + + controller.destroy() + end) + + it("does not retain a source that completes during subscribe", function() + local controller = setup(function() + return Observable.new(function(sub) + sub:Fire(1) + sub:Complete() + return nil + end) + end) + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + + expect(controller.relay._subscriptionCount[player]).toEqual(0) + expect(next(controller.relay._subscriptions[player])).toEqual(nil) + + controller.destroy() + end) +end) + +describe("RemotingObservableServerRelay unsubscribe", function() + it("runs the source cleanup and sends nothing back", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1", "hello") + local sentAfterSubscribe = #controller.sent + + controller.request(player, UNSUBSCRIBE, "c1") + + expect(controller.cleanupCount()).toEqual(1) + expect(#controller.sent).toEqual(sentAfterSubscribe) + expect(controller.relay._subscriptionCount[player]).toEqual(0) + + controller.destroy() + end) + + it("ignores an unknown subscription key", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, UNSUBSCRIBE, "never-subscribed") + + expect(controller.cleanupCount()).toEqual(0) + expect(#controller.sent).toEqual(0) + + controller.destroy() + end) + + it("ignores a repeated unsubscribe", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + controller.request(player, UNSUBSCRIBE, "c1") + controller.request(player, UNSUBSCRIBE, "c1") + + expect(controller.cleanupCount()).toEqual(1) + + controller.destroy() + end) + + it("does not touch another player's identically keyed subscription", function() + local controller = setup() + local playerA = controller.newPlayer(1) + local playerB = controller.newPlayer(2) + + controller.request(playerA, SUBSCRIBE, "c1") + controller.request(playerB, SUBSCRIBE, "c1") + + controller.request(playerA, UNSUBSCRIBE, "c1") + + expect(controller.relay._subscriptionCount[playerA]).toEqual(0) + expect(controller.relay._subscriptionCount[playerB]).toEqual(1) + + controller.destroy() + end) +end) + +describe("RemotingObservableServerRelay request validation", function() + it("ignores a non-Instance player", function() + local controller = setup() + + controller.request("not-a-player", SUBSCRIBE, "c1") + + expect(#controller.sent).toEqual(0) + + controller.destroy() + end) + + it("ignores a non-string subscription key", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, 1) + + expect(#controller.sent).toEqual(0) + + controller.destroy() + end) + + it("ignores an empty subscription key", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "") + + expect(#controller.sent).toEqual(0) + + controller.destroy() + end) + + it("ignores an oversized subscription key", function() + local controller = setup() + local player = controller.newPlayer(1) + + local oversized = string.rep("k", RemotingObservableConstants.MAX_SUBSCRIPTION_KEY_LENGTH + 1) + controller.request(player, SUBSCRIBE, oversized) + + expect(#controller.sent).toEqual(0) + + controller.destroy() + end) + + it("accepts a subscription key exactly at the length limit", function() + local controller = setup() + local player = controller.newPlayer(1) + + local atLimit = string.rep("k", RemotingObservableConstants.MAX_SUBSCRIPTION_KEY_LENGTH) + controller.request(player, SUBSCRIBE, atLimit) + + expect(#controller.sent).toEqual(1) + + controller.destroy() + end) + + it("ignores an unknown opcode", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, 999, "c1") + + expect(#controller.sent).toEqual(0) + + controller.destroy() + end) + + it("ignores a duplicate subscription key", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1", "first") + controller.request(player, SUBSCRIBE, "c1", "second") + + expect(#controller.sent).toEqual(1) + expect(controller.relay._subscriptionCount[player]).toEqual(1) + + controller.destroy() + end) +end) + +describe("RemotingObservableServerRelay subscription cap", function() + it("fails the request that exceeds the per-player limit", function() + local controller = setup() + local player = controller.newPlayer(1) + + local limit = RemotingObservableConstants.MAX_SUBSCRIPTIONS_PER_PLAYER + for index = 1, limit do + controller.request(player, SUBSCRIBE, string.format("c%d", index)) + end + + local sentAtLimit = #controller.sent + controller.request(player, SUBSCRIBE, "over") + + expect(controller.relay._subscriptionCount[player]).toEqual(limit) + expect(controller.sent[sentAtLimit + 1].opcode).toEqual(FAIL) + expect(controller.sent[sentAtLimit + 1].subscriptionKey).toEqual("over") + + controller.destroy() + end) + + it("frees a slot once a subscription is released", function() + local controller = setup() + local player = controller.newPlayer(1) + + local limit = RemotingObservableConstants.MAX_SUBSCRIPTIONS_PER_PLAYER + for index = 1, limit do + controller.request(player, SUBSCRIBE, string.format("c%d", index)) + end + + controller.request(player, UNSUBSCRIBE, "c1") + controller.request(player, SUBSCRIBE, "replacement") + + expect(controller.relay._subscriptionCount[player]).toEqual(limit) + expect(controller.relay._subscriptions[player]["replacement"]).never.toEqual(nil) + + controller.destroy() + end) + + it("budgets each player separately", function() + local controller = setup() + local playerA = controller.newPlayer(1) + local playerB = controller.newPlayer(2) + + local limit = RemotingObservableConstants.MAX_SUBSCRIPTIONS_PER_PLAYER + for index = 1, limit do + controller.request(playerA, SUBSCRIBE, string.format("c%d", index)) + end + + controller.request(playerB, SUBSCRIBE, "c1") + + expect(controller.relay._subscriptionCount[playerA]).toEqual(limit) + expect(controller.relay._subscriptionCount[playerB]).toEqual(1) + + controller.destroy() + end) +end) + +describe("RemotingObservableServerRelay teardown", function() + it("drops every subscription belonging to a leaving player", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.request(player, SUBSCRIBE, "c1") + controller.request(player, SUBSCRIBE, "c2") + + controller.relay:_cleanupPlayer(player) + + expect(controller.cleanupCount()).toEqual(2) + expect(controller.relay._subscriptions[player]).toEqual(nil) + expect(controller.relay._subscriptionCount[player]).toEqual(nil) + + controller.destroy() + end) + + it("leaves other players untouched when one leaves", function() + local controller = setup() + local playerA = controller.newPlayer(1) + local playerB = controller.newPlayer(2) + + controller.request(playerA, SUBSCRIBE, "c1") + controller.request(playerB, SUBSCRIBE, "c1") + + controller.relay:_cleanupPlayer(playerA) + + expect(controller.relay._subscriptions[playerA]).toEqual(nil) + expect(controller.relay._subscriptionCount[playerB]).toEqual(1) + + controller.destroy() + end) + + it("completes every live subscription on destroy", function() + local controller = setup() + local playerA = controller.newPlayer(1) + local playerB = controller.newPlayer(2) + + controller.request(playerA, SUBSCRIBE, "c1") + controller.request(playerB, SUBSCRIBE, "c2") + + local sentBeforeDestroy = #controller.sent + controller.relay:Destroy() + + local completes = 0 + for index = sentBeforeDestroy + 1, #controller.sent do + if controller.sent[index].opcode == COMPLETE then + completes += 1 + end + end + + expect(completes).toEqual(2) + expect(controller.cleanupCount()).toEqual(2) + + controller.destroy() + end) + + it("stops accepting requests after destroy", function() + local controller = setup() + local player = controller.newPlayer(1) + + controller.relay:Destroy() + + expect(function() + controller.request(player, SUBSCRIBE, "c1") + end).toThrow() + + controller.destroy() + end) +end)