Break subscription management out to properly handle unsubscribing

README.md

@@ -2,13 +2,13 @@
 
 Reactimal provides a set of primitives that can be used to express your code as a logic graph using
 reactive programming methods. Reactive programming inverts the dependency relationship between
-functions and their parameters. No longer do callers need to know the signature of the functions
+functions and their inputs. No longer do callers need to know the signature of the functions
 they call. Instead computed values subscribe to scalar properties or other computed values. The
 reactive method improves code readability and reduces the tendency toward spaghetti code.
 
 Reactimal provides the following reactive programming primitives. Each type of primitive is
-[subscribable](./docs/subscribable.md) and provides a single output value, but differ in how they
+[subscribable](./docs/subscribable.md) and provides a single output value (or promise), but differ
-receive their input.
+in how they receive their input.
 
 ## Input Primitives
 

src/computed.js

@@ -1,7 +1,8 @@
-import { id, registerFire, registerSubscriptions, call } from './util.js';
+import { id, call } from './util.js';
+import { subscribable } from './subscribable.js';
+
 
 export const hashableComputed = hash => (fn, dependencies = []) => {
-	let subscribers = [];
 	let isDirty = true;
 	let val;
 	let oldId;
@@ -25,29 +26,23 @@ export const hashableComputed = hash => (fn, dependencies = []) => {
 	};
 
 	// Add child nodes to the logic graph (value-based)
-	accessor.subscribe = registerSubscriptions(subscribers);
+	Object.assign(accessor, subscribable());
-	accessor._fire = registerFire(subscribers);
 
 	// Receive dirty flag from parent logic node (dependency).  Pass it down.
 	accessor._setDirty = function setDirty() {
 		if (!isDirty) {
 			isDirty = true;
-			subscribers.forEach(s => s._setDirty && s._setDirty());
+			accessor._forEachSubscription(s => s._setDirty && s._setDirty());
 		}
-		return subscribers.length && accessor;
+		return accessor._hasSubscribers() && accessor;
 	};
 
 	// Remove this node from the logic graph completely
 	accessor.detach = () => {
-		subscribers = [];
+		accessor.unsubscribeAll();
 		dependentSubscriptions.forEach(call);
 	};
 
-	// Remove child nodes from the logic graph
-	accessor.unsubscribeAll = () => {
-		subscribers = [];
-	};
-
 	const dependentSubscriptions = dependencies.map(d => d.subscribe(accessor._setDirty));
 
 	return accessor;

src/container.js

@@ -1,16 +1,10 @@
-import { registerSubscriptions, registerFire } from './util.js';
+import { subscribable } from './subscribable.js';
 
 export const hashableContainer = hash => store => {
-	let subscribers = [];
 	let id = hash && hash(store);
 	let lockCount = 0;
 
 	const containerMethods = {
-		subscribe: registerSubscriptions(subscribers),
-		_fire: registerFire(subscribers),
-		unsubscribeAll: () => {
-			subscribers = [];
-		},
 		_lock: () => {
 			lockCount += 1;
 			return p;
@@ -19,7 +13,8 @@ export const hashableContainer = hash => store => {
 			if (lockCount && --lockCount === 0) {
 				checkUpdate(store);
 			}
-		}
+		},
+		...subscribable()
 	};
 
 	function checkUpdate(target) {

src/container.test.js

@@ -137,4 +137,21 @@ describe('A container', () => {
 		a.add(1);
 		expect(order).toEqual('abc');
 	});
+
+	it('unsubscribeAll clears subscriptions', () => {
+		let order = '';
+
+		const a = containedSet(new Set());
+		a.subscribe(() => (order += 'a'));
+		a.subscribe(() => (order += 'b'));
+		a.subscribe(() => (order += 'c'));
+		a.unsubscribeAll()
+		a.add(1);
+		expect(order).toEqual('');
+	});
+
+	it('protects method properties.', () => {
+		const a = containedSet(new Set());
+		expect(() => a.subscribe = 'test').toThrow(new ReferenceError('Cannot set subscribe in [object Set]'));
+	});
 });

src/property.js

@@ -1,4 +1,6 @@
-import { id, registerSubscriptions, registerFire } from './util.js';
+import { id } from './util.js';
+import { subscribable } from './subscribable.js';
+
 
 export const hashableProperty = hash => store => {
 	let subscribers = [];
@@ -16,9 +18,8 @@ export const hashableProperty = hash => store => {
 		}
 		return store;
 	};
-	accessor.subscribe = registerSubscriptions(subscribers);
+	// Add child nodes to the logic graph (value-based)
-	accessor.unsubscribeAll = () => (subscribers = []);
+	Object.assign(accessor, subscribable());
-	accessor._fire = registerFire(subscribers);
 	accessor._lock = () => {
 		lockCount += 1;
 		return accessor();

src/stream.js

@@ -1,4 +1,6 @@
-import { id, registerFire, registerSubscriptions, call } from './util.js';
+import { id, call } from './util.js';
+import { subscribable } from './subscribable.js';
+
 
 export const hashableStream = hash => (fn, dependencies = []) => {
 	let subscribers = [];
@@ -37,29 +39,23 @@ export const hashableStream = hash => (fn, dependencies = []) => {
 	};
 
 	// Add child nodes to the logic graph (value-based)
-	accessor.subscribe = registerSubscriptions(subscribers);
+	Object.assign(accessor, subscribable());
-	accessor._fire = registerFire(subscribers);
 
 	// Receive dirty flag from parent logic node (dependency).  Pass it down.
 	accessor._setDirty = function setDirty() {
 		if (!isDirty) {
 			isDirty = true;
-			subscribers.forEach(s => s._setDirty && s._setDirty());
+			accessor._forEachSubscription(s => s._setDirty && s._setDirty());
 		}
-		return subscribers.length && accessor;
+		return accessor._hasSubscribers() && accessor;
 	};
 
 	// Remove this node from the logic graph completely
 	accessor.detach = () => {
-		subscribers = [];
+		accessor.unsubscribeAll();
 		dependentSubscriptions.forEach(call);
 	};
 
-	// Remove child nodes from the logic graph
-	accessor.unsubscribeAll = () => {
-		subscribers = [];
-	};
-
 	const dependentSubscriptions = dependencies.map(d => d.subscribe(accessor._setDirty));
 
 	return accessor;

src/stream.test.js

@@ -155,4 +155,28 @@ describe('A stream', () => {
 		await aDep();
 		expect(aCount).toBe(2);
 	});
+
+	it('detach breaks the logic chain', async () => {
+		const a = prop(0);
+		const b = computed(square, [a]);
+		const c = stream(delaySquare, [a]);
+		const d = stream(delayAdd, [a, c]);
+		const e = stream(delaySquare, [b]);
+
+		expect(await c()).toEqual(0);
+		expect(await d()).toEqual(0);
+		expect(await e()).toEqual(0);
+
+		a(1);
+		expect(await c()).toEqual(1);
+		expect(await d()).toEqual(2);
+		expect(await e()).toEqual(1);
+
+		c.detach();
+
+		a(2);
+		expect(await c()).toEqual(1);
+		expect(await d()).toEqual(3);
+		expect(await e()).toEqual(16);
+	});
 });

src/subscribable.js

@@ -0,0 +1,30 @@
+import { call } from './util.js';
+
+export const subscribable = () => {
+	const subscriptions = [];
+
+	return {
+		subscribe(fn) {
+			subscriptions.push(fn);
+			return () => {
+				const idx = subscriptions.indexOf(fn);
+				if (idx !== -1) {
+					subscriptions.splice(idx, 1);
+				}
+				return subscriptions.length;
+			};
+		},
+		unsubscribeAll() {
+			subscriptions.splice(0, Infinity);
+		},
+		_fire(val) {
+			return subscriptions.map(s => s(val)).forEach(call);
+		},
+		_forEachSubscription(fn) {
+			subscriptions.forEach(fn);
+		},
+		_hasSubscribers() {
+			return subscriptions.length > 0;
+		}
+	};
+}

src/util.js

@@ -3,20 +3,3 @@ export const id = a => a;
 export const pick = (id, def) => doc => (doc && doc.hasOwnProperty(id) ? doc[id] : def);
 
 export const call = a => (typeof a === 'function' ? a() : a);
-
-// internal utilities
-
-export const registerSubscriptions = subscriptionsArray => fn => {
-	subscriptionsArray.push(fn);
-	return () => {
-		const idx = subscriptionsArray.indexOf(fn);
-		if (idx !== -1) {
-			subscriptionsArray.splice(idx, 1);
-		}
-		return subscriptionsArray.length;
-	};
-};
-
-export const registerFire = subscriptionsArray => val => {
-	subscriptionsArray.map(s => s(val)).forEach(call);
-};