93 lines
2.4 KiB
Go
93 lines
2.4 KiB
Go
package webrtc
|
|
|
|
import (
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
|
|
"github.com/pion/webrtc/v4"
|
|
)
|
|
|
|
// WHEPHandler serves the WebRTC-HTTP Egress Protocol POST.
|
|
type WHEPHandler struct {
|
|
registry *Registry
|
|
factory *PeerFactory
|
|
config Config
|
|
|
|
mu sync.Mutex
|
|
peers map[string]*Peer // resourceID → Peer
|
|
peersCount int64 // atomic, for cap check without lock
|
|
}
|
|
|
|
// NewWHEPHandler constructs a handler with the given dependencies.
|
|
func NewWHEPHandler(r *Registry, f *PeerFactory, c Config) *WHEPHandler {
|
|
return &WHEPHandler{
|
|
registry: r,
|
|
factory: f,
|
|
config: c,
|
|
peers: make(map[string]*Peer),
|
|
}
|
|
}
|
|
|
|
// ServeHTTP handles POST /whep/{stream_id}. Other methods and paths return 405.
|
|
func (h *WHEPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
w.Header().Set("Allow", "POST")
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
|
|
// Extract stream_id from path: /whep/{stream_id}
|
|
streamID := strings.TrimPrefix(r.URL.Path, "/whep/")
|
|
if streamID == "" || strings.Contains(streamID, "/") {
|
|
http.Error(w, "invalid stream id", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
// Peer cap enforcement (happy path still respects the cap).
|
|
if atomic.LoadInt64(&h.peersCount) >= int64(h.config.MaxPeersTotal) {
|
|
http.Error(w, ErrPeerCapReached.Error(), http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
|
|
handle, ok := h.registry.Lookup(streamID)
|
|
if !ok {
|
|
http.Error(w, ErrStreamNotFound.Error(), http.StatusNotFound)
|
|
return
|
|
}
|
|
src, ok := handle.(*Source)
|
|
if !ok {
|
|
http.Error(w, "registered source is not a *Source", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
body, err := io.ReadAll(r.Body)
|
|
if err != nil {
|
|
http.Error(w, "read body: "+err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
if len(body) == 0 {
|
|
http.Error(w, ErrInvalidSDP.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)}
|
|
|
|
peer, err := h.factory.CreatePeer(r.Context(), src, offer)
|
|
if err != nil {
|
|
http.Error(w, "create peer: "+err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
h.mu.Lock()
|
|
h.peers[peer.ResourceID()] = peer
|
|
h.mu.Unlock()
|
|
atomic.AddInt64(&h.peersCount, 1)
|
|
|
|
w.Header().Set("Content-Type", "application/sdp")
|
|
w.Header().Set("Location", "/whep/"+streamID+"/"+peer.ResourceID())
|
|
w.WriteHeader(http.StatusCreated)
|
|
_, _ = io.WriteString(w, peer.Answer().SDP)
|
|
}
|