183 lines
5.6 KiB
Markdown
183 lines
5.6 KiB
Markdown
# Streaming Rdwr: A 9P Pattern for Filtered Subscriptions
|
||
|
||
This document describes a novel 9P pattern used by Ollie for filtered event subscriptions. The pattern extends the standard rdwr (read-write) file semantics to support streaming reads after an initial write.
|
||
|
||
## The Problem
|
||
|
||
Traditional pub/sub systems require:
|
||
1. A subscription API (connect, subscribe to topic, receive messages)
|
||
2. Multiple protocol operations
|
||
3. Separate endpoints for different subscription types
|
||
|
||
9P provides a simple file interface, but standard file semantics don't naturally support "subscribe to X, then stream matching events."
|
||
|
||
## The Pattern: Streaming Rdwr
|
||
|
||
A single file supports both unfiltered and filtered streaming through the write-then-read pattern:
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────────────────┐
|
||
│ event file │
|
||
├─────────────────────────────────────────────────────────────┤
|
||
│ Read-only open: Stream all events │
|
||
│ Read-write open: Write filter → Stream matching events │
|
||
└─────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
### Protocol Flow
|
||
|
||
**Unfiltered (read-only):**
|
||
```
|
||
Topen fid, OREAD
|
||
Tread fid, ... → blocks until event, returns "topic payload\n"
|
||
Tread fid, ... → blocks until event, returns "topic payload\n"
|
||
...
|
||
Tclunk fid
|
||
```
|
||
|
||
**Filtered (read-write):**
|
||
```
|
||
Topen fid, ORDWR
|
||
Twrite fid, "session.*.agent.*.state" → stores filter in per-fid state
|
||
Tread fid, ... → blocks until matching event, returns "topic payload\n"
|
||
Tread fid, ... → blocks until matching event, returns "topic payload\n"
|
||
...
|
||
Tclunk fid → cancels subscription, frees state
|
||
```
|
||
|
||
### Key Properties
|
||
|
||
1. **Single file, dual mode** — Same path serves both use cases
|
||
2. **Per-fid state** — Each open file descriptor maintains its own filter and subscription
|
||
3. **Lazy subscription** — Subscription is created on first read after write, not on write
|
||
4. **Clean teardown** — Clunk cancels the subscription automatically
|
||
5. **Standard 9P semantics** — No protocol extensions required
|
||
|
||
## Filter Syntax
|
||
|
||
Filters use a simple glob-like syntax:
|
||
|
||
| Pattern | Matches |
|
||
|---------|---------|
|
||
| `*` | Exactly one segment |
|
||
| `>` | One or more segments (must be last) |
|
||
| `literal` | Exact match |
|
||
|
||
Examples:
|
||
```
|
||
* # All events
|
||
session.*.agent.*.state # State changes for any agent in any session
|
||
session.abc123.> # All events for session abc123
|
||
session.*.bypass.> # All bypass events
|
||
```
|
||
|
||
## Implementation
|
||
|
||
### Server State
|
||
|
||
Per-fid structure includes subscription state:
|
||
|
||
```go
|
||
type fid struct {
|
||
// ... standard fid fields ...
|
||
|
||
// Event stream subscription
|
||
eventFilter string
|
||
eventCh <-chan Event
|
||
eventCancel context.CancelFunc
|
||
}
|
||
```
|
||
|
||
### Write Handler
|
||
|
||
```go
|
||
if path == "event" {
|
||
filter := strings.TrimSpace(string(data))
|
||
f.eventFilter = filter
|
||
// Clear existing subscription (recreated on next read)
|
||
if f.eventCancel != nil {
|
||
f.eventCancel()
|
||
}
|
||
f.eventCh = nil
|
||
return success
|
||
}
|
||
```
|
||
|
||
### Read Handler
|
||
|
||
```go
|
||
if path == "event" && f.eventFilter != "" {
|
||
if f.eventCh == nil {
|
||
// Create filtered subscription on first read
|
||
ctx, cancel := context.WithCancel(ctx)
|
||
f.eventCh = SubscribeEventsFiltered(ctx, f.eventFilter)
|
||
f.eventCancel = cancel
|
||
}
|
||
// Block until matching event
|
||
ev := <-f.eventCh
|
||
return formatEvent(ev)
|
||
}
|
||
// Fall through to default stream handling
|
||
```
|
||
|
||
### Clunk Handler
|
||
|
||
```go
|
||
if f.eventCancel != nil {
|
||
f.eventCancel()
|
||
}
|
||
```
|
||
|
||
## Client Usage
|
||
|
||
### Shell (ollie-9p)
|
||
|
||
```bash
|
||
# All events
|
||
ollie-9p read event
|
||
|
||
# Filtered events (rdwrs = streaming rdwr)
|
||
echo "session.*.agent.*.state" | ollie-9p rdwrs event
|
||
```
|
||
|
||
### Go
|
||
|
||
```go
|
||
fid, _ := fsys.Open("event", plan9.ORDWR)
|
||
fid.Write([]byte("session.*.agent.*.state"))
|
||
fid.Seek(0, 0)
|
||
for {
|
||
n, _ := fid.Read(buf)
|
||
fmt.Print(string(buf[:n]))
|
||
}
|
||
```
|
||
|
||
### C++/Qt (GUI)
|
||
|
||
The GUI uses the unfiltered stream for the global event bus, handling filtering client-side for flexibility.
|
||
|
||
## Comparison with Alternatives
|
||
|
||
| Approach | Pros | Cons |
|
||
|----------|------|------|
|
||
| **Separate files per filter** | Simple read semantics | Explosion of files, static filters |
|
||
| **Query parameter in path** | RESTful | Non-standard 9P, complex paths |
|
||
| **Custom protocol extension** | Full control | Breaks 9P compatibility |
|
||
| **Streaming rdwr** | Standard 9P, dynamic filters, single file | Requires per-fid state |
|
||
|
||
## Prior Art
|
||
|
||
- **Plan 9 /dev/cons** — Multiplexed per-process I/O through single file
|
||
- **9P2000.u extensions** — Per-fid state for Unix compatibility
|
||
- **NATS subscriptions** — Topic filtering with wildcards
|
||
|
||
The streaming rdwr pattern combines these ideas: per-fid state from 9P2000.u, wildcard filtering from NATS, and the "write configuration, read results" pattern common in Plan 9 control files.
|
||
|
||
## Limitations
|
||
|
||
1. **No replay** — Missed events are lost; no persistent queue
|
||
2. **Single filter per fid** — Rewriting filter resets subscription
|
||
3. **Server-side filtering** — CPU cost scales with event rate × subscribers
|
||
|
||
For high-throughput scenarios, consider client-side filtering on the unfiltered stream.
|