[READ-ONLY] Mirror of https://github.com/bombshell-dev/playground.
✨ add stdin/input Effection helpers to demo
Charles Lowell
(Jun 26, 2026, 11:00 AM +0300)
c20e9f3a
c0880c1d
+112
-1
+2
-1
packages/demo/package.json
+2
-1
packages/demo/package.json
+70
packages/demo/src/use-input.ts
+70
packages/demo/src/use-input.ts
···
1
+
import {
2
+
call,
3
+
createChannel,
4
+
each,
5
+
type Operation,
6
+
race,
7
+
resource,
8
+
sleep,
9
+
spawn,
10
+
type Stream,
11
+
suspend,
12
+
until,
13
+
} from "effection";
14
+
import {
15
+
createInput,
16
+
type InputEvent,
17
+
type InputOptions,
18
+
} from "@bomb.sh/tty";
19
+
20
+
function nothing() {
21
+
return suspend() as unknown as Operation<
22
+
IteratorResult<Uint8Array, void>
23
+
>;
24
+
}
25
+
26
+
// Parse a raw byte Stream into a Stream of decoded terminal InputEvents.
27
+
export function useInput(
28
+
stream: Stream<Uint8Array, void>,
29
+
options?: InputOptions,
30
+
): Stream<InputEvent, void> {
31
+
return resource(function* (provide) {
32
+
const input = yield* until(createInput(options));
33
+
const subscription = yield* stream;
34
+
35
+
let pending = nothing();
36
+
37
+
const events = createChannel<InputEvent, void>();
38
+
39
+
yield* spawn(function* () {
40
+
let next = yield* subscription.next();
41
+
while (!next.done) {
42
+
const result = input.scan(next.value);
43
+
pending = result.pending ? rescan(result.pending.delay) : nothing();
44
+
for (const event of result.events) {
45
+
yield* events.send(event);
46
+
}
47
+
next = yield* race([subscription.next(), pending]);
48
+
}
49
+
yield* events.close();
50
+
});
51
+
52
+
yield* race([provide(yield* events), drain(events)]);
53
+
});
54
+
}
55
+
56
+
function rescan(delay: number): ReturnType<typeof nothing> {
57
+
return call(function* (): Operation<IteratorResult<Uint8Array, void>> {
58
+
yield* sleep(delay);
59
+
return {
60
+
done: false,
61
+
value: new Uint8Array(),
62
+
};
63
+
});
64
+
}
65
+
66
+
function* drain<T, TClose>(stream: Stream<T, TClose>): Operation<void> {
67
+
for (const _ of yield* each(stream)) {
68
+
yield* each.next();
69
+
}
70
+
}
+37
packages/demo/src/use-stdin.ts
+37
packages/demo/src/use-stdin.ts
···
1
+
import {
2
+
createChannel,
3
+
each,
4
+
type Operation,
5
+
race,
6
+
resource,
7
+
spawn,
8
+
type Stream,
9
+
until,
10
+
} from "effection";
11
+
import { stdin } from "node:process";
12
+
13
+
// Bridge Node's process.stdin (raw bytes) into an Effection Stream.
14
+
export function useStdin(): Operation<Stream<Uint8Array, void>> {
15
+
return resource(function* (provide) {
16
+
const channel = createChannel<Uint8Array, void>();
17
+
18
+
const iterator = stdin[Symbol.asyncIterator]();
19
+
20
+
yield* spawn(function* () {
21
+
let next = yield* until(iterator.next());
22
+
while (!next.done) {
23
+
yield* channel.send(next.value);
24
+
next = yield* until(iterator.next());
25
+
}
26
+
yield* channel.close();
27
+
});
28
+
29
+
yield* race([provide(channel), drain(channel)]);
30
+
});
31
+
}
32
+
33
+
function* drain<T, TClose>(stream: Stream<T, TClose>): Operation<void> {
34
+
for (const _ of yield* each(stream)) {
35
+
yield* each.next();
36
+
}
37
+
}