Fix instance subscriptions (#655)

Remove some assumptions that prevent subscription listeners from working on an
instance.  This means allowing the host to forward the event from the instance
to the controller; correctly storing subscription listeners by address rather
than by connection; and a bunch of network and event stuff to correctly clean up
a listener.

Remove the assumption that all subscriptions events contains the updates
property.  This meant fixing inconsistent types; always checking if the property
existed; fixing lots of the tests that made incorrect assumptions.

Minior fixes:
- Fix double firing of listener callbacks on first connection.
- Fix synced being set to true too early on first connection.
- Fix edge case interaction between last response and last snapshot for values
  with updated set as 0.

Also change the the output made during the start of tests to indicate something
is happening when start up is slow and allow for silent testing to reduce the
amount of text output to the console.

packages/controller/src/BaseConnection.ts

@@ -35,7 +35,7 @@ export default class BaseConnection extends lib.Link {
 
 		this.handle(lib.SubscriptionRequest, this.handleSubscriptionRequest.bind(this));
 		this.connector.on("close", () => {
-			this._controller.subscriptions.unsubscribe(this);
+			this._controller.subscriptions.unsubscribeLink(this);
 		});
 	}
 

packages/controller/src/Controller.ts

@@ -1043,6 +1043,11 @@ export default class Controller {
 	}
 
 	instanceDetailsUpdated(instances: InstanceInfo[]) {
+		for (const instance of instances) {
+			if (instance.status === "stopped") {
+				this.subscriptions.unsubscribeAddress(lib.Address.fromShorthand({instanceId: instance.id}));
+			}
+		}
 		const updates = instances.map(instance => instance.toInstanceDetails());
 		this.subscriptions.broadcast(new lib.InstanceDetailsUpdatesEvent(updates));
 	}

packages/host/src/Instance.ts

@@ -470,6 +470,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 	notifyExit() {
 		this._loadedSave = null;
 		this.notifyStatus("stopped");
+		this.connector.emit("close");
 
 		this.config.off("fieldChanged", this._configFieldChanged);
 		clearTimeout(this._playerCheckInterval);

packages/lib/src/subscriptions.ts

@@ -1,11 +1,12 @@
 import { Type, Static } from "@sinclair/typebox";
-import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link";
-import { logger } from "./logging";
-import { Address, MessageRequest, IControllerUser } from "./data";
+import { Link, Event, EventClass, RequestHandler, WebSocketBaseConnector } from "./link";
+import { Address, MessageRequest, IControllerUser, plainJson, JsonBoolean } from "./data";
 import isDeepStrictEqual from "./is_deep_strict_equal";
+import { RequestError, SessionLost } from "./errors";
+import { logger } from "./logging";
 
 export type SubscriptionRequestHandler<T> = RequestHandler<SubscriptionRequest, Event<T> | null>;
-export type EventSubscriberCallback<T> = (updates: T[], synced: boolean) => void
+export type EventSubscriberCallback<T> = (event: T | null, synced: boolean) => void
 
 /**
  * A subscription request sent by a subscriber, this updates what events the subscriber will be sent
@@ -15,7 +16,7 @@ export type EventSubscriberCallback<T> = (updates: T[], synced: boolean) => void
 export class SubscriptionRequest {
 	declare ["constructor"]: typeof SubscriptionRequest;
 	static type = "request" as const;
-	static src = ["control", "instance"] as const;
+	static src = ["control", "host", "instance"] as const;
 	static dst = "controller" as const;
 	static permission(user: IControllerUser, message: MessageRequest) {
 		if (typeof message.data === "object" && message.data !== null) {
@@ -31,6 +32,9 @@ export class SubscriptionRequest {
 		}
 	}
 
+	/** Indicates if updates have been sent */
+	static Response = JsonBoolean;
+
 	constructor(
 		public eventName: string,
 		public subscribe: boolean,
@@ -56,9 +60,14 @@ export class SubscriptionRequest {
 	}
 }
 
+type Subscriber = {
+	link: Link,
+	dst: Address,
+}
+
 type EventData = {
 	subscriptionUpdate?: SubscriptionRequestHandler<unknown>,
-	subscriptions: Set<Link>,
+	subscriptions: Array<Subscriber>,
 };
 
 /**
@@ -85,7 +94,7 @@ export class SubscriptionController {
 		}
 		this._events.set(entry.name, {
 			subscriptionUpdate: subscriptionUpdate,
-			subscriptions: new Set(),
+			subscriptions: [],
 		});
 	}
 
@@ -102,13 +111,28 @@ export class SubscriptionController {
 		if (!eventData) {
 			throw new Error(`Event ${entry.name} is not a registered as subscribable`);
 		}
-		for (let link of eventData.subscriptions) {
-			if ((link.connector as WebSocketBaseConnector).closing) {
-				eventData.subscriptions.delete(link);
+		const toRemove: Subscriber[] = [];
+		for (const subscriber of eventData.subscriptions) {
+			if ((subscriber.link.connector as WebSocketBaseConnector).closing) {
+				toRemove.push(subscriber);
 			} else {
-				link.send(event);
+				subscriber.link.sendTo(subscriber.dst, event);
 			}
 		}
+		eventData.subscriptions = eventData.subscriptions.filter(subscriber => !toRemove.includes(subscriber));
+	}
+
+	/**
+	 * Unsubscribe from all events of a given address.
+	 * Used when an instance is stopped to stop all active subscriptions.
+	 * @param address - Address to stop sending events to.
+	 */
+	unsubscribeAddress(address: Address) {
+		for (let eventData of this._events.values()) {
+			eventData.subscriptions = eventData.subscriptions.filter(
+				subscriber => !subscriber.dst.addressedTo(address)
+			);
+		}
 	}
 
 	/**
@@ -116,9 +140,11 @@ export class SubscriptionController {
 	 * Used when a link is closed to stop all active subscriptions.
 	 * @param link - Link to stop sending events to.
 	 */
-	unsubscribe(link: Link) {
+	unsubscribeLink(link: Link) {
 		for (let eventData of this._events.values()) {
-			eventData.subscriptions.delete(link);
+			eventData.subscriptions = eventData.subscriptions.filter(
+				subscriber => subscriber.link !== link
+			);
 		}
 	}
 
@@ -138,16 +164,23 @@ export class SubscriptionController {
 			throw new Error(`Event ${event.eventName} is not a registered as subscribable`);
 		}
 		if (event.subscribe === false) {
-			eventData.subscriptions.delete(link);
+			const index = eventData.subscriptions.findLastIndex(subscriber => subscriber.dst.addressedTo(src));
+			if (index >= 0) {
+				eventData.subscriptions.splice(index, 1);
+			}
 		} else {
-			eventData.subscriptions.add(link);
+			if (!eventData.subscriptions.some(subscriber => subscriber.dst.addressedTo(src))) {
+				eventData.subscriptions.push({ link: link, dst: src });
+			}
 			if (eventData.subscriptionUpdate) {
 				const eventReplay = await eventData.subscriptionUpdate(event, src, dst);
 				if (eventReplay) {
-					link.send(eventReplay);
+					link.sendTo(src, eventReplay);
+					return true;
 				}
 			}
 		}
+		return false;
 	}
 }
 
@@ -157,9 +190,9 @@ export interface SubscribableValue {
 	isDeleted: boolean,
 }
 
-export interface EventSubscribable<T, V extends SubscribableValue> extends Event<T> {
+export type EventSubscribable<T, V extends SubscribableValue> = Event<T> & Partial<{
 	updates: V[],
-}
+}>
 
 /**
  * Component for subscribing to and tracking updates of a remote resource
@@ -167,43 +200,61 @@ export interface EventSubscribable<T, V extends SubscribableValue> extends Event
  */
 export class EventSubscriber<
 	T extends EventSubscribable<T, V>,
-	K extends string | number = T["updates"][number]["id"],
-	V extends SubscribableValue = T["updates"][number],
+	K extends string | number = NonNullable<T["updates"]>[number]["id"],
+	V extends SubscribableValue = NonNullable<T["updates"]>[number],
 > {
-	_callbacks = new Array<EventSubscriberCallback<V>>();
+	/** The time at which an event was last received, is less than 0 when there have been no events */
+	lastResponseTimeMs = -1;
 	/** Values of the subscribed resource */
 	values = new Map<K, V>();
 	/** True if this subscriber is currently synced with the source */
 	synced = false;
-	_snapshot?: readonly [ReadonlyMap<K, Readonly<V>>, boolean];
-	_snapshotLastUpdatedMs = 0;
-	lastResponseTimeMs = -1;
+	/** True if this subscriber is expecting to receive updates from the source */
+	_syncing = false;
+	/** Repeat calls to getSnapshot will return the same readonly copy unless values has updated */
+	_snapshot: readonly [ReadonlyMap<K, Readonly<V>>, boolean] = [new Map<K, V>(), false];
+	_snapshotLastUpdatedMs = -1;
+	/** Callbacks will be called when an event is received or the synced state changes */
+	_callbacks = new Array<EventSubscriberCallback<T>>();
 
 	constructor(
 		private Event: EventClass<T>,
 		public control: Link,
 	) {
-		control.handle(Event, this._handle.bind(this));
+		control.handle(Event, this._handleEvent.bind(this));
 		control.connector.on("connect", () => {
-			this._updateSubscription();
+			this.handleConnectionEvent("connect");
 		});
 		control.connector.on("close", () => {
-			if (this.synced) {
-				this.synced = false;
-				for (let callback of this._callbacks) {
-					callback([], this.synced);
-				}
-			}
+			this.handleConnectionEvent("close");
 		});
 	}
 
+	/**
+	 * Handle connection events from this connector or another in the chain to the controller
+	 * This should be called by an instance plugin during onControllerConnectionEvent
+	 * @param event - event type that occurred
+	 */
+	handleConnectionEvent(event: "connect" | "drop" | "resume" | "close") {
+		if (event === "connect" || event === "resume") {
+			this._updateSubscription();
+		} else if (this.synced) {
+			this.synced = false;
+			for (let callback of this._callbacks) {
+				callback(null, false);
+			}
+		}
+	}
+
 	/**
 	 * Handle incoming events and distribute it to the correct callbacks
 	 * @param event - event from subscribed resource
 	 * @internal
 	 */
-	async _handle(event: EventSubscribable<T, V>) {
-		for (const value of event.updates) {
+	async _handleEvent(event: T) {
+		// We support the automatic maintaining of a map for events with the updates property
+		const updates = event.updates ?? [];
+		for (const value of updates) {
 			this.lastResponseTimeMs = Math.max(this.lastResponseTimeMs, value.updatedAtMs);
 			// Warn about updates which changes content but don't update updatedAtMs
 			const existing = this.values.get(value.id as K);
@@ -219,8 +270,13 @@ export class EventSubscriber<
 				this.values.set(value.id as K, value);
 			}
 		}
-		for (let callback of this._callbacks) {
-			callback(event.updates, this.synced);
+		// Event handling logic
+		if (this._syncing && !this.synced) {
+			this._snapshotLastUpdatedMs = -1;
+			this.synced = true;
+		}
+		for (const callback of this._callbacks) {
+			callback(event, this.synced);
 		}
 	}
 
@@ -228,10 +284,10 @@ export class EventSubscriber<
 	 * Subscribe to receive all event notifications
 	 * @param handler -
 	 *     callback invoked whenever the subscribed resource changes or the
-	 *     synced property changes, in which case the updates will be empty.
+	 *     synced property changes, in which case the event will be null.
 	 * @returns function that will unsubscribe from notifications
 	 */
-	subscribe(handler: EventSubscriberCallback<V>) {
+	subscribe(handler: EventSubscriberCallback<T>) {
 		this._callbacks.push(handler);
 		if (this._callbacks.length === 1) {
 			this._updateSubscription();
@@ -277,26 +333,30 @@ export class EventSubscriber<
 	 * Update the subscription with the controller based on current handler counts
 	 */
 	async _updateSubscription() {
-		if (!(this.control.connector as WebSocketClientConnector).connected) {
+		if (!this.control.connector.valid) {
 			return;
 		}
 		const entry = Link._eventsByClass.get(this.Event)!;
 
 		try {
-			await this.control.send(new SubscriptionRequest(
+			this._syncing = this._callbacks.length > 0;
+			const updatesSent = await this.control.sendTo("controller", new SubscriptionRequest(
 				entry.name,
-				this._callbacks.length > 0,
+				this._syncing,
 				this.lastResponseTimeMs
 			));
-			this.synced = this._callbacks.length > 0;
-			this._snapshotLastUpdatedMs = 0;
-			for (let callback of this._callbacks) {
-				callback([], this.synced);
+			if (!updatesSent) {
+				this.synced = this._syncing;
+				for (const callback of this._callbacks) {
+					callback(null, this.synced);
+				}
 			}
 		} catch (err: any) {
-			logger.error(
-				`Unexpected error updating ${entry.name} subscription:\n${err.stack}`
-			);
+			if (!(err instanceof RequestError) || err.code !== "SessionLost") {
+				logger.error(
+					`Unexpected error updating ${entry.name} subscription:\n${err.stack}`
+				);
+			}
 		}
 	}
 }

test/integration/index.js

@@ -161,20 +161,30 @@ function getControl() {
 }
 
 function spawn(name, cmd, waitFor) {
+	// eslint-disable-next-line node/no-process-env
+	const silent = process.env.SILENT_TEST;
 	return new Promise((resolve, reject) => {
 		console.log(cmd);
 		let parts = cmd.split(" ");
 		let process = child_process.spawn(parts[0], parts.slice(1), { cwd: path.join("temp", "test") });
 		let stdout = new LineSplitter({ readableObjectMode: true });
-		stdout.on("data", line => {
+		let stderr = new LineSplitter({ readableObjectMode: true });
+		let onDataOut = line => {
 			line = line.toString("utf8");
 			if (waitFor.test(line)) {
+				if (silent) {
+					stdout.off("data", onDataOut);
+					stderr.off("data", onDataErr);
+				}
 				resolve(process);
 			}
 			console.log(name, line);
-		});
-		let stderr = new LineSplitter({ readableObjectMode: true });
-		stderr.on("data", line => { console.log(name, line.toString("utf8")); });
+		};
+		let onDataErr = line => {
+			console.log(name, line.toString("utf8"));
+		};
+		stdout.on("data", onDataOut);
+		stderr.on("data", onDataErr);
 		process.stdout.pipe(stdout);
 		process.stderr.pipe(stderr);
 	});
@@ -183,10 +193,17 @@ function spawn(name, cmd, waitFor) {
 before(async function() {
 	this.timeout(40000);
 
+	// eslint-disable-next-line node/no-process-env
+	const silent = process.env.SILENT_TEST;
+	if (silent) {
+		console.log("SILENT_TEST is present in env, loggers after bootstrap will be muted.");
+	}
+
 	// Some integration tests may cause log events
 	logger.add(new ConsoleTransport({
 		level: "info",
 		format: new lib.TerminalFormat(),
+		filter: () => !silent,
 	}));
 
 	// If fast test is enabled then output that it is
@@ -206,6 +223,7 @@ before(async function() {
 
 	await fs.ensureDir(path.join("temp", "test"));
 
+	console.log("Building Mods");
 	await fs.ensureDir(modsDir);
 	await lib.build({
 		build: true,
@@ -232,6 +250,7 @@ before(async function() {
 	}
 	await fs.copyFile(`mods/${ssZip}`, `${modsDir}/${ssZip}`);
 
+	console.log("Setting Controller Config");
 	await exec("node ../../packages/controller config set controller.auth_secret TestSecretDoNotUse");
 	await exec("node ../../packages/controller config set controller.http_port 8880");
 	await exec("node ../../packages/controller config set controller.https_port 4443");
@@ -240,12 +259,14 @@ before(async function() {
 	await exec("node ../../packages/controller config set controller.tls_certificate ../../test/file/tls/cert.pem");
 	await exec("node ../../packages/controller config set controller.tls_private_key ../../test/file/tls/key.pem");
 
+	console.log("Setting Controller Plugins");
 	await exec("node ../../packages/ctl plugin add ../../plugins/global_chat");
 	await exec("node ../../packages/ctl plugin add ../../plugins/research_sync");
 	await exec("node ../../packages/ctl plugin add ../../plugins/statistics_exporter");
 	await exec("node ../../packages/ctl plugin add ../../plugins/subspace_storage");
 	await exec("node ../../packages/ctl plugin add ../../plugins/player_auth");
 
+	console.log("Bootstrapping");
 	await exec("node ../../packages/controller bootstrap create-admin test");
 	await exec("node ../../packages/controller bootstrap create-ctl-config test");
 	await exec("node ../../packages/ctl control-config set control.tls_ca ../../test/file/tls/cert.pem");

test/lib/subscriptions.js

@@ -101,13 +101,13 @@ describe("lib/subscriptions", function() {
 		const connectorSetupDate = [{
 			id: 0,
 			request: new lib.SubscriptionRequest(RegisteredEvent.name, true),
-			src: addr({ controlId: 0 }),
-			dst: addr("controller"),
+			dst: addr({ controlId: 0 }),
+			src: addr("controller"),
 		}, {
 			id: 1,
 			request: new lib.SubscriptionRequest(RegisteredEvent.name, true),
-			src: addr({ controlId: 1 }),
-			dst: addr("controller"),
+			dst: addr({ controlId: 1 }),
+			src: addr("controller"),
 		}];
 
 		function getLink(id) {
@@ -163,7 +163,7 @@ describe("lib/subscriptions", function() {
 				subscriptions.handle(RegisteredEvent);
 				for (let connectorData of connectorSetupDate) {
 					subscriptions.handleRequest(
-						getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst
+						getLink(connectorData.id), connectorData.request, connectorData.dst, connectorData.src
 					);
 				}
 			});
@@ -190,7 +190,7 @@ describe("lib/subscriptions", function() {
 				const connectorData = connectorSetupDate[0];
 				const request = new lib.SubscriptionRequest(RegisteredEvent.name, false);
 				await subscriptions.handleRequest(
-					getLink(connectorData.id), request, connectorData.src, connectorData.dst,
+					getLink(connectorData.id), request, connectorData.dst, connectorData.src
 				);
 				subscriptions.broadcast(new RegisteredEvent());
 				await onceConnectorSend(1);
@@ -212,14 +212,22 @@ describe("lib/subscriptions", function() {
 				subscriptions.handle(RegisteredEvent);
 				for (let connectorData of connectorSetupDate) {
 					subscriptions.handleRequest(
-						getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst
+						getLink(connectorData.id), connectorData.request, connectorData.dst, connectorData.src
 					);
 				}
 			});
 
-			it("should remove an active subscription", async function() {
+			it("should remove an active subscription link", async function() {
 				const link = getLink(0);
-				subscriptions.unsubscribe(link);
+				subscriptions.unsubscribeLink(link);
+				subscriptions.broadcast(new RegisteredEvent());
+				await onceConnectorSend(1);
+				assertNoEvent(0);
+				assertLastEvent(1, RegisteredEvent);
+			});
+			it("should remove an active subscription address", async function() {
+				const link = getLink(0);
+				subscriptions.unsubscribeAddress(connectorSetupDate[0].dst);
 				subscriptions.broadcast(new RegisteredEvent());
 				await onceConnectorSend(1);
 				assertNoEvent(0);
@@ -251,7 +259,7 @@ describe("lib/subscriptions", function() {
 			it("should accept a subscription", async function() {
 				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
 				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst
+					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
 				);
 				subscriptions.broadcast(new RegisteredEvent());
 				await onceConnectorSend(0);
@@ -260,7 +268,7 @@ describe("lib/subscriptions", function() {
 			it("should accept an unsubscription", async function() {
 				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
 				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst
+					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
 				);
 				subscriptions.broadcast(new RegisteredEvent());
 				await onceConnectorSend(0);
@@ -271,7 +279,7 @@ describe("lib/subscriptions", function() {
 
 				const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false);
 				await subscriptions.handleRequest(
-					getLink(0), unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst
+					getLink(0), unsubRequest, connectorSetupDate[0].dst, connectorSetupDate[0].src
 				);
 				subscriptions.broadcast(new RegisteredEvent());
 				assertNoEvent(0);
@@ -283,7 +291,7 @@ describe("lib/subscriptions", function() {
 
 				const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true);
 				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst
+					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
 				);
 				await onceConnectorSend(0);
 				assertLastEvent(0, StringPermissionEvent);
@@ -301,7 +309,7 @@ describe("lib/subscriptions", function() {
 
 				const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true);
 				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst
+					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
 				);
 				await new Promise(resolve => setImmediate(resolve));
 				assertNoEvent(0);
@@ -319,12 +327,12 @@ describe("lib/subscriptions", function() {
 			registeredEvent = new lib.EventSubscriber(RegisteredEvent, mockControl);
 		});
 
-		function assertLastRequest(request) {
-			const Request = request.constructor;
+		function assertLastMessage(message) {
+			const Message = message.constructor;
 			const lastMessage = mockControl.connector.sentMessages.at(-1);
-			const name = Request.plugin ? `${Request.plugin}:${Request.name}` : Request.name;
+			const name = Message.plugin ? `${Message.plugin}:${Message.name}` : Message.name;
 			assert.equal(lastMessage.name, name);
-			assert.deepEqual(lastMessage.data, request);
+			assert.deepEqual(lastMessage.data, message);
 		}
 
 		function assertNoMessageSent() {
@@ -348,19 +356,19 @@ describe("lib/subscriptions", function() {
 				let calledWith = null;
 				let calledWithTwo = null;
 				const item = { id: 1, updatedAtMs: 1, isDeleted: false };
-				const event = new RegisteredEvent([item]);
-				registeredEvent.subscribe(function(updates, synced) { calledWith = [updates, synced]; });
-				registeredEvent.subscribe(function(updates, synced) { calledWithTwo = [updates, synced]; });
-				await registeredEvent._handle(event);
-				assert.deepEqual(calledWith, [[item], false]);
-				assert.deepEqual(calledWithTwo, [[item], false]);
+				const eventUpdate = new RegisteredEvent([item]);
+				registeredEvent.subscribe(function(event, synced) { calledWith = [event, synced]; });
+				registeredEvent.subscribe(function(event, synced) { calledWithTwo = [event, synced]; });
+				await registeredEvent._handleEvent(eventUpdate);
+				assert.deepEqual(calledWith, [eventUpdate, true]);
+				assert.deepEqual(calledWithTwo, [eventUpdate, true]);
 			});
 			it("should allow unsubscribing from an event", async function() {
 				let called = false;
 				let event = new RegisteredEvent();
 				function callback() { called = true; }
 				const unsubscribe = registeredEvent.subscribe(callback);
-				await registeredEvent._handle(event);
+				await registeredEvent._handleEvent(event);
 				assert.deepEqual(called, true);
 
 				called = false;
@@ -368,7 +376,7 @@ describe("lib/subscriptions", function() {
 				unsubscribe();
 				// Wait for unsubscribe to happen.
 				await new Promise(resolve => setImmediate(resolve));
-				await registeredEvent._handle(event);
+				await registeredEvent._handleEvent(event);
 				assert.equal(called, false);
 			});
 			it("should not throw error on incorrectly unsubscribing multiple times", async function() {
@@ -390,13 +398,13 @@ describe("lib/subscriptions", function() {
 				const update3 = [{ id: 1, updatedAtMs: 4, isDeleted: true }];
 				const snap0 = registeredEvent.getSnapshot();
 				assert.deepEqual(snap0, [new Map(), false]);
-				await registeredEvent._handle(new RegisteredEvent(update1));
+				await registeredEvent._handleEvent(new RegisteredEvent(update1));
 				const snap1 = registeredEvent.getSnapshot();
 				assert.deepEqual(snap1, [new Map([[1, update1[0]]]), false]);
-				await registeredEvent._handle(new RegisteredEvent(update2));
+				await registeredEvent._handleEvent(new RegisteredEvent(update2));
 				const snap2 = registeredEvent.getSnapshot();
 				assert.deepEqual(snap2, [new Map([[1, update2[0]], [2, update2[1]]]), false]);
-				await registeredEvent._handle(new RegisteredEvent(update3));
+				await registeredEvent._handleEvent(new RegisteredEvent(update3));
 				const snap3 = registeredEvent.getSnapshot();
 				assert.deepEqual(snap3, [new Map([[2, update2[1]]]), false]);
 				// validate previous snapshots were not modified
@@ -406,59 +414,108 @@ describe("lib/subscriptions", function() {
 			});
 		});
 
+		describe("handleConnectionEvent", function() {
+			it("should be called on connector connect event", async function() {
+				registeredEvent.subscribe(() => {});
+				await events.once(mockControl.connector, "send");
+				mockControl.connector.sentMessages.pop();
+
+				registeredEvent.control.connector.emit("connect");
+				await events.once(mockControl.connector, "send");
+				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
+				mockControl.connector.sentMessages.pop();
+
+				registeredEvent.handleConnectionEvent("connect");
+				await events.once(mockControl.connector, "send");
+				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
+				mockControl.connector.sentMessages.pop();
+
+				registeredEvent.handleConnectionEvent("resume");
+				await events.once(mockControl.connector, "send");
+				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
+			});
+			it("should set synced to false on connection close", async function() {
+				registeredEvent.synced = true;
+				registeredEvent.control.connector.emit("close");
+				assert(!registeredEvent.synced, "synced was not set to false during close event");
+
+				registeredEvent.synced = true;
+				registeredEvent.handleConnectionEvent("drop");
+				assert(!registeredEvent.synced, "synced was not set to false during drop");
+
+				registeredEvent.synced = true;
+				registeredEvent.handleConnectionEvent("close");
+				assert(!registeredEvent.synced, "synced was not set to false during close");
+			});
+			it("should invoke callbacks on connector close", async function() {
+				let calledWith;
+				registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
+				const [msg] = await events.once(mockControl.connector, "send");
+				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
+					lib.SubscriptionRequest.Response.fromJSON(false)
+				));
+				await new Promise(resolve => setImmediate(resolve));
+				assert(registeredEvent.synced, "synced was not set to true before close");
+				calledWith = undefined;
+
+				registeredEvent.control.connector.emit("close");
+				assert(!registeredEvent.synced, "synced was not set to false");
+				assert(calledWith !== undefined, "callback not called");
+				assert.deepEqual(calledWith, [null, false]);
+			});
+		});
+
 		describe("_updateSubscription()", function() {
-			it("should do nothing if not connected", async function() {
-				registeredEvent.control.connector.connected = false;
+			it("should do nothing if connection is not valid", async function() {
+				registeredEvent.control.connector.valid = false;
 				registeredEvent.subscribe(() => true);
 				await new Promise(resolve => setImmediate(resolve));
 				assertNoMessageSent();
 			});
 			it("should correctly request a subscription", async function() {
-				const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true, -1);
 				registeredEvent.subscribe(() => true);
 				await events.once(mockControl.connector, "send");
-				assertLastRequest(expected);
+				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
 			});
-			it("should correctly request no subscriptions", async function() {
-				const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false, -1);
+			it("should correctly request an unsubscription", async function() {
 				function callback() {}
 				const unsubscribe = registeredEvent.subscribe(callback);
 				await events.once(mockControl.connector, "send");
 				unsubscribe();
 				await events.once(mockControl.connector, "send");
-				assertLastRequest(expected);
+				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, false, -1));
 			});
 			it("should set synced and invoke callbacks when request is responded to", async function() {
 				let calledWith;
 				registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
 				const [msg] = await events.once(mockControl.connector, "send");
 				assert(!registeredEvent.synced, "synced was set before response");
-				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src));
+				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
+					lib.SubscriptionRequest.Response.fromJSON(false)
+				));
 				await new Promise(resolve => setImmediate(resolve));
 				assert(calledWith !== undefined, "callback not called");
-				assert.deepEqual(calledWith, [[], true]);
+				assert.deepEqual(calledWith, [null, true]);
 				assert(registeredEvent.synced, "synced was not set");
 			});
-			it("should be called on connector connect event", async function() {
-				const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true, -1);
-				registeredEvent._callbacks.push(() => {});
-				registeredEvent.control.connector.emit("connect");
-				await events.once(mockControl.connector, "send");
-				assertLastRequest(expected);
-			});
-			it("should revert synced and invoke callbacks on connector close event", async function() {
+			it("should set synced and invoke callbacks when request is responded to with data", async function() {
 				let calledWith;
 				registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
 				const [msg] = await events.once(mockControl.connector, "send");
-				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src));
+				assert(!registeredEvent.synced, "synced was set before response");
+				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
+					lib.SubscriptionRequest.Response.fromJSON(true)
+				));
 				await new Promise(resolve => setImmediate(resolve));
 
-				assert(registeredEvent.synced, "synced was not set before close");
-				calledWith = undefined;
-				registeredEvent.control.connector.emit("close");
+				assert(!registeredEvent.synced, "synced was set before update");
+				assert(calledWith === undefined, "callback called before update");
+				const eventUpdate = new RegisteredEvent();
+				registeredEvent._handleEvent(eventUpdate);
+
+				assert(registeredEvent.synced, "synced was not set");
 				assert(calledWith !== undefined, "callback not called");
-				assert.deepEqual(calledWith, [[], false]);
-				assert(!registeredEvent.synced, "synced was not reset");
+				assert.deepEqual(calledWith, [eventUpdate, true]);
 			});
 		});
 	});

test/mock.js

@@ -72,6 +72,7 @@ class MockConnector extends lib.BaseConnector {
 	constructor(src, dst) {
 		super(src, dst);
 
+		this.valid = true;
 		this.connected = true;
 		this.hasSession = true;
 		this.sentMessages = [];