feat(19.7-01): streaming routes, MCP mount hook, configurable authorize scopes
- surf.Router.MountStream and recoverStream: a raw, unbuffered route kind
whose http.Flusher reaches the client; panic after the first byte only
aborts the body
- surf.MCPBuildTag and the mountPier hook: mcp.enabled is inert without the
summercms_mcp tag until it is true, then boot fails naming the tag
(pier_off.go is the untagged side)
- wristband parseAuthorizeScopes takes the server's allowed set built from
Options.ScopesSupported; the default set is unchanged
- cooler.KeyPrefix exported for the MCP module's Redis keys
- smoke tests TestStreamingRouteSmoke, TestPierMountSmoke, TestMCPScopesSmoke
- surf, wristband and cooler READMEs, docs/services/{routing,oauth-server}.md
This commit is contained in:
108
modules/surf/streaming.go
Normal file
108
modules/surf/streaming.go
Normal file
@@ -0,0 +1,108 @@
|
||||
package surf
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/wire"
|
||||
)
|
||||
|
||||
// MCPBuildTag is the Go build tag that links the MCP module (pier) and its
|
||||
// SDK into a binary. List it under build.tags in summer.yaml so summer build
|
||||
// passes it to go build; a raw go build needs -tags summercms_mcp. Without
|
||||
// the tag mcp.enabled: true fails the start-up with an error naming the tag.
|
||||
const MCPBuildTag = "summercms_mcp"
|
||||
|
||||
// MountStream registers one streaming route under the currently bound plugin
|
||||
// (BindPlugin): a raw route (no house middleware, no default body limit)
|
||||
// whose response is not buffered, so the handler's http.Flusher reaches the
|
||||
// client while it is still running, for server-sent events and other
|
||||
// long-lived responses. Everything else is as for a raw route: the named
|
||||
// middleware (guards, throttles), a body.limit, the locale, the CORS profile
|
||||
// and the access-log pattern. A panic before the first byte answers an opaque
|
||||
// 500; after it the body is only cut short. A route another plugin already
|
||||
// registered for method and path fails with an error naming that plugin.
|
||||
func (r *Router) MountStream(method, path string, h http.Handler, middleware ...string) error {
|
||||
if r == nil {
|
||||
return fmt.Errorf("surf: router is nil")
|
||||
}
|
||||
method = strings.ToUpper(strings.TrimSpace(method))
|
||||
full := joinPath(r.prefix, path)
|
||||
if method == "" || strings.TrimSpace(path) == "" || h == nil {
|
||||
return fmt.Errorf("surf: plugin %q registered an empty stream route", r.pluginID)
|
||||
}
|
||||
if strings.ContainsAny(full, " \t\r\n") {
|
||||
return fmt.Errorf("surf: plugin %q registered stream route %s %q with whitespace in its path", r.pluginID, method, full)
|
||||
}
|
||||
key := method + " " + full
|
||||
if prev, ok := r.seen[key]; ok {
|
||||
return fmt.Errorf("surf: route %s registered by %s collides with the stream route plugin %q mounts", key, prev, r.pluginID)
|
||||
}
|
||||
r.seen[key] = r.pluginID
|
||||
mw := append([]string{}, r.middleware...)
|
||||
mw = append(mw, middleware...)
|
||||
r.routes = append(r.routes, route{
|
||||
pluginID: r.pluginID,
|
||||
method: method,
|
||||
path: full,
|
||||
handler: h,
|
||||
middleware: mw,
|
||||
raw: true,
|
||||
stream: true,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// recoverStream is the recovery layer of a stream route. It serves next on
|
||||
// the real ResponseWriter (no bufferedResponse, so Flush reaches the client)
|
||||
// and recovers a panic by writing wire's opaque 500, or the HTML error page
|
||||
// for a browser navigation, only while nothing has been sent yet; after the
|
||||
// first header, byte or flush a panic only aborts the body and marks the
|
||||
// request panicked for the access log. The panic value is never written.
|
||||
func recoverStream(pages *errorPages, next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
sw := &headerSentWriter{ResponseWriter: w}
|
||||
defer func() {
|
||||
if rec := recover(); rec != nil {
|
||||
markPanicked(r)
|
||||
if sw.sent {
|
||||
return
|
||||
}
|
||||
if pages.wants(r) {
|
||||
pages.write(w, r, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
wire.WriteOpaque500(w)
|
||||
}
|
||||
}()
|
||||
next.ServeHTTP(sw, r)
|
||||
})
|
||||
}
|
||||
|
||||
// headerSentWriter records whether the response has started (a WriteHeader,
|
||||
// Write or Flush) and passes Flush and Unwrap through to the real writer.
|
||||
type headerSentWriter struct {
|
||||
http.ResponseWriter
|
||||
sent bool
|
||||
}
|
||||
|
||||
func (w *headerSentWriter) WriteHeader(status int) {
|
||||
w.sent = true
|
||||
w.ResponseWriter.WriteHeader(status)
|
||||
}
|
||||
|
||||
func (w *headerSentWriter) Write(p []byte) (int, error) {
|
||||
w.sent = true
|
||||
return w.ResponseWriter.Write(p)
|
||||
}
|
||||
|
||||
// Flush sends the headers (200 when none were written) and the bytes so far.
|
||||
func (w *headerSentWriter) Flush() {
|
||||
w.sent = true
|
||||
if f, ok := w.ResponseWriter.(http.Flusher); ok {
|
||||
f.Flush()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *headerSentWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter }
|
||||
Reference in New Issue
Block a user