Document less-known APIs and fix SubscriptionFunc buffer leak · TeaEntityLab/fpRust@40ef803 · GitHub
Skip to content

Commit 40ef803

Browse files
johnteeedevin-ai-integration[bot]
johnteee
and
devin-ai-integration[bot]
committed
Document less-known APIs and fix SubscriptionFunc buffer leak
Add docs/PROJECT_NOTES.md cataloging the project origins and leaked public APIs (observe_on/subscribe_on, subscribe_blocking_queue, do_m_pattern!, coroutine macros), with a README section pointing to it. Add runnable examples for scheduler routing (examples/scheduler.rs) and the push-to-pull bridge (examples/publisher_queue.rs), and document the sync/async yield_from deadlock invariant in examples/cor.rs. Fix SubscriptionFunc buffering every value forever (memory leak) by starting its cached stream closed and only buffering once as_stream() opens it. Correct the misleading BlockingQueue::stop docs/dead-code that claimed to drop the sender (it dropped the MutexGuard, not the Sender behind Arc<Mutex<..>>). Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
1 parent 8e6b302 commit 40ef803

8 files changed

Lines changed: 328 additions & 18 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 4 additions & 0 deletions

‎README.md‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,16 @@ Module-scoped flags (`fp`, `maybe`, `sync`, `cor`, `actor`, `handler`, `monadio`
4343
- **Actor** (`fp_rust::actor`) — lightweight actor model with `Context`, parent/child messaging, and ask-style patterns
4444
- **Maybe** (`fp_rust::maybe`) — optional/monad helpers
4545

46+
### Less-known / advanced capabilities
47+
48+
Some public APIs are easy to miss. See [docs/PROJECT_NOTES.md](docs/PROJECT_NOTES.md) for the full catalog, origins, and scenarios.
49+
50+
- **Scheduler routing** — `MonadIO::observe_on` / `subscribe_on` (RxJava-style thread-hopping). **Trap:** with no `observe_on`, `subscribe_on` is a silent no-op and the effect runs on the caller. Set `observe_on` first.
51+
- **Push→pull bridge** — `Publisher::as_blocking_queue` / `subscribe_blocking_queue` forward published values into a `BlockingQueue` for deterministic pulling; `Publisher::subscribe_on` delivers off-thread.
52+
- **Sync↔async bridge** — `BlockingQueue::{take,poll}_result_as_future` (`for_futures`) `.await` a blocking queue.
53+
- **Pattern do-notation** — `do_m_pattern!` extends `do_m!` with typed `let`, reassignment, `exec`, and `ret`.
54+
- **Utility macros** — `map_insert!` (bulk `HashMap` fill), `cor_newmutex_and_start!` / `cor_yield!` (coroutine building), `contains!`, `reverse!`.
55+
4656
### Deferred / non-goals
4757

4858
**Pattern matching** macros or DSL support are **not planned** for the current roadmap. They are explicitly deferred, not silently removed features.
@@ -80,11 +90,13 @@ cargo run --example <name> --features=test_runtime
8090
|---------|----------------|
8191
| `monadio` | `MonadIO` sync/async subscribe, handler scheduling |
8292
| `publisher` | `Publisher` pub/sub with handlers |
83-
| `cor` | `Cor` yield / yield_from and async/sync setup |
93+
| `cor` | `Cor` yield / yield_from and the sync/async deadlock invariant |
8494
| `do_notation` | `do_m!` do-notation over coroutines |
8595
| `fp` | `compose!`, `pipe!`, `map!` / `reduce!` / `filter!` pipelines |
8696
| `actor` | Actor spawn, parent/child messaging, shutdown messages |
8797
| `actor_ask` | Ask-style replies via `LinkedListAsync` and `BlockingQueue` |
98+
| `scheduler` | `MonadIO` `observe_on` / `subscribe_on` thread-hopping (and the `subscribe_on`-alone no-op trap) |
99+
| `publisher_queue` | `Publisher::as_blocking_queue` push→pull bridge |
88100

89101
To run the full test suite with futures enabled:
90102

@@ -97,6 +109,7 @@ cargo test --features=test_runtime
97109
- API reference: [docs.rs/fp_rust](https://docs.rs/fp_rust/)
98110
- Changelog: [CHANGELOG.md](CHANGELOG.md)
99111
- Roadmap and backlog: [ROADMAP.md](ROADMAP.md)
112+
- Project notes (origins, lineage, less-known APIs): [docs/PROJECT_NOTES.md](docs/PROJECT_NOTES.md)
100113

101114
## License
102115

‎docs/PROJECT_NOTES.md‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
# fpRust — Project Notes: Origins, Identity & Less-Known Capabilities
2+
3+
> Durable engineering notes derived from a git-history + API-surface investigation
4+
> (204 commits, 2018‑07 → 2026‑07). This document records *what kind of project this
5+
> is*, *why it is shaped the way it is*, and the *public capabilities that are easy to
6+
> miss*. It is context for maintainers and advanced users; the [README](../README.md)
7+
> remains the primary user guide and [ROADMAP](../ROADMAP.md) the maintenance posture.
8+
9+
## Identity
10+
11+
fpRust is an **educational, polyglot functional-programming / reactive pattern
12+
library** — the Rust member of a tri-language family alongside **`fpGo`** (Go) and
13+
**`fpEs`** (ECMAScript). Its APIs deliberately mirror those siblings (`src/maybe.rs`
14+
documents the `map`/`chain`/`ap` parity), so the same FP/reactive concepts read the
15+
same way across Go, JavaScript, and Rust.
16+
17+
It is **not** a production async runtime or a Tokio/Actix competitor. It optimizes for
18+
**clarity and teachability** of cross-language patterns, and is a good fit for
19+
*learning & prototyping* and *selective module use* — not *production orchestration*
20+
(no unified shutdown protocol; see [ROADMAP](../ROADMAP.md) exclusions).
21+
22+
## Origin story (three eras)
23+
24+
| Era | Window | What landed | Influences |
25+
|-----|--------|-------------|------------|
26+
| **1 — Sync FP + threading core** | 2018‑07 (142 commits) | `Maybe` → `MonadIO` → `Handler`/`HandlerThread` → `BlockingQueue` → `Publisher`/`SubscriptionFunc` → `CountDownLatch` → `Cor` → fp macros (`map!`/`reduce!`/`filter!`/`fold*!`/`compose!`/`pipe!`) → `do_m!` | Haskell (monads, do‑notation), Android (`Handler` loop), Java `java.util.concurrent`, RxJava schedulers, Go coroutines (Cor "from golang version"), GoF Observer (via `eliovir/rust-examples`), `dtolnay/reduce` |
27+
| **2 — Java Future** | 2018‑08 (3 commits) | `WillAsync` ("to achieve JavaFuture") | Java `Future` callback/wait semantics |
28+
| **3 — Async Rust + Actors** | 2021‑06→08 (56 commits) | Rust‑2021 modernization (only external PR ever, #1 `bogdad/fix_compilation`), `futures::Stream` + `Future` integration, `subscribe_as_stream`, real async `CountDownLatch`, `BlockingQueue::{take,poll}_result_as_future`, **Akka‑style `ActorAsync`** (ask/tell + children), `LinkedListAsync`, feature‑flag modularization |
29+
30+
(2026‑07 added test coverage, docs/roadmap, and async‑waker correctness fixes — see [CHANGELOG](../CHANGELOG.md).)
31+
32+
### On the word "leaked"
33+
The 2018‑07‑11 commit **"Add leaked observe/subscribe_on"** is the author's own term
34+
for **capabilities shipped ahead of their documentation** — public, working, but never
35+
surfaced to users. This document is largely about paying down that gap.
36+
37+
## Less-known / "leaked" capabilities
38+
39+
These are fully public and working but absent from the README's main capability list.
40+
Ranked by user value; every claim traced to source.
41+
42+
### 1. `MonadIO::observe_on` / `subscribe_on` — scheduler routing (with a trap)
43+
`src/monadio.rs:105,110`. Inspired by RxJava's `observeOn`/`subscribeOn`, but the
44+
semantics **differ** — verified at `src/monadio.rs:141‑163`:
45+
46+
- `ob_handler == None` → the effect runs **synchronously on the calling thread** and
47+
**`subscribe_on` is ignored entirely**. So `subscribe_on` *alone* is a silent no‑op.
48+
- `ob_handler == Some` → the effect is posted to the observe handler, which then
49+
forwards effect execution **and** the subscriber's `on_next` to the subscribe
50+
handler thread (`Some`) or runs them on the observe thread (`None`).
51+
52+
**Rule:** if you want thread‑hopping, always set `observe_on` first; `subscribe_on`
53+
only takes effect when `observe_on` is set.
54+
55+
### 2. `Publisher::subscribe_blocking_queue` / `as_blocking_queue` — push→pull bridge
56+
`src/publisher.rs:99,107`. Turns the push‑based pub/sub stream into a **pull‑based
57+
`BlockingQueue`**: published values are forwarded into a queue a consumer thread drains
58+
on its own schedule (`queue.take()` / `take_result_as_future().await`). Ideal when the
59+
consumer wants to pull deterministically instead of running in a callback. Note: it is
60+
a delivery bridge, not a bounded backpressure mechanism — see the queue's own limits.
61+
62+
### 3. `Publisher::subscribe_on` — off-thread delivery
63+
`src/publisher.rs:58`. Unlike MonadIO's, this one behaves straightforwardly: with a
64+
handler set, `notify_observers` posts each delivery to that handler thread
65+
(`src/publisher.rs` notify path); with `None`, delivery is synchronous on the publish
66+
caller. Safe to use on its own.
67+
68+
### 4. `BlockingQueue::{take,poll}_result_as_future` — sync↔async bridge
69+
`src/sync.rs`. `for_futures` only. Offloads the blocking `recv` onto the shared futures
70+
thread pool so a synchronous `BlockingQueue` can be `.await`ed from async code.
71+
72+
### 5. `do_m_pattern!` — pattern-matching do-notation
73+
`src/cor.rs:51`. A richer `do_m!`: supports typed `let`, `let mut`, reassignment,
74+
`exec` side-effect statements, `yield_from` of coroutines, and a final `ret`. Live and
75+
tested (`test_cor_do_m_pattern*`, cor.rs:470/578/590) — **not** dead code.
76+
77+
### 6. Utility macros absent from the README
78+
- `map_insert!(m, { k: v, ... })` / `[ k => v, ... ]` (`src/common.rs`) — bulk
79+
`HashMap` population in a language without map literals (several syntaxes).
80+
- `cor_newmutex!`, `cor_newmutex_and_start!`, `cor_yield!` (`src/cor.rs`) — the actual
81+
macros for building/driving coroutines.
82+
- `contains!`, `reverse!` (`src/fp.rs`) — curried `Vec` helpers.
83+
- `spread_and_call!`, `partial_left_last_one!` (`src/fp.rs`) — currying **plumbing**
84+
behind the fp macros; internal, not intended as user entry points.
85+
86+
## Scenarios & better ways to use
87+
88+
### MonadIO — deferred/lazy IO with scheduler hopping
89+
*When:* defer a computation and run it off-thread, delivering the result on another
90+
thread. *Better way:* set **both** `observe_on` and `subscribe_on` (see the trap in
91+
§1); start both handlers before `subscribe`. Runnable: `examples/scheduler.rs`
92+
(proves the routing and the `subscribe_on`-alone no-op by asserting `ThreadId`);
93+
`examples/monadio.rs` shows the basic sync/async subscribe.
94+
95+
### Publisher — Rx pub/sub, three consumption styles
96+
Callback (`subscribe_fn`), **pull** (`as_blocking_queue` → `take`), or **stream**
97+
(`subscribe_as_stream` → `collect().await`). Choose pull when the consumer sets the
98+
pace; stream when composing with `futures`. Runnable: `examples/publisher_queue.rs`
99+
(push→pull bridge).
100+
101+
### Cor — coroutines & do-notation for sequencing
102+
*When:* sequence dependent steps without nesting. Use `do_m!`/`do_m_pattern!`.
103+
**Gotcha (verified invariant):** a `yield_from` target must be **async**
104+
(`set_async(true)`, the default for `cor_newmutex_and_start!`). Two coroutines that
105+
yield to each other while both sync **deadlock** — keep the entry coroutine sync and
106+
targets async (see README "Cor sync deadlock invariant" and `src/cor.rs`).
107+
108+
### Actor — Akka-style ask/tell with children
109+
`examples/actor.rs` (spawn + child lifecycle) and `examples/actor_ask.rs` (reply via a
110+
`BlockingQueue`, using a take-timeout, not sleep). Shutdown is per-actor and manual —
111+
do not assume structured teardown.
112+
113+
### Anti-patterns (grounded in this codebase)
114+
1. **`thread::sleep` for synchronization** → use `CountDownLatch::wait()`, a
115+
`BlockingQueue` take, or `to_future().await`. (This session removed 7 such sleeps;
116+
examples are sleep-free.)
117+
2. **`subscribe_on` without `observe_on` in MonadIO** → silent no-op (§1).
118+
3. **Sync `yield_from` target** → deadlock; keep targets async.
119+
120+
## Cross-language family `[INFERENCE]`
121+
`fpGo` (Go) and `fpEs` (ECMAScript) are the sibling ports; `Cor` was ported "from
122+
golang version". Specific fpGo/fpEs API details are **inferred** from this repo's
123+
comments/commits and are not verified against those repositories here.
124+
125+
## Provenance (attributions)
126+
Haskell (monads, do-notation) · Java `java.util.concurrent` (`BlockingQueue`,
127+
`CountDownLatch`, `Will`/future) · Android `Handler` loop (`HandlerThread`) · RxJava
128+
schedulers (`observe_on`/`subscribe_on`) · Go coroutines (`Cor`) · Akka/Erlang
129+
(`ActorAsync` ask) · `dtolnay/reduce` (`Reduce`) · `eliovir/rust-examples` (Observer).
130+
The `compose!`/`pipe!` macros were reimplemented from first principles this session to
131+
remove a CC BY-SA snippet. See [CONTRIBUTING](../CONTRIBUTING.md#source-credit-and-provenance).

‎examples/cor.rs‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,13 @@
1+
// Cor coroutines: `yield` / `yield_from` with the sync/async deadlock invariant.
2+
//
3+
// INVARIANT (see src/cor.rs): a `yield_from` TARGET must be async. Two
4+
// coroutines that yield to each other while BOTH sync deadlock — each blocks
5+
// its thread waiting for the other. The safe split, shown below:
6+
// - the yield TARGET (cor1) is async -> set_async(true)
7+
// - the ENTRY driver (cor2) stays sync -> set_async(false)
8+
//
9+
// Run: cargo run --example cor --features=test_runtime
10+
111
use std::thread;
212
use std::time::{Duration, Instant};
313

@@ -33,11 +43,11 @@ fn main() {
3343
);
3444

3545
{
36-
let cor1 = cor1.clone();
46+
// Target of yield_from: MUST be async, or the pair deadlocks.
3747
cor1.lock().unwrap().set_async(true);
3848
}
3949
{
40-
let cor2 = cor2.clone();
50+
// Entry driver: sync is fine because its target (cor1) is async.
4151
cor2.lock().unwrap().set_async(false);
4252
}
4353
cor_start!(cor1);

‎examples/publisher_queue.rs‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
// Publisher push -> pull bridge: `as_blocking_queue` / `subscribe_blocking_queue`.
2+
//
3+
// Turns a push-based pub/sub stream into a pull-based `BlockingQueue` so a
4+
// consumer drains values on its own schedule instead of running in a callback.
5+
// Note: this is a delivery bridge, not bounded backpressure (unbounded buffer).
6+
//
7+
// Run: cargo run --example publisher_queue --features=test_runtime
8+
9+
use std::sync::Arc;
10+
use std::thread;
11+
12+
use fp_rust::publisher::Publisher;
13+
use fp_rust::sync::Queue;
14+
15+
fn main() {
16+
let mut publisher = Publisher::new();
17+
18+
// Bridge the stream into a pull-based queue. `subscription` keeps delivery
19+
// alive; dropping it would unsubscribe.
20+
let (_subscription, queue) = publisher.as_blocking_queue();
21+
22+
// Consumer pulls on its own thread; take() blocks until an item arrives.
23+
let mut consumer_queue = queue.clone();
24+
let consumer = thread::spawn(move || {
25+
let mut received = Vec::new();
26+
for _ in 0..3 {
27+
let v: Arc<i32> = consumer_queue.take().expect("value");
28+
println!("consumer pulled {:?}", v);
29+
received.push(*v);
30+
}
31+
received
32+
});
33+
34+
// Producer publishes at its own pace; values buffer in the queue until
35+
// pulled. No handler => delivery is synchronous on this publish() call.
36+
for i in 1..=3 {
37+
publisher.publish(i);
38+
}
39+
40+
let received = consumer.join().expect("consumer thread");
41+
assert_eq!(vec![1, 2, 3], received);
42+
println!("publisher_queue: pulled {:?} in order", received);
43+
}

‎examples/scheduler.rs‎

Lines changed: 70 additions & 0 deletions

0 commit comments

Comments
 (0)