neoris/core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add WHIP server implementation with Docker support

Browse Source 
- Introduced a new Dockerfile for building the WHIP server with Go and FFmpeg.
- Implemented the WHIPHandler for managing WHIP publish sessions and generating URLs for clients.
- Added pionProvider for handling WebRTC sessions using the pion/webrtc library.
- Created the main WHIP server logic to handle SDP offers and manage active publishing streams.
- Implemented HTTP handlers for publishing, deleting, and retrieving stream information.
- Added support for SDP generation for FFmpeg consumption.
This commit is contained in:
Cesar Mendivil 2026-03-14 12:26:24 -07:00
parent 8b66753a27
commit 7a8073eedd
16 changed files with 1843 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -6,7 +6,7 @@
/data/**
/test/**
.vscode
/vendor
*.ts
*.ts.tmp
*.m3u8

40
Dockerfile.whip-test Normal file
View File

@ -0,0 +1,40 @@
ARG GOLANG_IMAGE=golang:1.22-alpine3.19
ARG FFMPEG_IMAGE=datarhei/base:alpine-ffmpeg-latest
FROM --platform=$BUILDPLATFORM $GOLANG_IMAGE AS builder
RUN apk add git make
# Cache go module downloads separately from source code.
# This layer only rebuilds when go.mod/go.sum/vendor change.
WORKDIR /dist/core
COPY go.mod go.sum ./
COPY vendor/ ./vendor/
# Now copy source and build. This layer rebuilds on any .go file change.
COPY . .
RUN make release && make import && make ffmigrate
FROM $FFMPEG_IMAGE
COPY --from=builder /dist/core/core /core/bin/core
COPY --from=builder /dist/core/import /core/bin/import
COPY --from=builder /dist/core/ffmigrate /core/bin/ffmigrate
COPY --from=builder /dist/core/mime.types /core/mime.types
COPY --from=builder /dist/core/run.sh /core/bin/run.sh
RUN chmod +x /core/bin/run.sh && mkdir -p /core/config /core/data
ENV CORE_CONFIGFILE=/core/config/config.json
ENV CORE_STORAGE_DISK_DIR=/core/data
ENV CORE_DB_DIR=/core/config
ENV CORE_WHIP_ENABLE=true
ENV CORE_WHIP_ADDRESS=:8555
ENV CORE_API_AUTH_ENABLE=false
EXPOSE 8080/tcp
EXPOSE 8555/tcp
VOLUME ["/core/data", "/core/config"]
ENTRYPOINT ["/core/bin/run.sh"]
WORKDIR /core

81
app/api/api.go
View File

@ -40,6 +40,7 @@ import (
"github.com/datarhei/core/v16/session"
"github.com/datarhei/core/v16/srt"
"github.com/datarhei/core/v16/update"
"github.com/datarhei/core/v16/whip"
"github.com/caddyserver/certmagic"
"go.uber.org/zap"
@ -73,6 +74,7 @@ type api struct {
s3fs map[string]fs.Filesystem
rtmpserver rtmp.Server
srtserver srt.Server
whipserver whip.Server
metrics monitor.HistoryMonitor
prom prometheus.Metrics
service service.Service
@ -98,6 +100,7 @@ type api struct {
rtmp log.Logger
rtmps log.Logger
srt log.Logger
whip log.Logger
}
}
@ -348,6 +351,11 @@ func (a *api) start() error {
return fmt.Errorf("unable to register session collector: %w", err)
}
whipCollector, err := sessions.Register("whip", config)
if err != nil {
return fmt.Errorf("unable to register session collector: %w", err)
}
if _, err := sessions.Register("http", config); err != nil {
return fmt.Errorf("unable to register session collector: %w", err)
}
@ -373,9 +381,13 @@ func (a *api) start() error {
srt.AddCompanion(ffmpeg)
srt.AddCompanion(rtmp)
whipCollector.AddCompanion(hls)
whipCollector.AddCompanion(ffmpeg)
ffmpeg.AddCompanion(hls)
ffmpeg.AddCompanion(rtmp)
ffmpeg.AddCompanion(srt)
ffmpeg.AddCompanion(whipCollector)
} else {
sessions, _ := session.New(session.Config{})
a.sessions = sessions
@ -570,6 +582,23 @@ func (a *api) start() error {
}, map[string]string{
"latency": "20000", // 20 milliseconds, FFmpeg requires microseconds
})
// WHIP: {whip} in an input address expands to the FFmpeg-readable SDP relay
// endpoint served by the local WHIP server (http://localhost:{port}/whip/{name}/sdp).
// Encoders and browsers push to the companion URL:
// http://{host}:{port}/whip/{name} (plain RTP / no ICE)
// or, with a WebRTCProvider injected, full WebRTC is negotiated.
a.replacer.RegisterTemplateFunc("whip", func(config *restreamapp.Config, section string) string {
_, whipPort, _ := gonet.SplitHostPort(cfg.WHIP.Address)
if whipPort == "" {
whipPort = "8555"
}
url := "http://localhost:" + whipPort + "/whip/{name}/sdp"
if len(cfg.WHIP.Token) != 0 {
url += "?token=" + cfg.WHIP.Token
}
return url
}, nil)
}
filesystems := []fs.Filesystem{
@ -924,6 +953,23 @@ func (a *api) start() error {
a.srtserver = srtserver
}
if cfg.WHIP.Enable {
a.log.logger.whip = a.log.logger.core.WithComponent("WHIP").WithField("address", cfg.WHIP.Address)
whipserver, err := whip.New(whip.Config{
Addr: cfg.WHIP.Address,
Token: cfg.WHIP.Token,
Logger: a.log.logger.whip,
Collector: a.sessions.Collector("whip"),
WebRTCProvider: whip.NewPionProvider(),
})
if err != nil {
return fmt.Errorf("unable to create WHIP server: %w", err)
}
a.whipserver = whipserver
}
logcontext := "HTTP"
if cfg.TLS.Enable {
logcontext = "HTTPS"
@ -1014,6 +1060,7 @@ func (a *api) start() error {
},
RTMP: a.rtmpserver,
SRT: a.srtserver,
WHIP: a.whipserver,
JWT: a.httpjwt,
Config: a.config.store,
Sessions: a.sessions,
@ -1194,6 +1241,32 @@ func (a *api) start() error {
}()
}
if a.whipserver != nil {
wgStart.Add(1)
a.wgStop.Add(1)
go func() {
logger := a.log.logger.whip
defer func() {
logger.Info().Log("Server exited")
a.wgStop.Done()
}()
wgStart.Done()
logger.Info().Log("Server started")
err := a.whipserver.ListenAndServe()
if err != nil && err != whip.ErrServerClosed {
err = fmt.Errorf("WHIP server: %w", err)
} else {
err = nil
}
sendError(err)
}()
}
wgStart.Add(1)
a.wgStop.Add(1)
@ -1354,6 +1427,14 @@ func (a *api) stop() {
a.srtserver = nil
}
// Stop the WHIP server
if a.whipserver != nil {
a.log.logger.whip.Info().Log("Stopping ...")
a.whipserver.Close()
a.whipserver = nil
}
// Stop the RTMP server
if a.rtmpserver != nil {
a.log.logger.rtmp.Info().Log("Stopping ...")

6
config/config.go
View File

@ -98,6 +98,7 @@ func (d *Config) Clone() *Config {
data.Storage = d.Storage
data.RTMP = d.RTMP
data.SRT = d.SRT
data.WHIP = d.WHIP
data.FFmpeg = d.FFmpeg
data.Playout = d.Playout
data.Debug = d.Debug
@ -227,6 +228,11 @@ func (d *Config) init() {
d.vars.Register(value.NewBool(&d.SRT.Log.Enable, false), "srt.log.enable", "CORE_SRT_LOG_ENABLE", nil, "Enable SRT server logging", false, false)
d.vars.Register(value.NewStringList(&d.SRT.Log.Topics, []string{}, ","), "srt.log.topics", "CORE_SRT_LOG_TOPICS", nil, "List of topics to log", false, false)
// WHIP
d.vars.Register(value.NewBool(&d.WHIP.Enable, false), "whip.enable", "CORE_WHIP_ENABLE", nil, "Enable WHIP (WebRTC HTTP Ingestion Protocol) server", false, false)
d.vars.Register(value.NewAddress(&d.WHIP.Address, ":8555"), "whip.address", "CORE_WHIP_ADDRESS", nil, "WHIP server HTTP listen address", false, false)
d.vars.Register(value.NewString(&d.WHIP.Token, ""), "whip.token", "CORE_WHIP_TOKEN", nil, "WHIP bearer token for publishing clients", false, true)
// FFmpeg
d.vars.Register(value.NewExec(&d.FFmpeg.Binary, "ffmpeg", d.fs), "ffmpeg.binary", "CORE_FFMPEG_BINARY", nil, "Path to ffmpeg binary", true, false)
d.vars.Register(value.NewInt64(&d.FFmpeg.MaxProcesses, 0), "ffmpeg.max_processes", "CORE_FFMPEG_MAXPROCESSES", nil, "Max. allowed simultaneously running ffmpeg instances, 0 for unlimited", false, false)

14
config/data.go
View File

@ -113,6 +113,20 @@ type Data struct {
Topics []string `json:"topics"`
} `json:"log"`
} `json:"srt"`
// WHIP is the configuration for the WHIP (WebRTC HTTP Ingestion Protocol) server.
// The server accepts WebRTC/RTP streams from encoders and browsers, making them
// available to FFmpeg via an SDP relay endpoint.
//
// FFmpeg input address template: http://localhost:{Address}/whip/{name}/sdp
// (Use the {whip} replacer in process configs.)
WHIP struct {
// Enable activates the WHIP HTTP ingestion server.
Enable bool `json:"enable"`
// Address is the HTTP listen address for the WHIP endpoint, e.g. ":8555".
Address string `json:"address"`
// Token is an optional bearer token required from WHIP publishing clients.
Token string `json:"token"`
} `json:"whip"`
FFmpeg struct {
Binary string `json:"binary"`
MaxProcesses int64 `json:"max_processes" format:"int64"`

152
docs/swagger.json
View File

@ -1896,6 +1896,102 @@
}
}
},
"/api/v3/whip": {
"get": {
"security": [
{
"ApiKeyAuth": []
}
],
"description": "List all currently active WHIP (WebRTC HTTP Ingestion Protocol) streams.",
"produces": [
"application/json"
],
"tags": [
"v16.?.?"
],
"summary": "List all publishing WHIP streams",
"operationId": "whip-3-list-channels",
"responses": {
"200": {
"description": "OK",
"schema": {
"type": "array",
"items": {
"$ref": "#/definitions/api.WHIPChannel"
}
}
}
}
}
},
"/api/v3/whip/url": {
"get": {
"security": [
{
"ApiKeyAuth": []
}
],
"description": "Returns the base publish URL and configuration so the UI can display it in a WHIP Server panel.",
"produces": [
"application/json"
],
"tags": [
"v16.?.?"
],
"summary": "Get WHIP server base URL",
"operationId": "whip-3-get-server-url",
"responses": {
"200": {
"description": "OK",
"schema": {
"$ref": "#/definitions/api.WHIPServerInfo"
}
}
}
}
},
"/api/v3/whip/{name}/url": {
"get": {
"security": [
{
"ApiKeyAuth": []
}
],
"description": "Returns the URL to configure in OBS and the internal SDP relay URL for FFmpeg.",
"produces": [
"application/json"
],
"tags": [
"v16.?.?"
],
"summary": "Get WHIP publish URL for a stream key",
"operationId": "whip-3-get-url",
"parameters": [
{
"type": "string",
"description": "Stream key",
"name": "name",
"in": "path",
"required": true
}
],
"responses": {
"200": {
"description": "OK",
"schema": {
"$ref": "#/definitions/api.WHIPURLs"
}
},
"400": {
"description": "Bad Request",
"schema": {
"$ref": "#/definitions/api.Error"
}
}
}
}
},
"/api/v3/widget/process/{id}": {
"get": {
"description": "Fetch minimal statistics about a process, which is not protected by any auth.",
@ -3439,6 +3535,62 @@
}
}
},
"api.WHIPChannel": {
"type": "object",
"properties": {
"name": {
"description": "Stream key used in the WHIP URL path.",
"type": "string"
},
"published_at": {
"description": "RFC 3339 timestamp when the publisher connected.",
"type": "string",
"format": "date-time"
}
}
},
"api.WHIPURLs": {
"type": "object",
"properties": {
"publish_url": {
"description": "URL to enter in OBS Settings Stream Server.",
"type": "string"
},
"sdp_url": {
"description": "Internal FFmpeg-readable relay SDP URL.",
"type": "string"
},
"stream_key": {
"description": "Stream key component used in both URLs.",
"type": "string"
}
}
},
"api.WHIPServerInfo": {
"type": "object",
"properties": {
"base_publish_url": {
"description": "Base URL for OBS. Append the stream key.",
"type": "string"
},
"base_sdp_url": {
"description": "Base internal SDP relay URL.",
"type": "string"
},
"example_obs_url": {
"description": "Example URL with placeholder stream key.",
"type": "string"
},
"has_token": {
"description": "Whether a bearer token is required.",
"type": "boolean"
},
"input_address_template": {
"description": "Process input address template for WHIP.",
"type": "string"
}
}
},
"api.SRTChannels": {
"type": "object",
"properties": {

103
docs/swagger.yaml
View File

@ -976,6 +976,46 @@ definitions:
name:
type: string
type: object
api.WHIPChannel:
properties:
name:
description: Stream key used in the WHIP URL path.
type: string
published_at:
description: RFC 3339 timestamp when the publisher connected.
format: date-time
type: string
type: object
api.WHIPURLs:
properties:
publish_url:
description: URL to enter in OBS Settings Stream Server.
type: string
sdp_url:
description: Internal FFmpeg-readable relay SDP URL.
type: string
stream_key:
description: Stream key component used in both URLs.
type: string
type: object
api.WHIPServerInfo:
properties:
base_publish_url:
description: Base URL for OBS. Append the stream key.
type: string
base_sdp_url:
description: Base internal SDP relay URL.
type: string
example_obs_url:
description: Example URL with placeholder stream key.
type: string
has_token:
description: Whether a bearer token is required.
type: boolean
input_address_template:
description: Process input address template for WHIP.
type: string
type: object
api.SRTChannels:
properties:
connections:
@ -3186,6 +3226,69 @@ paths:
summary: List all publishing SRT treams
tags:
- v16.9.0
/api/v3/whip:
get:
description: List all currently active WHIP (WebRTC HTTP Ingestion Protocol)
streams.
operationId: whip-3-list-channels
produces:
- application/json
responses:
"200":
description: OK
schema:
items:
$ref: '#/definitions/api.WHIPChannel'
type: array
security:
- ApiKeyAuth: []
summary: List all publishing WHIP streams
tags:
- v16.?.?
/api/v3/whip/url:
get:
description: Returns the base publish URL and configuration so the UI can display
it in a WHIP Server panel.
operationId: whip-3-get-server-url
produces:
- application/json
responses:
"200":
description: OK
schema:
$ref: '#/definitions/api.WHIPServerInfo'
security:
- ApiKeyAuth: []
summary: Get WHIP server base URL
tags:
- v16.?.?
/api/v3/whip/{name}/url:
get:
description: Returns the URL to configure in OBS and the internal SDP relay
URL for FFmpeg.
operationId: whip-3-get-url
parameters:
- description: Stream key
in: path
name: name
required: true
type: string
produces:
- application/json
responses:
"200":
description: OK
schema:
$ref: '#/definitions/api.WHIPURLs'
"400":
description: Bad Request
schema:
$ref: '#/definitions/api.Error'
security:
- ApiKeyAuth: []
summary: Get WHIP publish URL for a stream key
tags:
- v16.?.?
/api/v3/widget/process/{id}:
get:
description: Fetch minimal statistics about a process, which is not protected

17
go.mod
View File

@ -22,6 +22,8 @@ require (
github.com/lithammer/shortuuid/v4 v4.0.0
github.com/mattn/go-isatty v0.0.20
github.com/minio/minio-go/v7 v7.0.70
github.com/pion/interceptor v0.1.29
github.com/pion/webrtc/v3 v3.3.6
github.com/prep/average v0.0.0-20200506183628-d26c465f48c3
github.com/prometheus/client_golang v1.19.1
github.com/puzpuzpuz/xsync/v3 v3.1.0
@ -73,6 +75,20 @@ require (
github.com/miekg/dns v1.1.59 // indirect
github.com/minio/md5-simd v1.1.2 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
github.com/pion/datachannel v1.5.8 // indirect
github.com/pion/dtls/v2 v2.2.12 // indirect
github.com/pion/ice/v2 v2.3.38 // indirect
github.com/pion/logging v0.2.2 // indirect
github.com/pion/mdns v0.0.12 // indirect
github.com/pion/randutil v0.1.0 // indirect
github.com/pion/rtcp v1.2.14 // indirect
github.com/pion/rtp v1.8.7 // indirect
github.com/pion/sctp v1.8.19 // indirect
github.com/pion/sdp/v3 v3.0.9 // indirect
github.com/pion/srtp/v2 v2.0.20 // indirect
github.com/pion/stun v0.6.1 // indirect
github.com/pion/transport/v2 v2.2.10 // indirect
github.com/pion/turn/v2 v2.1.6 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
github.com/prometheus/client_model v0.6.1 // indirect
@ -88,6 +104,7 @@ require (
github.com/urfave/cli/v2 v2.27.2 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/valyala/fasttemplate v1.2.2 // indirect
github.com/wlynxg/anet v0.0.3 // indirect
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
github.com/xrash/smetrics v0.0.0-20240312152122-5f08fbb34913 // indirect

96
go.sum
View File

@ -77,6 +77,7 @@ github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
@ -98,8 +99,11 @@ github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa02
github.com/klauspost/cpuid/v2 v2.0.12/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM=
github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/labstack/echo-jwt v0.0.0-20221127215225-c84d41a71003 h1:FyalHKl9hnJvhNbrABJXXjC2hG7gvIF0ioW9i0xHNQU=
@ -134,6 +138,48 @@ github.com/minio/minio-go/v7 v7.0.70 h1:1u9NtMgfK1U42kUxcsl5v0yj6TEOPR497OAQxpJn
github.com/minio/minio-go/v7 v7.0.70/go.mod h1:4yBA8v80xGA30cfM3fz0DKYMXunWl/AV/6tWEs9ryzo=
github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
github.com/pion/datachannel v1.5.8 h1:ph1P1NsGkazkjrvyMfhRBUAWMxugJjq2HfQifaOoSNo=
github.com/pion/datachannel v1.5.8/go.mod h1:PgmdpoaNBLX9HNzNClmdki4DYW5JtI7Yibu8QzbL3tI=
github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk=
github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
github.com/pion/ice/v2 v2.3.38 h1:DEpt13igPfvkE2+1Q+6e8mP30dtWnQD3CtMIKoRDRmA=
github.com/pion/ice/v2 v2.3.38/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ=
github.com/pion/interceptor v0.1.29 h1:39fsnlP1U8gw2JzOFWdfCU82vHvhW9o0rZnZF56wF+M=
github.com/pion/interceptor v0.1.29/go.mod h1:ri+LGNjRUc5xUNtDEPzfdkmSqISixVTBF/z/Zms/6T4=
github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
github.com/pion/mdns v0.0.12 h1:CiMYlY+O0azojWDmxdNr7ADGrnZ+V6Ilfner+6mSVK8=
github.com/pion/mdns v0.0.12/go.mod h1:VExJjv8to/6Wqm1FXK+Ii/Z9tsVk/F5sD/N70cnYFbk=
github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
github.com/pion/rtcp v1.2.12/go.mod h1:sn6qjxvnwyAkkPzPULIbVqSKI5Dv54Rv7VG0kNxh9L4=
github.com/pion/rtcp v1.2.14 h1:KCkGV3vJ+4DAJmvP0vaQShsb0xkRfWkO540Gy102KyE=
github.com/pion/rtcp v1.2.14/go.mod h1:sn6qjxvnwyAkkPzPULIbVqSKI5Dv54Rv7VG0kNxh9L4=
github.com/pion/rtp v1.8.3/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU=
github.com/pion/rtp v1.8.7 h1:qslKkG8qxvQ7hqaxkmL7Pl0XcUm+/Er7nMnu6Vq+ZxM=
github.com/pion/rtp v1.8.7/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU=
github.com/pion/sctp v1.8.19 h1:2CYuw+SQ5vkQ9t0HdOPccsCz1GQMDuVy5PglLgKVBW8=
github.com/pion/sctp v1.8.19/go.mod h1:P6PbDVA++OJMrVNg2AL3XtYHV4uD6dvfyOovCgMs0PE=
github.com/pion/sdp/v3 v3.0.9 h1:pX++dCHoHUwq43kuwf3PyJfHlwIj4hXA7Vrifiq0IJY=
github.com/pion/sdp/v3 v3.0.9/go.mod h1:B5xmvENq5IXJimIO4zfp6LAe1fD9N+kFv+V/1lOdz8M=
github.com/pion/srtp/v2 v2.0.20 h1:HNNny4s+OUmG280ETrCdgFndp4ufx3/uy85EawYEhTk=
github.com/pion/srtp/v2 v2.0.20/go.mod h1:0KJQjA99A6/a0DOVTu1PhDSw0CXF2jTkqOoMg3ODqdA=
github.com/pion/stun v0.6.1 h1:8lp6YejULeHBF8NmV8e2787BogQhduZugh5PdhDyyN4=
github.com/pion/stun v0.6.1/go.mod h1:/hO7APkX4hZKu/D0f2lHzNyvdkTGtIy3NDmLR7kSz/8=
github.com/pion/transport/v2 v2.2.1/go.mod h1:cXXWavvCnFF6McHTft3DWS9iic2Mftcz1Aq29pGcU5g=
github.com/pion/transport/v2 v2.2.3/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0=
github.com/pion/transport/v2 v2.2.4/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0=
github.com/pion/transport/v2 v2.2.10 h1:ucLBLE8nuxiHfvkFKnkDQRYWYfp8ejf4YBOPfaQpw6Q=
github.com/pion/transport/v2 v2.2.10/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E=
github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9SzK5f5xE0=
github.com/pion/transport/v3 v3.0.2 h1:r+40RJR25S9w3jbA6/5uEPTzcdn7ncyU44RWCbHkLg4=
github.com/pion/transport/v3 v3.0.2/go.mod h1:nIToODoOlb5If2jF9y2Igfx3PFYWfuXi37m0IlWa/D0=
github.com/pion/turn/v2 v2.1.3/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
github.com/pion/turn/v2 v2.1.6 h1:Xr2niVsiPTB0FPtt+yAWKFUkU1eotQbGgpTIld4x1Gc=
github.com/pion/turn/v2 v2.1.6/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
github.com/pion/webrtc/v3 v3.3.6 h1:7XAh4RPtlY1Vul6/GmZrv7z+NnxKA6If0KStXBI2ZLE=
github.com/pion/webrtc/v3 v3.3.6/go.mod h1:zyN7th4mZpV27eXybfR/cnUf3J2DRy8zw/mdjD9JTNM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@ -176,6 +222,7 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
github.com/stretchr/testify v1.3.1-0.20190311161405-34c6fa2dc709/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
@ -199,6 +246,8 @@ github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQ
github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
github.com/vektah/gqlparser/v2 v2.5.12 h1:COMhVVnql6RoaF7+aTBWiTADdpLGyZWU3K/NwW0ph98=
github.com/vektah/gqlparser/v2 v2.5.12/go.mod h1:WQQjFc+I1YIzoPvZBhUQX7waZgg3pMLi0r8KymvAE2w=
github.com/wlynxg/anet v0.0.3 h1:PvR53psxFXstc12jelG6f1Lv4MWqE0tI76/hHGjh9rg=
github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo=
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
@ -208,6 +257,7 @@ github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17
github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
github.com/xrash/smetrics v0.0.0-20240312152122-5f08fbb34913 h1:+qGGcbkzsfDQNPPe9UDgpxAWQrhbbBXOYJFQDq/dtJw=
github.com/xrash/smetrics v0.0.0-20240312152122-5f08fbb34913/go.mod h1:4aEEwZQutDLsQv2Deui4iYQ6DWTxR14g6m8Wv88+Xqk=
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
github.com/zeebo/assert v1.1.0 h1:hU1L1vLTHsnO8x8c9KAR5GmM5QscxHg5RNU5z5qbUWY=
@ -222,35 +272,81 @@ go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE=
golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw=
golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA=
golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac=
golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY=
golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
golang.org/x/term v0.11.0/go.mod h1:zC9APTIj3jG3FdV/Ons+XE1riIZXG4aZ4GTHiPZJPIU=
golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
golang.org/x/tools v0.21.0 h1:qc0xYgIbsSDt9EyWz05J5wfa7LOVW0YTLOXrqdLAWIw=
golang.org/x/tools v0.21.0/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/protobuf v1.34.1 h1:9ddQBjfCyZPOHPUiPxpYESBLc+T8P3E+Vo4IbKZgFWg=
google.golang.org/protobuf v1.34.1/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=

1
http/api/config.go
View File

@ -107,6 +107,7 @@ func (rscfg *SetConfig) MergeTo(cfg *config.Config) {
cfg.Storage = rscfg.Storage
cfg.RTMP = rscfg.RTMP
cfg.SRT = rscfg.SRT
cfg.WHIP = rscfg.WHIP
cfg.FFmpeg = rscfg.FFmpeg
cfg.Playout = rscfg.Playout
cfg.Debug = rscfg.Debug

166
http/handler/api/whip.go Normal file
View File

@ -0,0 +1,166 @@
package api
import (
"fmt"
"net/http"
"strings"
"time"
cfgstore "github.com/datarhei/core/v16/config/store"
"github.com/datarhei/core/v16/whip"
"github.com/labstack/echo/v4"
)
// WHIPHandler provides HTTP API access to the WHIP server channel information.
type WHIPHandler struct {
whip whip.Server
config cfgstore.Store
}
// NewWHIP returns a new WHIPHandler backed by the provided WHIP server.
func NewWHIP(whip whip.Server, config cfgstore.Store) *WHIPHandler {
return &WHIPHandler{
whip: whip,
config: config,
}
}
// WHIPChannel represents a currently active WHIP publish session.
type WHIPChannel struct {
// Name is the stream identifier used in the WHIP URL path.
Name string `json:"name" jsonschema:"minLength=1"`
// PublishedAt is the RFC 3339 timestamp when the publisher connected.
PublishedAt time.Time `json:"published_at"`
}
// WHIPURLs holds the publish URL (for OBS) and the internal SDP relay URL (for FFmpeg).
type WHIPURLs struct {
// PublishURL is the URL to enter in OBS → Settings → Stream → Server.
// Format: http://<public_host>:<port>/whip/<stream_key>
PublishURL string `json:"publish_url"`
// SDPURL is the internal FFmpeg-readable relay URL.
// Format: http://localhost:<port>/whip/<stream_key>/sdp
SDPURL string `json:"sdp_url"`
// StreamKey is the key component used in both URLs.
StreamKey string `json:"stream_key"`
}
// whipPublicBase returns "http://<host>:<port>" derived from the active config.
func (h *WHIPHandler) whipPublicBase() (string, string, error) {
if h.config == nil {
return "", "", fmt.Errorf("config not available")
}
cfg := h.config.GetActive()
addr := cfg.WHIP.Address // e.g. ":8555" or "0.0.0.0:8555"
// Extract port from address.
port := addr
if idx := strings.LastIndex(addr, ":"); idx >= 0 {
port = addr[idx+1:]
}
// Pick public host.
host := "localhost"
if len(cfg.Host.Name) > 0 && cfg.Host.Name[0] != "" {
host = cfg.Host.Name[0]
}
return fmt.Sprintf("http://%s:%s", host, port),
fmt.Sprintf("http://localhost:%s", port),
nil
}
// GetURL returns the WHIP publish URL for a given stream key.
// @Summary Get WHIP publish URL for a stream key
// @Description Returns the URL to configure in OBS and the internal SDP relay URL for FFmpeg.
// @Tags v16.?.?
// @ID whip-3-get-url
// @Produce json
// @Param name path string true "Stream key"
// @Success 200 {object} WHIPURLs
// @Security ApiKeyAuth
// @Router /api/v3/whip/{name}/url [get]
func (h *WHIPHandler) GetURL(c echo.Context) error {
name := c.Param("name")
if name == "" {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "stream key is required"})
}
publicBase, localBase, err := h.whipPublicBase()
if err != nil {
return c.JSON(http.StatusServiceUnavailable, map[string]string{"error": err.Error()})
}
return c.JSON(http.StatusOK, WHIPURLs{
PublishURL: fmt.Sprintf("%s/whip/%s", publicBase, name),
SDPURL: fmt.Sprintf("%s/whip/%s/sdp", localBase, name),
StreamKey: name,
})
}
// GetServerURL returns the base WHIP server URL (without a stream key).
// @Summary Get WHIP server base URL
// @Description Returns the base publish URL and configuration so the UI can display it in a WHIP Server panel.
// @Tags v16.?.?
// @ID whip-3-get-server-url
// @Produce json
// @Success 200 {object} WHIPServerInfo
// @Security ApiKeyAuth
// @Router /api/v3/whip/url [get]
func (h *WHIPHandler) GetServerURL(c echo.Context) error {
publicBase, localBase, err := h.whipPublicBase()
if err != nil {
return c.JSON(http.StatusServiceUnavailable, map[string]string{"error": err.Error()})
}
var token string
if h.config != nil {
token = h.config.GetActive().WHIP.Token
}
return c.JSON(http.StatusOK, WHIPServerInfo{
BasePublishURL: fmt.Sprintf("%s/whip/", publicBase),
BaseSDPURL: fmt.Sprintf("%s/whip/", localBase),
HasToken: token != "",
ExampleOBS: fmt.Sprintf("%s/whip/<stream-key>", publicBase),
InputAddressTemplate: "{whip:name=<stream-key>}",
})
}
// WHIPServerInfo holds the base URLs and metadata shown in the WHIP Server panel.
type WHIPServerInfo struct {
// BasePublishURL is the base URL for OBS. Append the stream key.
BasePublishURL string `json:"base_publish_url"`
// BaseSDPURL is the base internal SDP relay URL.
BaseSDPURL string `json:"base_sdp_url"`
// HasToken indicates whether a bearer token is required.
HasToken bool `json:"has_token"`
// ExampleOBS shows an example URL with placeholder stream key.
ExampleOBS string `json:"example_obs_url"`
// InputAddressTemplate is the process input address template for WHIP.
InputAddressTemplate string `json:"input_address_template"`
}
// ListChannels lists all currently active WHIP publishers.
// @Summary List all publishing WHIP streams
// @Description List all currently active WHIP (WebRTC HTTP Ingestion Protocol) streams.
// @Tags v16.?.?
// @ID whip-3-list-channels
// @Produce json
// @Success 200 {array} WHIPChannel
// @Security ApiKeyAuth
// @Router /api/v3/whip [get]
func (h *WHIPHandler) ListChannels(c echo.Context) error {
channels := h.whip.Channels()
list := make([]WHIPChannel, 0, len(channels.Publisher))
for name, since := range channels.Publisher {
list = append(list, WHIPChannel{
Name: name,
PublishedAt: since,
})
}
return c.JSON(http.StatusOK, list)
}

17
http/server.go
View File

@ -51,6 +51,7 @@ import (
"github.com/datarhei/core/v16/rtmp"
"github.com/datarhei/core/v16/session"
"github.com/datarhei/core/v16/srt"
"github.com/datarhei/core/v16/whip"
mwcache "github.com/datarhei/core/v16/http/middleware/cache"
mwcors "github.com/datarhei/core/v16/http/middleware/cors"
@ -86,6 +87,7 @@ type Config struct {
Cors CorsConfig
RTMP rtmp.Server
SRT srt.Server
WHIP whip.Server
JWT jwt.JWT
Config cfgstore.Store
Cache cache.Cacher
@ -120,6 +122,7 @@ type server struct {
playout *api.PlayoutHandler
rtmp *api.RTMPHandler
srt *api.SRTHandler
whip *api.WHIPHandler
config *api.ConfigHandler
session *api.SessionHandler
widget *api.WidgetHandler
@ -268,6 +271,13 @@ func NewServer(config Config) (Server, error) {
)
}
if config.WHIP != nil {
s.v3handler.whip = api.NewWHIP(
config.WHIP,
config.Config,
)
}
if config.Config != nil {
s.v3handler.config = api.NewConfig(
config.Config,
@ -628,6 +638,13 @@ func (s *server) setRoutesV3(v3 *echo.Group) {
v3.GET("/srt", s.v3handler.srt.ListChannels)
}
// v3 WHIP
if s.v3handler.whip != nil {
v3.GET("/whip", s.v3handler.whip.ListChannels)
v3.GET("/whip/url", s.v3handler.whip.GetServerURL)
v3.GET("/whip/:name/url", s.v3handler.whip.GetURL)
}
// v3 Config
if s.v3handler.config != nil {
v3.GET("/config", s.v3handler.config.Get)

19
restream/restream.go
View File

@ -1546,6 +1546,25 @@ func resolvePlaceholders(config *app.Config, r replace.Replacer) {
input.Address = r.Replace(input.Address, "fs:*", "", vars, config, "input")
input.Address = r.Replace(input.Address, "rtmp", "", vars, config, "input")
input.Address = r.Replace(input.Address, "srt", "", vars, config, "input")
input.Address = r.Replace(input.Address, "whip", "", vars, config, "input")
// WHIP SDP relay URLs (http://…/whip/…/sdp) require FFmpeg to open nested
// rtp:// and udp:// sub-protocols, which are blocked by default when the
// top-level protocol is HTTP. Inject -protocol_whitelist automatically so
// that both the probe and the live process can open the relay without
// requiring the user to set it manually.
if strings.Contains(input.Address, "/whip/") && strings.HasSuffix(input.Address, "/sdp") {
hasWhitelist := false
for _, opt := range input.Options {
if strings.Contains(opt, "protocol_whitelist") {
hasWhitelist = true
break
}
}
if !hasWhitelist {
input.Options = append([]string{"-protocol_whitelist", "file,crypto,http,rtp,udp,tcp"}, input.Options...)
}
}
for j, option := range input.Options {
// Replace any known placeholders

102
vendor/modules.txt vendored
View File

@ -255,6 +255,102 @@ github.com/minio/minio-go/v7/pkg/tags
# github.com/mitchellh/mapstructure v1.5.0
## explicit; go 1.14
github.com/mitchellh/mapstructure
# github.com/pion/datachannel v1.5.8
## explicit; go 1.19
github.com/pion/datachannel
# github.com/pion/dtls/v2 v2.2.12
## explicit; go 1.13
github.com/pion/dtls/v2
github.com/pion/dtls/v2/internal/ciphersuite
github.com/pion/dtls/v2/internal/ciphersuite/types
github.com/pion/dtls/v2/internal/closer
github.com/pion/dtls/v2/internal/util
github.com/pion/dtls/v2/pkg/crypto/ccm
github.com/pion/dtls/v2/pkg/crypto/ciphersuite
github.com/pion/dtls/v2/pkg/crypto/clientcertificate
github.com/pion/dtls/v2/pkg/crypto/elliptic
github.com/pion/dtls/v2/pkg/crypto/fingerprint
github.com/pion/dtls/v2/pkg/crypto/hash
github.com/pion/dtls/v2/pkg/crypto/prf
github.com/pion/dtls/v2/pkg/crypto/signature
github.com/pion/dtls/v2/pkg/crypto/signaturehash
github.com/pion/dtls/v2/pkg/protocol
github.com/pion/dtls/v2/pkg/protocol/alert
github.com/pion/dtls/v2/pkg/protocol/extension
github.com/pion/dtls/v2/pkg/protocol/handshake
github.com/pion/dtls/v2/pkg/protocol/recordlayer
# github.com/pion/ice/v2 v2.3.38
## explicit; go 1.13
github.com/pion/ice/v2
github.com/pion/ice/v2/internal/atomic
github.com/pion/ice/v2/internal/fakenet
github.com/pion/ice/v2/internal/stun
# github.com/pion/interceptor v0.1.29
## explicit; go 1.19
github.com/pion/interceptor
github.com/pion/interceptor/internal/ntp
github.com/pion/interceptor/internal/sequencenumber
github.com/pion/interceptor/pkg/nack
github.com/pion/interceptor/pkg/report
github.com/pion/interceptor/pkg/twcc
# github.com/pion/logging v0.2.2
## explicit; go 1.12
github.com/pion/logging
# github.com/pion/mdns v0.0.12
## explicit; go 1.19
github.com/pion/mdns
# github.com/pion/randutil v0.1.0
## explicit; go 1.14
github.com/pion/randutil
# github.com/pion/rtcp v1.2.14
## explicit; go 1.13
github.com/pion/rtcp
# github.com/pion/rtp v1.8.7
## explicit; go 1.19
github.com/pion/rtp
github.com/pion/rtp/codecs
github.com/pion/rtp/codecs/av1/obu
github.com/pion/rtp/codecs/vp9
# github.com/pion/sctp v1.8.19
## explicit; go 1.19
github.com/pion/sctp
# github.com/pion/sdp/v3 v3.0.9
## explicit; go 1.13
github.com/pion/sdp/v3
# github.com/pion/srtp/v2 v2.0.20
## explicit; go 1.14
github.com/pion/srtp/v2
# github.com/pion/stun v0.6.1
## explicit; go 1.12
github.com/pion/stun
github.com/pion/stun/internal/hmac
# github.com/pion/transport/v2 v2.2.10
## explicit; go 1.12
github.com/pion/transport/v2
github.com/pion/transport/v2/connctx
github.com/pion/transport/v2/deadline
github.com/pion/transport/v2/packetio
github.com/pion/transport/v2/replaydetector
github.com/pion/transport/v2/stdnet
github.com/pion/transport/v2/udp
github.com/pion/transport/v2/utils/xor
github.com/pion/transport/v2/vnet
# github.com/pion/turn/v2 v2.1.6
## explicit; go 1.13
github.com/pion/turn/v2
github.com/pion/turn/v2/internal/allocation
github.com/pion/turn/v2/internal/client
github.com/pion/turn/v2/internal/ipnet
github.com/pion/turn/v2/internal/proto
github.com/pion/turn/v2/internal/server
# github.com/pion/webrtc/v3 v3.3.6
## explicit; go 1.17
github.com/pion/webrtc/v3
github.com/pion/webrtc/v3/internal/fmtp
github.com/pion/webrtc/v3/internal/mux
github.com/pion/webrtc/v3/internal/util
github.com/pion/webrtc/v3/pkg/media
github.com/pion/webrtc/v3/pkg/rtcerr
# github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2
## explicit
github.com/pmezard/go-difflib/difflib
@ -343,6 +439,9 @@ github.com/vektah/gqlparser/v2/lexer
github.com/vektah/gqlparser/v2/parser
github.com/vektah/gqlparser/v2/validator
github.com/vektah/gqlparser/v2/validator/rules
# github.com/wlynxg/anet v0.0.3
## explicit; go 1.20
github.com/wlynxg/anet
# github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb
## explicit
github.com/xeipuuv/gojsonpointer
@ -392,6 +491,8 @@ golang.org/x/crypto/argon2
golang.org/x/crypto/blake2b
golang.org/x/crypto/cryptobyte
golang.org/x/crypto/cryptobyte/asn1
golang.org/x/crypto/curve25519
golang.org/x/crypto/curve25519/internal/field
golang.org/x/crypto/ocsp
golang.org/x/crypto/pbkdf2
golang.org/x/crypto/sha3
@ -403,6 +504,7 @@ golang.org/x/mod/semver
# golang.org/x/net v0.25.0
## explicit; go 1.18
golang.org/x/net/bpf
golang.org/x/net/dns/dnsmessage
golang.org/x/net/html
golang.org/x/net/html/atom
golang.org/x/net/http/httpguts

176
whip/pion_provider.go Normal file
View File

@ -0,0 +1,176 @@
package whip
import (
"fmt"
"net"
"strings"
"sync"
"time"
"github.com/pion/interceptor"
"github.com/pion/rtcp"
"github.com/pion/webrtc/v3"
)
// pionProvider implements WebRTCProvider using pion/webrtc v3.
// It handles full ICE + DTLS-SRTP negotiation and forwards the decrypted
// RTP to the local UDP relay sockets that FFmpeg reads.
type pionProvider struct {
mu sync.Mutex
pcs map[string]*webrtc.PeerConnection
}
// NewPionProvider returns a WebRTCProvider backed by pion/webrtc v3.
func NewPionProvider() WebRTCProvider {
return &pionProvider{
pcs: make(map[string]*webrtc.PeerConnection),
}
}
// OpenSession implements WebRTCProvider.
func (p *pionProvider) OpenSession(offerSDP string, videoPort, audioPort int) (string, error) {
m := &webrtc.MediaEngine{}
if err := m.RegisterDefaultCodecs(); err != nil {
return "", fmt.Errorf("pion: register codecs: %w", err)
}
ir := &interceptor.Registry{}
if err := webrtc.RegisterDefaultInterceptors(m, ir); err != nil {
return "", fmt.Errorf("pion: register interceptors: %w", err)
}
api := webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(ir))
pc, err := api.NewPeerConnection(webrtc.Configuration{})
if err != nil {
return "", fmt.Errorf("pion: new peer connection: %w", err)
}
_, err = pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo,
webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionRecvonly})
if err != nil {
pc.Close()
return "", fmt.Errorf("pion: add video transceiver: %w", err)
}
_, err = pc.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio,
webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionRecvonly})
if err != nil {
pc.Close()
return "", fmt.Errorf("pion: add audio transceiver: %w", err)
}
if err := pc.SetRemoteDescription(webrtc.SessionDescription{
Type: webrtc.SDPTypeOffer,
SDP: offerSDP,
}); err != nil {
pc.Close()
return "", fmt.Errorf("pion: set remote description: %w", err)
}
answer, err := pc.CreateAnswer(nil)
if err != nil {
pc.Close()
return "", fmt.Errorf("pion: create answer: %w", err)
}
gatherComplete := webrtc.GatheringCompletePromise(pc)
if err := pc.SetLocalDescription(answer); err != nil {
pc.Close()
return "", fmt.Errorf("pion: set local description: %w", err)
}
<-gatherComplete
finalSDP := pc.LocalDescription().SDP
videoDst, err := net.ResolveUDPAddr("udp4", fmt.Sprintf("127.0.0.1:%d", videoPort))
if err != nil {
pc.Close()
return "", fmt.Errorf("pion: resolve video relay addr: %w", err)
}
audioDst, err := net.ResolveUDPAddr("udp4", fmt.Sprintf("127.0.0.1:%d", audioPort))
if err != nil {
pc.Close()
return "", fmt.Errorf("pion: resolve audio relay addr: %w", err)
}
videoConn, err := net.DialUDP("udp4", nil, videoDst)
if err != nil {
pc.Close()
return "", fmt.Errorf("pion: dial video relay: %w", err)
}
audioConn, err := net.DialUDP("udp4", nil, audioDst)
if err != nil {
videoConn.Close()
pc.Close()
return "", fmt.Errorf("pion: dial audio relay: %w", err)
}
pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
var dst *net.UDPConn
if strings.EqualFold(track.Kind().String(), "video") {
dst = videoConn
// Request a keyframe immediately so consumers (ffprobe, FFmpeg) can
// determine the video resolution without waiting for the next IDR.
// Then send PLI every 2 s to keep keyframes flowing.
go func() {
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
for {
pc.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{
MediaSSRC: uint32(track.SSRC()),
}})
<-ticker.C
if pc.ConnectionState() != webrtc.PeerConnectionStateConnected {
return
}
}
}()
} else {
dst = audioConn
}
buf := make([]byte, 1500)
for {
n, _, err := track.Read(buf)
if err != nil {
return
}
dst.Write(buf[:n])
}
})
ufrag := extractICEUfrag(offerSDP)
p.mu.Lock()
p.pcs[ufrag] = pc
p.mu.Unlock()
pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
if state == webrtc.PeerConnectionStateClosed ||
state == webrtc.PeerConnectionStateFailed ||
state == webrtc.PeerConnectionStateDisconnected {
videoConn.Close()
audioConn.Close()
p.mu.Lock()
delete(p.pcs, ufrag)
p.mu.Unlock()
}
})
return finalSDP, nil
}
func extractICEUfrag(sdp string) string {
for _, line := range strings.Split(sdp, "\n") {
line = strings.TrimRight(line, "\r")
if strings.HasPrefix(line, "a=ice-ufrag:") {
return strings.TrimPrefix(line, "a=ice-ufrag:")
}
}
if len(sdp) > 32 {
return sdp[:32]
}
return sdp
}

852
whip/whip.go Normal file
View File

@ -0,0 +1,852 @@
// Package whip provides a WHIP (WebRTC HTTP Ingestion Protocol, RFC 9328) server.
//
// The server listens on an HTTP port for WHIP publishing requests. Clients
// (OBS Studio 30+, GStreamer, FFmpeg, browsers) POST an SDP offer to
// /whip/{streamID}. The server allocates UDP receive sockets, responds with
// an SDP answer containing the server's RTP receive ports, and forwards the
// received RTP packets to internal relay sockets that FFmpeg reads via a
// dynamically-generated SDP served at /whip/{streamID}/sdp.
//
// Usage in a Restreamer process config (input address template):
//
// http://localhost:8555/whip/{name}/sdp
//
// This uses the {whip} template, which expands to the above URL pattern.
//
// # WebRTC / ICE / DTLS-SRTP
//
// This implementation uses plain UDP RTP without ICE negotiation or DTLS-SRTP
// encryption. It is compatible with encoders that support plain-RTP mode
// (e.g. FFmpeg with -f whip pointed at this server, GStreamer without DTLS).
//
// For full browser and OBS WHIP support (which requires ICE + DTLS-SRTP),
// inject a WebRTCProvider implementation (e.g. via github.com/pion/webrtc/v3)
// through Config.WebRTCProvider. When set, the provider handles ICE negotiation
// and SRTP decryption and delivers decrypted RTP to the server's relay layer.
package whip
import (
"context"
"encoding/base64"
"fmt"
"io"
"net"
"net/http"
"strings"
"sync"
"time"
"github.com/datarhei/core/v16/log"
"github.com/datarhei/core/v16/session"
)
// ErrServerClosed is returned by ListenAndServe when the server is closed via Close().
var ErrServerClosed = http.ErrServerClosed
// WebRTCProvider is an optional interface for full WebRTC (ICE + DTLS-SRTP) support.
// When provided, the server delegates SDP negotiation and media reception to the
// provider rather than using plain UDP RTP.
//
// To add browser/OBS WHIP support, implement this interface using pion/webrtc/v3:
//
// type pionProvider struct { ... }
// func (p *pionProvider) OpenSession(offerSDP string, videoPort, audioPort int) (answerSDP string, err error) { ... }
type WebRTCProvider interface {
// OpenSession performs ICE + DTLS-SRTP negotiation.
// offerSDP is the SDP offer from the client.
// videoPort and audioPort are the local UDP ports where the provider should
// deliver decrypted RTP after the handshake.
// Returns the SDP answer to send back to the client, or an error.
OpenSession(offerSDP string, videoPort, audioPort int) (answerSDP string, err error)
}
// Config is the configuration for the WHIP server.
type Config struct {
// Addr is the HTTP listen address for the WHIP endpoint, e.g. ":8555".
Addr string
// Token is an optional bearer token required from WHIP clients.
// If empty, no authentication is required.
Token string
// Logger is optional.
Logger log.Logger
// Collector is optional. Used for session statistics.
Collector session.Collector
// WebRTCProvider is optional. When provided, it handles ICE negotiation and
// DTLS-SRTP for full browser/OBS WebRTC support. When nil, the server falls
// back to plain UDP RTP (no ICE, no encryption).
WebRTCProvider WebRTCProvider
}
// Server defines the WHIP server interface.
type Server interface {
// ListenAndServe starts the WHIP HTTP server. Blocks until Close() is called.
ListenAndServe() error
// Close shuts down the server and releases all resources.
Close()
// Channels returns the currently active (publishing) WHIP streams.
Channels() Channels
}
// Channels describes the active WHIP streams.
type Channels struct {
// Publisher maps stream name to its publish start time.
Publisher map[string]time.Time
}
// channel holds the relay sockets and state for a single WHIP stream.
type channel struct {
name string
// videoRelayPort / audioRelayPort: loopback UDP ports that FFmpeg reads from.
// Ports are pre-allocated by briefly binding to :0, then released so that
// FFmpeg (or any RTP consumer) can bind to them.
videoRelayPort int
audioRelayPort int
// videoRx / audioRx: UDP sockets that receive RTP from the WHIP client.
// These are present only while a client is actively publishing.
videoRx *net.UDPConn
audioRx *net.UDPConn
// codec parameters negotiated from the SDP offer.
videoPayloadType int
videoCodec string
videoClockRate int
videoFmtp string // extra fmtp params from offer (e.g. sprop-parameter-sets)
// SPS/PPS captured from in-band RTP stream (populated after first keyframe).
videoSPS []byte
videoPPS []byte
spropped bool // true once sprop-parameter-sets has been built from in-band NALs
audioPayloadType int
audioCodec string
audioClockRate int
audioChannels int
publishedAt time.Time
collector session.Collector
rxCancel context.CancelFunc
lock sync.RWMutex
}
func newChannel(name string, collector session.Collector) (*channel, error) {
ch := &channel{
name: name,
collector: collector,
// Defaults matching the most common WebRTC codecs.
videoPayloadType: 96,
videoCodec: "H264",
videoClockRate: 90000,
audioPayloadType: 111,
audioCodec: "opus",
audioClockRate: 48000,
audioChannels: 2,
}
// Pre-allocate relay port numbers. We bind briefly to get an ephemeral port
// assigned by the OS, then immediately close so FFmpeg can bind to that port.
vp, err := allocateRelayPort()
if err != nil {
return nil, fmt.Errorf("whip: allocate video relay port: %w", err)
}
ch.videoRelayPort = vp
ap, err := allocateRelayPort()
if err != nil {
return nil, fmt.Errorf("whip: allocate audio relay port: %w", err)
}
ch.audioRelayPort = ap
return ch, nil
}
// allocateRelayPort binds a UDP socket on a random loopback port, records the
// port number, then immediately closes the socket so that an external process
// (FFmpeg) can bind to the same port shortly afterwards.
func allocateRelayPort() (int, error) {
conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
if err != nil {
return 0, err
}
port := conn.LocalAddr().(*net.UDPAddr).Port
conn.Close()
return port, nil
}
// relayVideoPort returns the relay port for the video RTP stream (FFmpeg reads this).
func (ch *channel) relayVideoPort() int {
return ch.videoRelayPort
}
// relayAudioPort returns the relay port for the audio RTP stream (FFmpeg reads this).
func (ch *channel) relayAudioPort() int {
return ch.audioRelayPort
}
// ffmpegSDP returns the SDP describing the relay streams for FFmpeg to consume.
func (ch *channel) ffmpegSDP() string {
ch.lock.RLock()
vpt := ch.videoPayloadType
vpc := ch.videoCodec
vclk := ch.videoClockRate
vfmtp := ch.videoFmtp
apt := ch.audioPayloadType
apc := ch.audioCodec
aclk := ch.audioClockRate
ach := ch.audioChannels
ch.lock.RUnlock()
videoPort := ch.relayVideoPort()
audioPort := ch.relayAudioPort()
sdp := "v=0\r\n"
sdp += "o=- 0 0 IN IP4 127.0.0.1\r\n"
sdp += "s=WHIP Stream\r\n"
sdp += "c=IN IP4 127.0.0.1\r\n"
sdp += "t=0 0\r\n"
sdp += fmt.Sprintf("m=video %d RTP/AVP %d\r\n", videoPort, vpt)
sdp += fmt.Sprintf("a=rtpmap:%d %s/%d\r\n", vpt, vpc, vclk)
if vfmtp != "" {
sdp += fmt.Sprintf("a=fmtp:%d %s\r\n", vpt, vfmtp)
} else if vpc == "H264" {
sdp += fmt.Sprintf("a=fmtp:%d packetization-mode=1\r\n", vpt)
}
sdp += fmt.Sprintf("m=audio %d RTP/AVP %d\r\n", audioPort, apt)
if ach > 1 {
sdp += fmt.Sprintf("a=rtpmap:%d %s/%d/%d\r\n", apt, apc, aclk, ach)
} else {
sdp += fmt.Sprintf("a=rtpmap:%d %s/%d\r\n", apt, apc, aclk)
}
return sdp
}
// startRx allocates receive sockets, starts the relay goroutines, and begins
// forwarding incoming RTP packets from the WHIP client to the relay sockets.
// It returns the SDP answer to send to the WHIP client.
func (ch *channel) startRx(serverIP string, provider WebRTCProvider, offerSDP string) (answerSDP string, videoRxPort, audioRxPort int, err error) {
ch.lock.Lock()
defer ch.lock.Unlock()
// Stop any previous rx session.
if ch.rxCancel != nil {
ch.rxCancel()
}
if ch.videoRx != nil {
ch.videoRx.Close()
}
if ch.audioRx != nil {
ch.audioRx.Close()
}
// Parse codec parameters from the SDP offer (best-effort).
ch.updateCodecsFromSDP(offerSDP)
// Allocate rx sockets for the WHIP client to send RTP to.
ch.videoRx, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP(serverIP), Port: 0})
if err != nil {
return "", 0, 0, fmt.Errorf("whip: allocate video rx socket: %w", err)
}
ch.audioRx, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP(serverIP), Port: 0})
if err != nil {
ch.videoRx.Close()
ch.videoRx = nil
return "", 0, 0, fmt.Errorf("whip: allocate audio rx socket: %w", err)
}
videoRxPort = ch.videoRx.LocalAddr().(*net.UDPAddr).Port
audioRxPort = ch.audioRx.LocalAddr().(*net.UDPAddr).Port
ctx, cancel := context.WithCancel(context.Background())
ch.rxCancel = cancel
ch.publishedAt = time.Now()
// Build the SDP answer for the WHIP client.
if provider != nil {
answerSDP, err = provider.OpenSession(offerSDP, videoRxPort, audioRxPort)
if err != nil {
cancel()
ch.videoRx.Close()
ch.audioRx.Close()
ch.videoRx = nil
ch.audioRx = nil
return "", 0, 0, fmt.Errorf("whip: WebRTC provider: %w", err)
}
} else {
answerSDP = ch.buildPlainRTPAnswer(serverIP, videoRxPort, audioRxPort)
}
// Start relay goroutines (send decrypted RTP to the loopback ports where FFmpeg listens).
go ch.relayUDP(ctx, ch.videoRx, ch.videoRelayPort, ch.snoopH264NALs)
go ch.relayUDP(ctx, ch.audioRx, ch.audioRelayPort, nil)
// Track session bandwidth ingress.
if ch.collector.IsCollectableIP(serverIP) {
ch.collector.RegisterAndActivate(ch.name, ch.name, "publish:"+ch.name, serverIP)
}
return answerSDP, videoRxPort, audioRxPort, nil
}
// stopRx tears down the receive side of the channel.
func (ch *channel) stopRx() {
ch.lock.Lock()
defer ch.lock.Unlock()
if ch.rxCancel != nil {
ch.rxCancel()
ch.rxCancel = nil
}
if ch.videoRx != nil {
ch.videoRx.Close()
ch.videoRx = nil
}
if ch.audioRx != nil {
ch.audioRx.Close()
ch.audioRx = nil
}
ch.publishedAt = time.Time{}
// Reset SPS/PPS snoop so next publisher gets fresh capture.
ch.videoSPS = nil
ch.videoPPS = nil
ch.spropped = false
}
// close releases all resources held by the channel.
func (ch *channel) close() {
ch.stopRx()
}
// isPublishing returns true when a WHIP client is currently sending to the channel.
func (ch *channel) isPublishing() bool {
ch.lock.RLock()
defer ch.lock.RUnlock()
return !ch.publishedAt.IsZero()
}
// publishedSince returns when the current publisher started.
func (ch *channel) publishedSince() time.Time {
ch.lock.RLock()
defer ch.lock.RUnlock()
return ch.publishedAt
}
// relayUDP reads RTP packets from rx and forwards them to relayPort on loopback.
// FFmpeg (or any RTP consumer) is expected to be listening on relayPort.
// onPacket, if non-nil, is called with a copy of each RTP packet before forwarding.
func (ch *channel) relayUDP(ctx context.Context, rx *net.UDPConn, relayPort int, onPacket func([]byte)) {
buf := make([]byte, 1500)
// Send to loopback so FFmpeg (listening on relayPort) receives.
dst := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: relayPort}
sendConn, err := net.DialUDP("udp4", nil, dst)
if err != nil {
return
}
defer sendConn.Close()
for {
select {
case <-ctx.Done():
return
default:
}
rx.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
n, _, err := rx.ReadFromUDP(buf)
if err != nil {
if isNetTimeout(err) {
continue
}
return
}
sendConn.Write(buf[:n])
if onPacket != nil {
pkt := make([]byte, n)
copy(pkt, buf[:n])
go onPacket(pkt)
}
}
}
// snoopH264NALs inspects an RTP packet for H.264 SPS (type 7) and PPS (type 8)
// NAL units. Once both are found, it updates ch.videoFmtp with sprop-parameter-sets
// so that FFmpeg can determine the video resolution from the relay SDP.
func (ch *channel) snoopH264NALs(pkt []byte) {
ch.lock.RLock()
done := ch.spropped
ch.lock.RUnlock()
if done {
return
}
nals := parseH264NALsFromRTP(pkt)
if len(nals) == 0 {
return
}
ch.lock.Lock()
defer ch.lock.Unlock()
if ch.spropped {
return
}
for _, nal := range nals {
if len(nal) == 0 {
continue
}
nalType := nal[0] & 0x1f
if nalType == 7 && ch.videoSPS == nil {
ch.videoSPS = nal
} else if nalType == 8 && ch.videoPPS == nil {
ch.videoPPS = nal
}
}
if ch.videoSPS != nil && ch.videoPPS != nil {
ch.videoFmtp = "sprop-parameter-sets=" +
base64.StdEncoding.EncodeToString(ch.videoSPS) + "," +
base64.StdEncoding.EncodeToString(ch.videoPPS) +
";packetization-mode=1"
ch.spropped = true
}
}
// parseH264NALsFromRTP returns the H.264 NAL units contained in a raw RTP packet.
// It handles Single NAL unit packets and STAP-A aggregation packets.
// FU-A fragmented packets are skipped (SPS/PPS are never fragmented in practice).
func parseH264NALsFromRTP(rtpPkt []byte) [][]byte {
if len(rtpPkt) < 13 {
return nil
}
// Skip fixed 12-byte RTP header + optional CSRC list.
cc := int(rtpPkt[0] & 0x0f)
offset := 12 + cc*4
// Skip RTP extension header if present.
if rtpPkt[0]&0x10 != 0 {
if offset+4 > len(rtpPkt) {
return nil
}
extLen := int(rtpPkt[offset+2])<<8 | int(rtpPkt[offset+3])
offset += 4 + extLen*4
}
if offset >= len(rtpPkt) {
return nil
}
payload := rtpPkt[offset:]
nalType := payload[0] & 0x1f
switch {
case nalType >= 1 && nalType <= 23: // Single NAL unit packet
return [][]byte{payload}
case nalType == 24: // STAP-A
var nals [][]byte
pos := 1
for pos+2 <= len(payload) {
size := int(payload[pos])<<8 | int(payload[pos+1])
pos += 2
if size == 0 || pos+size > len(payload) {
break
}
nals = append(nals, payload[pos:pos+size])
pos += size
}
return nals
}
// FU-A and other types: ignore
return nil
}
// buildPlainRTPAnswer constructs a minimal SDP answer for plain (non-DTLS) RTP.
func (ch *channel) buildPlainRTPAnswer(serverIP string, videoPort, audioPort int) string {
sdp := "v=0\r\n"
sdp += "o=- 0 0 IN IP4 " + serverIP + "\r\n"
sdp += "s=WHIP Answer\r\n"
sdp += "c=IN IP4 " + serverIP + "\r\n"
sdp += "t=0 0\r\n"
sdp += fmt.Sprintf("m=video %d RTP/AVP %d\r\n", videoPort, ch.videoPayloadType)
sdp += fmt.Sprintf("a=rtpmap:%d %s/%d\r\n", ch.videoPayloadType, ch.videoCodec, ch.videoClockRate)
sdp += "a=recvonly\r\n"
sdp += fmt.Sprintf("m=audio %d RTP/AVP %d\r\n", audioPort, ch.audioPayloadType)
if ch.audioChannels > 1 {
sdp += fmt.Sprintf("a=rtpmap:%d %s/%d/%d\r\n", ch.audioPayloadType, ch.audioCodec, ch.audioClockRate, ch.audioChannels)
} else {
sdp += fmt.Sprintf("a=rtpmap:%d %s/%d\r\n", ch.audioPayloadType, ch.audioCodec, ch.audioClockRate)
}
sdp += "a=recvonly\r\n"
return sdp
}
// updateCodecsFromSDP performs a best-effort parse of the SDP offer to extract
// the first video and audio payload types and codec names.
func (ch *channel) updateCodecsFromSDP(sdp string) {
type mediaState struct {
inVideo bool
inAudio bool
}
var ms mediaState
for _, line := range strings.Split(sdp, "\n") {
line = strings.TrimRight(line, "\r")
if strings.HasPrefix(line, "m=video") {
ms.inVideo = true
ms.inAudio = false
// Extract payload type from "m=video 9 ... 96 97 ..."
parts := strings.Fields(line)
if len(parts) >= 4 {
var pt int
if _, err := fmt.Sscanf(parts[3], "%d", &pt); err == nil {
ch.videoPayloadType = pt
}
}
} else if strings.HasPrefix(line, "m=audio") {
ms.inAudio = true
ms.inVideo = false
parts := strings.Fields(line)
if len(parts) >= 4 {
var pt int
if _, err := fmt.Sscanf(parts[3], "%d", &pt); err == nil {
ch.audioPayloadType = pt
}
}
} else if strings.HasPrefix(line, "a=fmtp:") {
// Format: a=fmtp:<pt> <params>
var pt int
if _, err := fmt.Sscanf(line, "a=fmtp:%d ", &pt); err == nil {
params := strings.SplitN(line, " ", 2)
if len(params) == 2 && ms.inVideo && pt == ch.videoPayloadType {
ch.videoFmtp = params[1]
}
}
} else if strings.HasPrefix(line, "a=rtpmap:") {
// Format: a=rtpmap:<pt> <codec>/<clockrate>[/<channels>]
var pt int
var rest string
if _, err := fmt.Sscanf(line, "a=rtpmap:%d %s", &pt, &rest); err != nil {
continue
}
parts := strings.Split(rest, "/")
codec := parts[0]
var clk int
if len(parts) >= 2 {
fmt.Sscanf(parts[1], "%d", &clk)
}
var channels int = 1
if len(parts) >= 3 {
fmt.Sscanf(parts[2], "%d", &channels)
}
if ms.inVideo && pt == ch.videoPayloadType {
ch.videoCodec = codec
if clk > 0 {
ch.videoClockRate = clk
}
} else if ms.inAudio && pt == ch.audioPayloadType {
ch.audioCodec = codec
if clk > 0 {
ch.audioClockRate = clk
}
ch.audioChannels = channels
}
}
}
}
// --- Server implementation ---
type server struct {
addr string
token string
logger log.Logger
collector session.Collector
provider WebRTCProvider
httpServer *http.Server
channels map[string]*channel
lock sync.RWMutex
}
// New creates a new WHIP server from the given config.
func New(config Config) (Server, error) {
if config.Addr == "" {
return nil, fmt.Errorf("whip: listen address is required")
}
s := &server{
addr: config.Addr,
token: config.Token,
logger: config.Logger,
collector: config.Collector,
provider: config.WebRTCProvider,
channels: make(map[string]*channel),
}
if s.logger == nil {
s.logger = log.New("")
}
if s.collector == nil {
s.collector = session.NewNullCollector()
}
mux := http.NewServeMux()
mux.HandleFunc("/whip/", s.dispatch)
s.httpServer = &http.Server{
Addr: s.addr,
Handler: mux,
ReadHeaderTimeout: 10 * time.Second,
IdleTimeout: 60 * time.Second,
}
return s, nil
}
// ListenAndServe starts the WHIP HTTP server.
func (s *server) ListenAndServe() error {
s.logger.Info().Log("Server started")
err := s.httpServer.ListenAndServe()
if err == http.ErrServerClosed {
return ErrServerClosed
}
return err
}
// Close shuts down the WHIP server.
func (s *server) Close() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
s.httpServer.Shutdown(ctx)
s.lock.Lock()
defer s.lock.Unlock()
for _, ch := range s.channels {
ch.close()
}
s.channels = make(map[string]*channel)
s.logger.Info().Log("Server closed")
}
// Channels returns the current active publishers.
func (s *server) Channels() Channels {
s.lock.RLock()
defer s.lock.RUnlock()
c := Channels{Publisher: make(map[string]time.Time)}
for name, ch := range s.channels {
if ch.isPublishing() {
c.Publisher[name] = ch.publishedSince()
}
}
return c
}
// dispatch routes incoming HTTP requests to the appropriate handler.
func (s *server) dispatch(w http.ResponseWriter, r *http.Request) {
// Trim "/whip/" prefix.
path := strings.TrimPrefix(r.URL.Path, "/whip/")
path = strings.TrimSuffix(path, "/")
// GET /whip/{name}/sdp → SDP for FFmpeg to consume the relay.
if strings.HasSuffix(path, "/sdp") && r.Method == http.MethodGet {
name := strings.TrimSuffix(path, "/sdp")
if !isValidStreamName(name) {
http.Error(w, "invalid stream name", http.StatusBadRequest)
return
}
s.handleSDPRead(w, r, name)
return
}
// All other methods require token validation.
if s.token != "" && !s.checkToken(r) {
w.Header().Set("WWW-Authenticate", `Bearer realm="WHIP"`)
http.Error(w, "Unauthorized", http.StatusUnauthorized)
return
}
name := path
if !isValidStreamName(name) {
http.Error(w, "invalid stream name", http.StatusBadRequest)
return
}
switch r.Method {
case http.MethodPost:
s.handlePublish(w, r, name)
case http.MethodDelete:
s.handleDelete(w, r, name)
case http.MethodPatch:
// Trickle ICE (RFC 9328 §4.3) — respond 405 if not supported.
if s.provider == nil {
http.Error(w, "trickle ICE not supported in plain-RTP mode", http.StatusMethodNotAllowed)
return
}
http.Error(w, "trickle ICE not implemented", http.StatusNotImplemented)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
// handleSDPRead serves the SDP file that FFmpeg uses to read the WHIP relay stream.
// The relay sockets (and thus their ports) are pre-allocated here if the channel
// does not yet exist, so that FFmpeg can bind to them before a WHIP client arrives.
// If a publisher is active, this handler waits up to 5 seconds for in-band SPS/PPS
// to be captured so that FFmpeg can determine the video resolution immediately.
func (s *server) handleSDPRead(w http.ResponseWriter, r *http.Request, name string) {
ch := s.getOrCreateChannel(name)
// If a publisher is active, wait for SPS/PPS to be snooped from the RTP stream.
if ch.isPublishing() {
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
ch.lock.RLock()
done := ch.spropped
ch.lock.RUnlock()
if done {
break
}
time.Sleep(50 * time.Millisecond)
}
}
sdp := ch.ffmpegSDP()
w.Header().Set("Content-Type", "application/sdp")
w.Header().Set("Cache-Control", "no-cache")
w.WriteHeader(http.StatusOK)
io.WriteString(w, sdp)
}
// handlePublish handles the WHIP POST request (SDP offer/answer exchange).
func (s *server) handlePublish(w http.ResponseWriter, r *http.Request, name string) {
ct := r.Header.Get("Content-Type")
if !strings.Contains(ct, "application/sdp") {
http.Error(w, "Content-Type must be application/sdp", http.StatusUnsupportedMediaType)
return
}
body, err := io.ReadAll(io.LimitReader(r.Body, 16*1024))
if err != nil {
http.Error(w, "failed to read SDP offer", http.StatusBadRequest)
return
}
offerSDP := string(body)
// Identify the client IP so we can send RTP back on the same interface.
clientIP, _, _ := net.SplitHostPort(r.RemoteAddr)
serverIP := s.serverIPFor(clientIP)
ch := s.getOrCreateChannel(name)
answerSDP, _, _, err := ch.startRx(serverIP, s.provider, offerSDP)
if err != nil {
s.logger.Error().WithField("stream", name).WithField("error", err).Log("Failed to start WHIP session")
http.Error(w, "internal server error", http.StatusInternalServerError)
return
}
s.logger.Info().WithField("stream", name).WithField("client", clientIP).Log("WHIP publisher connected")
// Respond per RFC 9328: 201 Created with Location header and SDP answer.
location := "/whip/" + name
w.Header().Set("Content-Type", "application/sdp")
w.Header().Set("Location", location)
w.WriteHeader(http.StatusCreated)
io.WriteString(w, answerSDP)
}
// handleDelete ends a WHIP session (RFC 9328 §4.2).
func (s *server) handleDelete(w http.ResponseWriter, r *http.Request, name string) {
s.lock.RLock()
ch, ok := s.channels[name]
s.lock.RUnlock()
if !ok || !ch.isPublishing() {
http.Error(w, "stream not found", http.StatusNotFound)
return
}
ch.stopRx()
clientIP, _, _ := net.SplitHostPort(r.RemoteAddr)
s.logger.Info().WithField("stream", name).WithField("client", clientIP).Log("WHIP publisher disconnected")
w.WriteHeader(http.StatusOK)
}
// getOrCreateChannel returns the channel for the given name, creating it if needed.
func (s *server) getOrCreateChannel(name string) *channel {
s.lock.Lock()
defer s.lock.Unlock()
ch, ok := s.channels[name]
if !ok {
var err error
ch, err = newChannel(name, s.collector)
if err != nil {
s.logger.Error().WithField("stream", name).WithField("error", err).Log("Failed to create channel")
// Return a dummy channel rather than nil to avoid panics; it won't relay.
ch = &channel{name: name}
}
s.channels[name] = ch
}
return ch
}
// checkToken validates the bearer token from the Authorization header or ?token= query param.
func (s *server) checkToken(r *http.Request) bool {
if t := r.URL.Query().Get("token"); t != "" {
return t == s.token
}
auth := r.Header.Get("Authorization")
if strings.HasPrefix(auth, "Bearer ") {
return strings.TrimPrefix(auth, "Bearer ") == s.token
}
return false
}
// serverIPFor returns the server-side IP to use in SDP based on the client IP.
// For loopback clients, returns 127.0.0.1; for external clients returns 0.0.0.0
// (the OS will choose an appropriate local address when binding).
func (s *server) serverIPFor(clientIP string) string {
ip := net.ParseIP(clientIP)
if ip == nil || ip.IsLoopback() {
return "127.0.0.1"
}
return "0.0.0.0"
}
// isValidStreamName checks that a stream name contains only safe characters.
func isValidStreamName(name string) bool {
if name == "" {
return false
}
for _, c := range name {
if !((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
(c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.') {
return false
}
}
return true
}
// isNetTimeout returns true for Go net timeout errors.
func isNetTimeout(err error) bool {
if ne, ok := err.(net.Error); ok {
return ne.Timeout()
}
return false
}