Effects API experiments

lib/agent.spec.ts

@@ -2,7 +2,7 @@ import { expect, console } from '~/test-utils';
 import { Agent } from './agent';
 import { Task, NoAction } from './task';
 import { Sensor, Subscriber } from './sensor';
-import { Observable } from './observable';
+import { IO, fromPipe, when } from './effects';
 
 import { setTimeout } from 'timers/promises';
 
@@ -24,7 +24,7 @@ describe('Agent', () => {
 			});
 			agent.seek({ never: true });
 			await expect(agent.wait(1000)).to.be.rejected;
-			await agent.stop();
+			agent.stop();
 		});
 
 		it('it continues looking for plan unless max retries is set', async () => {
@@ -64,18 +64,29 @@ describe('Agent', () => {
 			expect(count).to.deep.equal([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
 		});
 
-		it('it allows to use observables as actions', async () => {
+		it('allows to use observables as actions', async () => {
 			const counter = Task.of({
-				condition: (state: number, { target }) => state < target,
-				effect: (_: number, { target }) => target,
-				action: (state: number, { target }) =>
-					Observable.of(async (s) => {
-						while (state < target) {
-							state = state + 1;
-							s.next(state);
-							await setTimeout(10);
-						}
-					}),
+				effect: (state: number, { target }) =>
+					fromPipe(
+						state,
+						when(
+							(s) => s < target,
+							(s) =>
+								// this IO call uses a generator function as the async
+								// side, to produce multiple values while the computation is performed
+								// The sync side just tells us that the effect of the computation is that
+								// the counter reaches the target
+								IO(
+									async function* () {
+										while (s < target) {
+											yield ++s;
+											await setTimeout(10);
+										}
+									},
+									() => target,
+								),
+						),
+					),
 			});
 			const agent = Agent.of({
 				initial: 0,

lib/agent/index.ts

@@ -1,6 +1,6 @@
 import assert from '../assert';
 import { NullLogger } from '../logger';
-import { Observable, Subject } from '../observable';
+import { Subscribable, Subject } from '../observable';
 import { Planner } from '../planner';
 import { Sensor } from '../sensor';
 import { Target } from '../target';
@@ -10,7 +10,7 @@ import { AgentOpts, NotStarted, Result } from './types';
 
 export * from './types';
 
-export interface Agent<TState = any> extends Observable<TState> {
+export interface Agent<TState = any> extends Subscribable<TState> {
 	/**
 	 * Tells the agent to seek a new target.
 	 *

lib/agent/runtime.ts

@@ -1,7 +1,7 @@
 import { setTimeout as delay } from 'timers/promises';
 import { diff, patch, Operation as PatchOperation } from 'mahler-wasm';
 
-import { Observer, Observable } from '../observable';
+import { Observer } from '../observable';
 import { Planner, Node, EmptyNode } from '../planner';
 import { Sensor, Subscription } from '../sensor';
 import { Target } from '../target';
@@ -100,34 +100,29 @@ export class Runtime<TState> {
 			// what we need to compare the updated state to
 			const before = this.state;
 			const res = action(before);
-			if (Observable.is<TState>(res)) {
-				const runtime = this;
-				// If the action result is an observable, then
-				// we need to subscribe to it and update the internal
-				// state as the observable emits new values
-				return new Promise((resolve, reject) => {
-					res.subscribe({
-						next(s) {
-							const changes = diff(before, s);
-							if (changes.length > 0) {
-								runtime.state = patch(runtime.state, changes);
-								runtime.observer.next(runtime.state);
-							}
-						},
-						complete() {
-							// There should be no more changes to perform
-							// here
-							resolve([]);
-						},
-						error(e) {
-							reject(e);
-						},
-					});
+			const runtime = this;
+			// If the action result is an observable, then
+			// we need to subscribe to it and update the internal
+			// state as the observable emits new values
+			return new Promise((resolve, reject) => {
+				res.subscribe({
+					next(s) {
+						const changes = diff(before, s);
+						if (changes.length > 0) {
+							runtime.state = patch(runtime.state, changes);
+							runtime.observer.next(runtime.state);
+						}
+					},
+					complete() {
+						// There should be no more changes to perform
+						// here
+						resolve([]);
+					},
+					error(e) {
+						reject(e);
+					},
 				});
-			} else {
-				const after = await res;
-				return diff(before, after);
-			}
+			});
 		} catch (e) {
 			throw new ActionRunFailed(action, e);
 		}