pico
created pr with
104.1
changed status to
accepted
cmds
checkout latest patchset:
ssh pr.pico.sh print 104 | git am -3checkout any patchset in a patch request:
ssh pr.pico.sh print 104.[rev] | git am -3add changes to patch request:
git format-patch main --stdout | ssh pr.pico.sh pr add 104set PR to open (enables RSS notifications):
ssh pr.pico.sh pr open 104set PR to draft (stops RSS notifications):
ssh pr.pico.sh pr draft 104
Patchset
104.1
feat(pipe): ?prefix= query param for pub and pipe
Eric Bower
2026-01-22T23:51:10ZSemantic diff summary
0 added,
3 modified,
0 signature changed,
0 removed
across 1 analyzed file
+53
-7
pkg/apps/pipe/api.go
#
| ... | ... | @@ -245,7 +248,12 @@ func handlePub(pubsub bool) http.HandlerFunc { | |
| 245 | 248 | case <-r.Context().Done(): | |
| 246 | 249 | break outer | |
| 247 | 250 | default: | |
| 248 | - | n, err := p.Write(first) | |
| 251 | + | messageToWrite := first | |
| 252 | + | if prefix != "" { | |
| 253 | + | messageToWrite = append([]byte(prefix), messageToWrite...) | |
| 254 | + | } | |
| 255 | + | ||
| 256 | + | n, err := p.Write(messageToWrite) | |
| 249 | 257 | if err != nil { | |
| 250 | 258 | logger.Error("pub write error", "topic", topic, "info", clientInfo, "err", err.Error()) | |
| 251 | 259 | http.Error(w, "server error", http.StatusInternalServerError) |
| ... | ... | @@ -314,6 +322,8 @@ func handlePipe() http.HandlerFunc { | |
| 314 | 322 | params += fmt.Sprintf(" -a=%s", cleanList) | |
| 315 | 323 | } | |
| 316 | 324 | ||
| 325 | + | prefix := r.URL.Query().Get("prefix") | |
| 326 | + | ||
| 317 | 327 | id := uuid.NewString() | |
| 318 | 328 | ||
| 319 | 329 | p, err := sshClient.AddSession(id, fmt.Sprintf("pipe %s %s", params, topic), 0, -1, -1) |
| ... | ... | @@ -373,12 +385,46 @@ func handlePipe() http.HandlerFunc { | |
| 373 | 385 | break | |
| 374 | 386 | } | |
| 375 | 387 | ||
| 376 | - | buf = buf[:n] | |
| 377 | - | ||
| 378 | - | err = c.WriteMessage(messageType, buf) | |
| 379 | - | if err != nil { | |
| 380 | - | logger.Error("pipe write error", "topic", topic, "info", clientInfo, "err", err.Error()) | |
| 381 | - | break | |
| 388 | + | messageBuffer = append(messageBuffer, buf[:n]...) | |
| 389 | + | ||
| 390 | + | if prefix != "" { | |
| 391 | + | // Buffer and split on prefix boundaries | |
| 392 | + | for { | |
| 393 | + | firstIdx := bytes.Index(messageBuffer, []byte(prefix)) | |
| 394 | + | if firstIdx == -1 { | |
| 395 | + | // No prefix found, clear buffer (shouldn't happen in normal use) | |
| 396 | + | messageBuffer = nil | |
| 397 | + | break | |
| 398 | + | } | |
| 399 | + | ||
| 400 | + | // Look for next prefix after the first one | |
| 401 | + | secondIdx := bytes.Index(messageBuffer[firstIdx+len(prefix):], []byte(prefix)) | |
| 402 | + | if secondIdx == -1 { | |
| 403 | + | // No complete message yet, keep buffer as is | |
| 404 | + | break | |
| 405 | + | } | |
| 406 | + | ||
| 407 | + | // We have a complete message, extract and send it | |
| 408 | + | messageToSend := messageBuffer[firstIdx : firstIdx+len(prefix)+secondIdx] | |
| 409 | + | err = c.WriteMessage(messageType, messageToSend) | |
| 410 | + | if err != nil { | |
| 411 | + | logger.Error("pipe write error", "topic", topic, "info", clientInfo, "err", err.Error()) | |
| 412 | + | break | |
| 413 | + | } | |
| 414 | + | ||
| 415 | + | // Update buffer to remove sent message | |
| 416 | + | messageBuffer = messageBuffer[firstIdx+len(prefix)+secondIdx:] | |
| 417 | + | } | |
| 418 | + | } else { | |
| 419 | + | // No prefix set, send all data as-is | |
| 420 | + | if len(messageBuffer) > 0 { | |
| 421 | + | err = c.WriteMessage(messageType, messageBuffer) | |
| 422 | + | if err != nil { | |
| 423 | + | logger.Error("pipe write error", "topic", topic, "info", clientInfo, "err", err.Error()) | |
| 424 | + | break | |
| 425 | + | } | |
| 426 | + | messageBuffer = nil | |
| 427 | + | } | |
| 382 | 428 | } | |
| 383 | 429 | } | |
| 384 | 430 | }() |