Skip to content

Commit 207fd4c

Browse files
ping-maxwellping-maxwell
committed
Change event publish to return result and completion
1 parent cfd27f3 commit 207fd4c

2 files changed

Lines changed: 81 additions & 44 deletions

File tree

‎packages/experimental/src/event.test.ts‎

Lines changed: 49 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -22,19 +22,24 @@ describe("v.event", () => {
2222
await next();
2323
});
2424

25-
const result = await signUp.publish("emailOpt", {
25+
const [result, complete] = await signUp.publish("emailOpt", {
2626
id: "1",
2727
email: "ada@example.com",
2828
});
29-
expect(result).toEqual({ id: "1", email: "patched@example.com" });
29+
await expect(complete()).resolves.toEqual({
30+
id: "1",
31+
email: "patched@example.com",
32+
});
33+
expect(result).toEqual({ id: "1", email: "ada@example.com" });
3034
expect(seen).toEqual(["emailOpt"]);
3135

3236
unsubscribe();
33-
await signUp.publish("email", {
37+
const [, completeEmail] = await signUp.publish("email", {
3438
id: "2",
3539
name: "Ada",
3640
email: "ada@example.com",
3741
});
42+
await completeEmail();
3843
expect(seen).toEqual(["emailOpt"]);
3944
});
4045

@@ -48,7 +53,12 @@ describe("v.event", () => {
4853
expect(() => (bus as any).publish("pong", { n: 1 })).toThrow(
4954
/unknown event kind/,
5055
);
51-
expect(bus.publish("ping", { n: 1 })).toEqual({ n: 1 });
56+
const [result, complete] = bus.publish("ping", { n: 1 }) as [
57+
{ n: number },
58+
() => Promise<{ n: number }>,
59+
];
60+
expect(result).toEqual({ n: 1 });
61+
return expect(complete()).resolves.toEqual({ n: 1 });
5262
});
5363

5464
it("skips the chain when next is not called (veto)", async () => {
@@ -67,7 +77,8 @@ describe("v.event", () => {
6777
order.push("inner");
6878
await next();
6979
});
70-
const result = await bus.publish("x", { v: 1 });
80+
const [result, complete] = await bus.publish("x", { v: 1 });
81+
await expect(complete()).resolves.toEqual({ v: 1 });
7182
expect(result).toEqual({ v: 1 });
7283
expect(order).toEqual(["outer", "veto"]);
7384
});
@@ -87,8 +98,9 @@ describe("v.event", () => {
8798
order.push("inner");
8899
await next({ v: 2 });
89100
});
90-
const result = await bus.publish("x", { v: 1 });
91-
expect(result).toEqual({ v: 2 });
101+
const [result, complete] = await bus.publish("x", { v: 1 });
102+
await expect(complete()).resolves.toEqual({ v: 2 });
103+
expect(result).toEqual({ v: 1 });
92104
expect(order).toEqual(["outer", "inner"]);
93105
});
94106

@@ -99,7 +111,8 @@ describe("v.event", () => {
99111
bus.subscribe(async (_e, next) => {
100112
await next({ v: "nope" } as never);
101113
});
102-
await expect(bus.publish("x", { v: 1 })).rejects.toThrow(/expected number/);
114+
const [, complete] = await bus.publish("x", { v: 1 });
115+
await expect(complete()).rejects.toThrow(/expected number/);
103116
});
104117

105118
it("does not re-transform untouched fields on next() patches", async () => {
@@ -119,8 +132,9 @@ describe("v.event", () => {
119132
expect(e.data.a).toBe("hi!");
120133
await next({ b: 2 });
121134
});
122-
const result = await bus.publish("x", { a: "hi", b: 1 });
123-
expect(result).toEqual({ a: "hi!", b: 2 });
135+
const [result, complete] = await bus.publish("x", { a: "hi", b: 1 });
136+
await expect(complete()).resolves.toEqual({ a: "hi!", b: 2 });
137+
expect(result).toEqual({ a: "hi!", b: 1 });
124138
expect(transforms).toBe(1);
125139
});
126140
});
@@ -157,13 +171,21 @@ describe("event extension + modules", () => {
157171
// Mounting a fn that uses the module registers event listeners.
158172
v.fn("evt_mod.app", { use: [core, hooks] }, () => "ok");
159173

160-
const fromEmail = await signUp.publish("email", {
174+
const [fromEmail, completeEmail] = await signUp.publish("email", {
175+
id: "1",
176+
email: "a@b.co",
177+
});
178+
await expect(completeEmail()).resolves.toEqual({
161179
id: "1",
162180
email: "a@b.co",
163181
});
164182
expect(fromEmail).toEqual({ id: "1", email: "a@b.co" });
165183

166-
const fromOauth = await withOauth.publish("oauth", {
184+
const [fromOauth, completeOauth] = await withOauth.publish("oauth", {
185+
provider: "github",
186+
id: "42",
187+
});
188+
await expect(completeOauth()).resolves.toEqual({
167189
provider: "github",
168190
id: "42",
169191
});
@@ -206,7 +228,9 @@ describe("event extension + modules", () => {
206228
await next();
207229
});
208230
const f = v.fn({ use: [{ bus }] }, async (c) => {
209-
return c.bus.publish("ping", { n: 7 });
231+
const [result, complete] = await c.bus.publish("ping", { n: 7 });
232+
await complete();
233+
return result;
210234
});
211235
await expect(f()).resolves.toEqual({ n: 7 });
212236
expect(seen).toEqual([7]);
@@ -228,10 +252,12 @@ describe("event extension + modules", () => {
228252
});
229253

230254
const f = v.fn({ use: [{ bus, account, withTag }] }, async (c) => {
231-
const result = c.bus.publish("created", { id: "1", tag: "vip" });
232-
expectTypeOf(result).toEqualTypeOf<
233-
{ id: string; tag: string } | Promise<{ id: string; tag: string }>
234-
>();
255+
const [result, complete] = await c.bus.publish("created", {
256+
id: "1",
257+
tag: "vip",
258+
});
259+
expectTypeOf(result).toEqualTypeOf<{ id: string; tag: string }>();
260+
await complete();
235261
return result;
236262
});
237263
await expect(f()).resolves.toEqual({ id: "1", tag: "vip" });
@@ -251,10 +277,9 @@ describe("event extension + modules", () => {
251277
});
252278
const bus = v.event("evt_var_bare_bus", { created: account });
253279
const f = v.fn({ use: [{ bus, account }] }, async (c) => {
254-
const result = c.bus.publish("created", { id: "1" });
255-
expectTypeOf(result).toEqualTypeOf<
256-
{ id: string } | Promise<{ id: string }>
257-
>();
280+
const [result, complete] = await c.bus.publish("created", { id: "1" });
281+
expectTypeOf(result).toEqualTypeOf<{ id: string }>();
282+
await complete();
258283
return result;
259284
});
260285
await expect(f()).resolves.toEqual({ id: "1" });
@@ -269,9 +294,11 @@ describe("event extension + modules", () => {
269294
created: { account },
270295
});
271296
const f = v.fn({ use: [{ bus, account, withTag }] }, async (c) => {
272-
return c.bus.publish("created", {
297+
const [result, complete] = await c.bus.publish("created", {
273298
account: { id: "2", tag: "gold" },
274299
});
300+
await complete();
301+
return result;
275302
});
276303
await expect(f()).resolves.toEqual({
277304
account: { id: "2", tag: "gold" },

‎packages/experimental/src/event.ts‎

Lines changed: 32 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ export type EventMessage<T> = Prettify<
2121

2222
/** `next(mutate?)` continues the chain and optionally patches this publish's
2323
* payload. Skipping `next` stops the chain (veto), same idea as `v.on`. */
24-
export type EventNext<D> = (mutate?: Partial<D>) => void | Promise<void>;
24+
export type EventNext<D> = (mutate?: Partial<D>) => D | Promise<D>;
2525

2626
export type EventHandler<T> = (
2727
event: EventMessage<T>,
@@ -48,13 +48,17 @@ export interface EventDefination<
4848
subscribe: (handler: EventHandler<T>) => () => void;
4949
/**
5050
* Validate `data` against the kind's schema, run subscribers (direct +
51-
* any mounted via `v.on` / modules), merge `next` mutations, return the
52-
* final payload.
51+
* any mounted via `v.on` / modules), merge `next` mutations, and return:
52+
* - `result`: the validated payload at publish-time
53+
* - `complete`: a promise function resolving to the final payload after
54+
* the full subscriber chain finishes
5355
*/
5456
publish: <K extends keyof T & string>(
5557
type: K,
5658
data: EventPayloads<T>[K],
57-
) => EventPayloads<T>[K] | Promise<EventPayloads<T>[K]>;
59+
) =>
60+
| [EventPayloads<T>[K], () => Promise<EventPayloads<T>[K]>]
61+
| Promise<[EventPayloads<T>[K], () => Promise<EventPayloads<T>[K]>]>;
5862
/**
5963
* Mint a NEW event def under the same name with more kinds - the
6064
* re-export pattern (`customize` for vars). Shared bus; widened types.
@@ -202,17 +206,13 @@ const runHandlers = (
202206
schema: unknown,
203207
path: string,
204208
): unknown | Promise<unknown> => {
205-
let current = initial;
206-
const run = (i: number): void | Promise<void> => {
207-
if (i >= handlers.length) return;
209+
const run = (i: number, current: unknown): unknown | Promise<unknown> => {
210+
if (i >= handlers.length) return current;
208211
let called = false;
209-
let downstream: void | Promise<void>;
212+
let downstream: unknown | Promise<unknown> = current;
210213
const next = (mutate?: Partial<unknown>) => {
211214
called = true;
212-
const continueChain = (value: unknown) => {
213-
current = value;
214-
return run(i + 1);
215-
};
215+
const continueChain = (value: unknown) => run(i + 1, value);
216216
if (
217217
mutate !== undefined &&
218218
mutate !== null &&
@@ -222,20 +222,20 @@ const runHandlers = (
222222
applyPatch(schema, current, mutate as Record<string, unknown>, path),
223223
continueChain,
224224
);
225-
} else {
226-
downstream = continueChain(current);
225+
return downstream;
227226
}
227+
downstream = continueChain(current);
228228
return downstream;
229229
};
230230
const handler = handlers[i];
231-
if (!handler) return;
231+
if (!handler) return current;
232232
const result = handler({ type, data: current }, next);
233233
return thenMaybe(result, () => {
234-
if (!called) return;
234+
if (!called) return current;
235235
return downstream;
236236
});
237237
};
238-
return thenMaybe(run(0), () => current);
238+
return run(0, initial);
239239
};
240240

241241
/** A var extension's extra fields, applied to event payloads that infer
@@ -296,7 +296,9 @@ const publishOn = (
296296
type: string,
297297
data: unknown,
298298
varExts: readonly EventVarExt[] = [],
299-
): unknown | Promise<unknown> => {
299+
):
300+
| [unknown, () => Promise<unknown>]
301+
| Promise<[unknown, () => Promise<unknown>]> => {
300302
const bus = getBus(name);
301303
const schema = bus.types[type];
302304
const path = `event.${name}.${type}`;
@@ -305,9 +307,14 @@ const publishOn = (
305307
}
306308
const effective = applyVarExtsToSchema(schema, varExts);
307309
const handlers = [...bus.mounted, ...bus.direct];
308-
return thenMaybe(validate(asType(effective), data, path), (parsed) =>
309-
runHandlers(handlers, type, parsed, effective, path),
310-
);
310+
return thenMaybe(validate(asType(effective), data, path), (parsed) => {
311+
const done = runHandlers(handlers, type, parsed, effective, path);
312+
return [
313+
parsed,
314+
() =>
315+
isThenable(done) ? (done as Promise<unknown>) : Promise.resolve(done),
316+
] as [unknown, () => Promise<unknown>];
317+
});
311318
};
312319

313320
/** Publish against a named bus, folding mounted var extensions into
@@ -319,7 +326,10 @@ export const publishEvent = (
319326
type: string,
320327
data: unknown,
321328
varExts: readonly EventVarExt[] = [],
322-
): unknown | Promise<unknown> => publishOn(name, type, data, varExts);
329+
):
330+
| [unknown, () => Promise<unknown>]
331+
| Promise<[unknown, () => Promise<unknown>]> =>
332+
publishOn(name, type, data, varExts);
323333

324334
/** Register a module-mounted event listener (no-op if already present). */
325335
export const mountEventOn = (entry: EventOnEntry<string>) => {

0 commit comments

Comments
 (0)