diff --git a/trickle/README.md b/trickle/README.md index 8455001494..ce4a8706b5 100644 --- a/trickle/README.md +++ b/trickle/README.md @@ -22,6 +22,8 @@ Publishers POST to /channel-name/seq Subscribers GET to /channel-name/seq +Clients can query the next segment seq with GET to /channel-name/next + The `channel-name` is any valid HTTP path part. The `seq` is an integer sequence number indicating the segment, and must increase sequentially without gaps. @@ -52,6 +54,10 @@ Subscribers can initiate a subscribe with a `seq` of -1 to retrieve the most rec Subscribers can retrieve the current `seq` with the `Lp-Trickle-Seq` metadata (HTTP header). This is useful in case `-1` was used to initiate the subscription; the subscribing client can then pre-connect to `Lp-Trickle-Seq + 1` +GET `/channel-name/next` returns the next segment seq as plain text in the response body and in the `Lp-Trickle-Latest` response header. +If the channel does not exist, the server returns 404. +If the channel is closed, the server also includes `Lp-Trickle-Closed: terminated`. + Subscribers can initiate a subscribe with a `seq` of -N to get the Nth-from-last segment. (TODO) The server should send subscribers `Lp-Trickle-Size` metadata to indicate the size of the content up until now. This allows clients to know where the live edge is, eg video implementations can decode-and-discard frames up until the edge to achieve immediate playback without waiting for the next segment. (TODO) diff --git a/trickle/trickle_server.go b/trickle/trickle_server.go index 4fefa35159..895677f9cf 100644 --- a/trickle/trickle_server.go +++ b/trickle/trickle_server.go @@ -117,6 +117,7 @@ func ConfigureServer(config TrickleServerConfig) *Server { ) mux.HandleFunc("POST "+basePath+"{streamName}", streamManager.handleCreate) + mux.HandleFunc("GET "+basePath+"{streamName}/next", streamManager.handleNext) mux.HandleFunc("GET "+basePath+"{streamName}/{idx}", streamManager.handleGet) mux.HandleFunc("POST "+basePath+"{streamName}/{idx}", streamManager.handlePost) mux.HandleFunc("DELETE "+basePath+"{streamName}/{idx}", streamManager.closeSeq) @@ -507,6 +508,25 @@ func (sm *Server) handleGet(w http.ResponseWriter, r *http.Request) { stream.handleGet(w, r, idx) } +func (sm *Server) handleNext(w http.ResponseWriter, r *http.Request) { + stream, exists := sm.getStream(r.PathValue("streamName")) + if !exists { + http.Error(w, "Stream not found", http.StatusNotFound) + return + } + stream.mutex.RLock() + nextWrite := stream.nextWrite + closed := stream.closed + stream.mutex.RUnlock() + + next := strconv.Itoa(nextWrite) + w.Header().Set("Lp-Trickle-Latest", next) + if closed { + w.Header().Set("Lp-Trickle-Closed", "terminated") + } + w.Write([]byte(next)) +} + func (s *Stream) handleGet(w http.ResponseWriter, r *http.Request, idx int) { segment, latestSeq, exists, closed := s.getForRead(idx) if !exists {