gilles/go2rtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1762 from seydx/preload

Browse Source 
Preload Streams
This commit is contained in:
Alex X
2025-10-01 17:22:19 +03:00
committed by GitHub
parent 549da0257e 4dd1f73a18
commit 37d7409e82
6 changed files with 194 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+21
View File
@@ -76,6 +76,7 @@ Ultimate camera streaming application with support for RTSP, WebRTC, HomeKit, FF
* [Incoming sources](#incoming-sources)
* [Stream to camera](#stream-to-camera)
* [Publish stream](#publish-stream)
* [Preload stream](#preload-stream)
* [Module: API](#module-api)
* [Module: RTSP](#module-rtsp)
* [Module: RTMP](#module-rtmp)
@@ -835,6 +836,26 @@ streams:
- **Telegram Desktop App** > Any public or private channel or group (where you admin) > Live stream > Start with... > Start streaming.
- **YouTube** > Create > Go live > Stream latency: Ultra low-latency > Copy: Stream URL + Stream key.
### Preload stream
You can preload any stream on go2rtc start. This is useful for cameras that take a long time to start up.
```yaml
preload:
camera1: # default: video&audio = ANY
camera2: "video" # preload only video track
camera3: "video=h264&audio=opus" # preload H264 video and OPUS audio
streams:
camera1:
- rtsp://192.168.1.100/stream
camera2:
- rtsp://192.168.1.101/stream
camera3:
- rtsp://192.168.1.102/h265stream
- ffmpeg:camera3#video=h264#audio=opus#hardware
```
### Module: API
The HTTP API is the main part for interacting with the application. Default address: `http://localhost:1984/`.
api/openapi.yaml
+48
View File
@@ -237,6 +237,54 @@ paths:
/api/preload:
put:
summary: Preload new stream
tags: [ Streams list ]
parameters:
- name: src
in: query
description: Stream source (name)
required: true
schema: { type: string }
example: "camera1"
- name: video
in: query
description: Video codecs filter
required: false
schema: { type: string }
example: all,h264,h265,...
- name: audio
in: query
description: Audio codecs filter
required: false
schema: { type: string }
example: all,aac,opus,...
- name: microphone
in: query
description: Microphone codecs filter
required: false
schema: { type: string }
example: all,aac,opus,...
responses:
default:
description: Default response
delete:
summary: Delete preloaded stream
tags: [ Streams list ]
parameters:
- name: src
in: query
description: Stream source (name)
required: true
schema: { type: string }
example: "camera1"
responses:
default:
description: Default response
/api/streams?src={src}:
get:
summary: Get stream info in JSON format
internal/streams/api.go
+50 -1
View File
@@ -5,6 +5,7 @@ import (
"github.com/AlexxIT/go2rtc/internal/api"
"github.com/AlexxIT/go2rtc/internal/app"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/probe"
)
@@ -27,7 +28,7 @@ func apiStreams(w http.ResponseWriter, r *http.Request) {
return
}
cons := probe.NewProbe(query)
cons := probe.Create("probe", query)
if len(cons.Medias) != 0 {
cons.WithRequest(r)
if err := stream.AddConsumer(cons); err != nil {
@@ -122,3 +123,51 @@ func apiStreamsDOT(w http.ResponseWriter, r *http.Request) {
api.Response(w, dot, "text/vnd.graphviz")
}
func apiPreload(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
src := query.Get("src")
// check if stream exists
stream := Get(src)
if stream == nil {
http.Error(w, "", http.StatusNotFound)
return
}
switch r.Method {
case "PUT":
// it's safe to delete from map while iterating
for k := range query {
switch k {
case core.KindVideo, core.KindAudio, "microphone":
default:
delete(query, k)
}
}
rawQuery := query.Encode()
if err := AddPreload(stream, rawQuery); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if err := app.PatchConfig([]string{"preload", src}, rawQuery); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
case "DELETE":
if err := DelPreload(stream); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if err := app.PatchConfig([]string{"preload", src}, nil); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
default:
http.Error(w, "", http.StatusMethodNotAllowed)
}
}
internal/streams/preload.go
+58
View File
@@ -0,0 +1,58 @@
package streams
import (
"errors"
"net/url"
"sync"
"github.com/AlexxIT/go2rtc/pkg/probe"
)
var preloads = map[*Stream]*probe.Probe{}
var preloadsMu sync.Mutex
func Preload(stream *Stream, rawQuery string) {
if err := AddPreload(stream, rawQuery); err != nil {
log.Error().Err(err).Caller().Send()
}
}
func AddPreload(stream *Stream, rawQuery string) error {
if rawQuery == "" {
rawQuery = "video&audio"
}
query, err := url.ParseQuery(rawQuery)
if err != nil {
return err
}
preloadsMu.Lock()
defer preloadsMu.Unlock()
if cons := preloads[stream]; cons != nil {
stream.RemoveConsumer(cons)
}
cons := probe.Create("preload", query)
if err = stream.AddConsumer(cons); err != nil {
return err
}
preloads[stream] = cons
return nil
}
func DelPreload(stream *Stream) error {
preloadsMu.Lock()
defer preloadsMu.Unlock()
if cons := preloads[stream]; cons != nil {
stream.RemoveConsumer(cons)
delete(preloads, stream)
return nil
}
return errors.New("streams: preload not found")
}
internal/streams/streams.go
+11 -3
View File
@@ -14,8 +14,9 @@ import (
func Init() {
var cfg struct {
Streams map[string]any `yaml:"streams"`
Publish map[string]any `yaml:"publish"`
Streams map[string]any `yaml:"streams"`
Publish map[string]any `yaml:"publish"`
Preload map[string]string `yaml:"preload"`
}
app.LoadConfig(&cfg)
@@ -28,17 +29,24 @@ func Init() {
api.HandleFunc("api/streams", apiStreams)
api.HandleFunc("api/streams.dot", apiStreamsDOT)
api.HandleFunc("api/preload", apiPreload)
if cfg.Publish == nil {
if cfg.Publish == nil && cfg.Preload == nil {
return
}
time.AfterFunc(time.Second, func() {
// range for nil map is OK
for name, dst := range cfg.Publish {
if stream := Get(name); stream != nil {
Publish(stream, dst)
}
}
for name, rawQuery := range cfg.Preload {
if stream := Get(name); stream != nil {
Preload(stream, rawQuery)
}
}
})
}
pkg/probe/producer.go → pkg/probe/consumer.go
+6 -27
View File
@@ -11,7 +11,7 @@ type Probe struct {
core.Connection
}
func NewProbe(query url.Values) *Probe {
func Create(name string, query url.Values) *Probe {
medias := core.ParseQuery(query)
for _, value := range query["microphone"] {
@@ -32,39 +32,18 @@ func NewProbe(query url.Values) *Probe {
return &Probe{
Connection: core.Connection{
ID: core.NewID(),
FormatName: "probe",
FormatName: name,
Medias: medias,
},
}
}
func (p *Probe) GetMedias() []*core.Media {
return p.Medias
}
func (p *Probe) AddTrack(media *core.Media, codec *core.Codec, track *core.Receiver) error {
sender := core.NewSender(media, track.Codec)
sender.Bind(track)
sender.Handler = func(pkt *core.Packet) {
p.Send += len(pkt.Payload)
}
sender.HandleRTP(track)
p.Senders = append(p.Senders, sender)
return nil
}
func (p *Probe) GetTrack(media *core.Media, codec *core.Codec) (*core.Receiver, error) {
receiver := core.NewReceiver(media, codec)
p.Receivers = append(p.Receivers, receiver)
return receiver, nil
}
func (p *Probe) Start() error {
return nil
}
func (p *Probe) Stop() error {
for _, receiver := range p.Receivers {
receiver.Close()
}
for _, sender := range p.Senders {
sender.Close()
}
return nil
}