pico
created pr with
103.1
added 103.2
1: 66cf9b9 = 1: 66cf9b9 feat(pipe): add pipe_monitors table
2: 83b0c70 = 2: 83b0c70 chore(pipe): add new db methods
3: 43aa4a1 = 3: 43aa4a1 chore(pipe): add db impl
4: a75186b = 4: a75186b chore(pipe): add tests for monitoring
5: 9c7fcda = 5: 9c7fcda feat(pipe): add monitor cli
6: 2b86dec = 6: 2b86dec feat(pipe): status and rss commands
7: 8df2aa2 = 7: 8df2aa2 feat(pipe): monitor pub and pipe cmd
8: 11a5bce = 8: 11a5bce chore(pipe): monitor help text
9: 670350c = 9: 670350c refactor(pipe): monitor pipes on throttle interval
10: ade2b83 = 10: ade2b83 fix: window and ping fixes
11: 68f5cd1 = 11: 68f5cd1 chore(pipe): add tests
-: ------- > 12: 74fb40f chore(pipe): add logging stmts
-: ------- > 13: c422818 refactor(pipe): create pipe monitors history table
-: ------- > 14: e0c3a88 fix(pipe): only update monitor if within window
-: ------- > 15: 9a8fbde chore(pipe): implement monitor history db interface
-: ------- > 16: f168507 feat(pipe): record historical monitors
-: ------- > 17: 16df77f feat(pipe): monitor calculate uptime
-: ------- > 18: 10ffdf1 feat(pipe): uptime cli command
-: ------- > 19: 8231ac5 chore(pipe): monitor history stubs
-: ------- > 20: 65a63a9 fix(pipe): uptime cmd
added 103.3
3: 43aa4a1 ! 1: 603ca6e chore(pubsub): add more tests
1: 66cf9b9 < -: ------- feat(pipe): add pipe_monitors table
2: 83b0c70 < -: ------- chore(pipe): add new db methods
4: a75186b ! 2: 17e00b2 feat(pubsub): round robin
13: c422818 ! 3: e3136bd fix(pubsub): check for eof before processing and skip empty byte reads
-: ------- > 4: 9a6d19e fix: rr
5: 9c7fcda < -: ------- feat(pipe): add monitor cli
16: f168507 ! 5: 5b3f3a1 fix: sending 0 byte read
6: 2b86dec < -: ------- feat(pipe): status and rss commands
11: 68f5cd1 ! 6: 4fff471 refactor: fixes
7: 8df2aa2 < -: ------- feat(pipe): monitor pub and pipe cmd
14: e0c3a88 ! 7: d0dfc85 chore: SetDispatch on Broker
8: 11a5bce < -: ------- chore(pipe): monitor help text
9: 670350c < -: ------- refactor(pipe): monitor pipes on throttle interval
10: ade2b83 < -: ------- fix: window and ping fixes
12: 74fb40f < -: ------- chore(pipe): add logging stmts
15: 9a8fbde < -: ------- chore(pipe): implement monitor history db interface
17: 16df77f < -: ------- feat(pipe): monitor calculate uptime
18: 10ffdf1 < -: ------- feat(pipe): uptime cli command
19: 8231ac5 < -: ------- chore(pipe): monitor history stubs
20: 65a63a9 < -: ------- fix(pipe): uptime cmd
cmds
checkout latest patchset:
ssh pr.pico.sh print 103 | git am -3checkout any patchset in a patch request:
ssh pr.pico.sh print 103.[rev] | git am -3add changes to patch request:
git format-patch main --stdout | ssh pr.pico.sh pr add 103
Patchset
103.3
chore(pubsub): add more tests
Eric Bower
feat(pubsub): round robin
2026-01-25T16:38:25ZEric Bower
→ fix(pubsub): check for eof before processing and skip empty byte reads
2026-01-25T16:38:25ZEric Bower
fix: rr
2026-01-28T01:42:34ZEric Bower
fix: sending 0 byte read
2026-01-28T01:50:04ZEric Bower
refactor: fixes
2026-01-28T02:50:25ZEric Bower
chore: SetDispatch on Broker
2026-01-29T02:44:01ZEric Bower
2026-02-01T16:37:59Z
fix(pubsub): check for eof before processing and skip empty byte reads
Eric Bower
2026-01-28T01:42:34ZSemantic diff summary
0 added,
1 modified,
0 signature changed,
0 removed
across 1 analyzed file
+15
-8
pkg/pubsub/broker.go
#
| ... | ... | @@ -104,6 +104,21 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error) | |
| 104 | 104 | for { | |
| 105 | 105 | data := make([]byte, 32*1024) | |
| 106 | 106 | n, err := client.ReadWriter.Read(data) | |
| 107 | + | ||
| 108 | + | // Check for EOF before processing | |
| 109 | + | if err != nil { | |
| 110 | + | if errors.Is(err, io.EOF) { | |
| 111 | + | return | |
| 112 | + | } | |
| 113 | + | inputErr = err | |
| 114 | + | return | |
| 115 | + | } | |
| 116 | + | ||
| 117 | + | // Skip empty reads | |
| 118 | + | if n == 0 { | |
| 119 | + | continue | |
| 120 | + | } | |
| 121 | + | ||
| 107 | 122 | data = data[:n] | |
| 108 | 123 | ||
| 109 | 124 | channelMessage := ChannelMessage{ |