Rewrite Strix from scratch as single binary
Complete architecture rewrite following go2rtc patterns: - pkg/ for pure logic (camdb, tester, probe, generate) - internal/ for application glue with Init() modules - Single HTTP server on :4567 with all endpoints - zerolog with password masking and memory ring buffer - Environment-based config only (no YAML files) API endpoints: /api/search, /api/streams, /api/test, /api/probe, /api/generate, /api/health, /api/log Dependencies: go2rtc v1.9.14, go-sqlite3, miekg/dns, zerolog
This commit is contained in:
eduard256
@@ -0,0 +1,97 @@
|
||||
package tester
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const SessionTTL = 30 * time.Minute
|
||||
|
||||
type Session struct {
|
||||
ID string `json:"session_id"`
|
||||
Status string `json:"status"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
ExpiresAt time.Time `json:"expires_at,omitempty"`
|
||||
Total int `json:"total"`
|
||||
Tested int `json:"tested"`
|
||||
Alive int `json:"alive"`
|
||||
WithScreen int `json:"with_screenshot"`
|
||||
Results []*Result `json:"results"`
|
||||
Screenshots [][]byte `json:"-"`
|
||||
|
||||
cancel chan struct{}
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
Source string `json:"source"`
|
||||
Screenshot string `json:"screenshot,omitempty"`
|
||||
Codecs []string `json:"codecs,omitempty"`
|
||||
LatencyMs int64 `json:"latency_ms,omitempty"`
|
||||
Skipped bool `json:"skipped,omitempty"`
|
||||
}
|
||||
|
||||
func NewSession(id string, total int) *Session {
|
||||
return &Session{
|
||||
ID: id,
|
||||
Status: "running",
|
||||
CreatedAt: time.Now(),
|
||||
Total: total,
|
||||
cancel: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Session) AddResult(r *Result) {
|
||||
s.mu.Lock()
|
||||
s.Results = append(s.Results, r)
|
||||
s.Alive++
|
||||
if r.Screenshot != "" {
|
||||
s.WithScreen++
|
||||
}
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *Session) AddTested() {
|
||||
s.mu.Lock()
|
||||
s.Tested++
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *Session) AddScreenshot(data []byte) int {
|
||||
s.mu.Lock()
|
||||
idx := len(s.Screenshots)
|
||||
s.Screenshots = append(s.Screenshots, data)
|
||||
s.mu.Unlock()
|
||||
return idx
|
||||
}
|
||||
|
||||
func (s *Session) GetScreenshot(idx int) []byte {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if idx < 0 || idx >= len(s.Screenshots) {
|
||||
return nil
|
||||
}
|
||||
return s.Screenshots[idx]
|
||||
}
|
||||
|
||||
func (s *Session) Done() {
|
||||
s.mu.Lock()
|
||||
s.Status = "done"
|
||||
s.ExpiresAt = time.Now().Add(SessionTTL)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *Session) Cancel() {
|
||||
select {
|
||||
case <-s.cancel:
|
||||
default:
|
||||
close(s.cancel)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Session) Cancelled() <-chan struct{} {
|
||||
return s.cancel
|
||||
}
|
||||
|
||||
func (s *Session) Lock() { s.mu.Lock() }
|
||||
func (s *Session) Unlock() { s.mu.Unlock() }
|
||||
@@ -0,0 +1,50 @@
|
||||
package tester
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/AlexxIT/go2rtc/pkg/core"
|
||||
"github.com/AlexxIT/go2rtc/pkg/rtsp"
|
||||
)
|
||||
|
||||
// SourceHandler tests stream URL, returns Producer or error
|
||||
type SourceHandler func(rawURL string) (core.Producer, error)
|
||||
|
||||
var handlers = map[string]SourceHandler{}
|
||||
|
||||
func RegisterSource(scheme string, handler SourceHandler) {
|
||||
handlers[scheme] = handler
|
||||
}
|
||||
|
||||
func GetHandler(rawURL string) SourceHandler {
|
||||
if i := strings.IndexByte(rawURL, ':'); i > 0 {
|
||||
return handlers[rawURL[:i]]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
RegisterSource("rtsp", rtspHandler)
|
||||
RegisterSource("rtsps", rtspHandler)
|
||||
RegisterSource("rtspx", rtspHandler)
|
||||
}
|
||||
|
||||
// rtspHandler -- Dial + Describe. Proves: port open, RTSP responds, auth OK, SDP received.
|
||||
func rtspHandler(rawURL string) (core.Producer, error) {
|
||||
rawURL, _, _ = strings.Cut(rawURL, "#")
|
||||
|
||||
conn := rtsp.NewClient(rawURL)
|
||||
conn.Backchannel = false
|
||||
|
||||
if err := conn.Dial(); err != nil {
|
||||
return nil, fmt.Errorf("rtsp: dial: %w", err)
|
||||
}
|
||||
|
||||
if err := conn.Describe(); err != nil {
|
||||
_ = conn.Stop()
|
||||
return nil, fmt.Errorf("rtsp: describe: %w", err)
|
||||
}
|
||||
|
||||
return conn, nil
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
package tester
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"time"
|
||||
|
||||
"github.com/AlexxIT/go2rtc/pkg/core"
|
||||
"github.com/AlexxIT/go2rtc/pkg/magic"
|
||||
)
|
||||
|
||||
const workers = 20
|
||||
|
||||
func RunWorkers(s *Session, urls []string) {
|
||||
ch := make(chan string, len(urls))
|
||||
for _, u := range urls {
|
||||
ch <- u
|
||||
}
|
||||
close(ch)
|
||||
|
||||
done := make(chan struct{})
|
||||
|
||||
n := workers
|
||||
if len(urls) < n {
|
||||
n = len(urls)
|
||||
}
|
||||
|
||||
for i := 0; i < n; i++ {
|
||||
go func() {
|
||||
for rawURL := range ch {
|
||||
select {
|
||||
case <-s.Cancelled():
|
||||
return
|
||||
default:
|
||||
}
|
||||
testURL(s, rawURL)
|
||||
}
|
||||
done <- struct{}{}
|
||||
}()
|
||||
}
|
||||
|
||||
for i := 0; i < n; i++ {
|
||||
<-done
|
||||
}
|
||||
|
||||
s.Done()
|
||||
}
|
||||
|
||||
func testURL(s *Session, rawURL string) {
|
||||
defer s.AddTested()
|
||||
|
||||
handler := GetHandler(rawURL)
|
||||
if handler == nil {
|
||||
return
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
|
||||
prod, err := handler(rawURL)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer func() { _ = prod.Stop() }()
|
||||
|
||||
latency := time.Since(start).Milliseconds()
|
||||
|
||||
var codecs []string
|
||||
for _, media := range prod.GetMedias() {
|
||||
if media.Direction != core.DirectionRecvonly {
|
||||
continue
|
||||
}
|
||||
for _, codec := range media.Codecs {
|
||||
codecs = append(codecs, codec.Name)
|
||||
}
|
||||
}
|
||||
|
||||
r := &Result{
|
||||
Source: rawURL,
|
||||
Codecs: codecs,
|
||||
LatencyMs: latency,
|
||||
}
|
||||
|
||||
if raw, codecName := getScreenshot(prod); raw != nil {
|
||||
var jpeg []byte
|
||||
|
||||
switch codecName {
|
||||
case core.CodecH264, core.CodecH265:
|
||||
jpeg = toJPEG(raw)
|
||||
case core.CodecJPEG:
|
||||
jpeg = raw
|
||||
default:
|
||||
jpeg = raw
|
||||
}
|
||||
|
||||
if jpeg != nil {
|
||||
idx := s.AddScreenshot(jpeg)
|
||||
r.Screenshot = fmt.Sprintf("/api/test/screenshot?id=%s&i=%d", s.ID, idx)
|
||||
}
|
||||
}
|
||||
|
||||
s.AddResult(r)
|
||||
}
|
||||
|
||||
// getScreenshot connects Keyframe consumer to producer, waits for first keyframe with 10s timeout
|
||||
func getScreenshot(prod core.Producer) ([]byte, string) {
|
||||
cons := magic.NewKeyframe()
|
||||
|
||||
for _, prodMedia := range prod.GetMedias() {
|
||||
if prodMedia.Kind != core.KindVideo || prodMedia.Direction != core.DirectionRecvonly {
|
||||
continue
|
||||
}
|
||||
for _, consMedia := range cons.GetMedias() {
|
||||
prodCodec, consCodec := prodMedia.MatchMedia(consMedia)
|
||||
if prodCodec == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
track, err := prod.GetTrack(prodMedia, prodCodec)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if err = cons.AddTrack(consMedia, consCodec, track); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
goto matched
|
||||
}
|
||||
}
|
||||
|
||||
return nil, ""
|
||||
|
||||
matched:
|
||||
go func() {
|
||||
_ = prod.Start()
|
||||
}()
|
||||
|
||||
once := &core.OnceBuffer{}
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
_, _ = cons.WriteTo(once)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(10 * time.Second):
|
||||
_ = prod.Stop()
|
||||
return nil, ""
|
||||
}
|
||||
|
||||
return once.Buffer(), cons.CodecName()
|
||||
}
|
||||
|
||||
func toJPEG(raw []byte) []byte {
|
||||
cmd := exec.Command("ffmpeg",
|
||||
"-hide_banner", "-loglevel", "error",
|
||||
"-i", "-",
|
||||
"-frames:v", "1",
|
||||
"-f", "image2", "-c:v", "mjpeg",
|
||||
"-",
|
||||
)
|
||||
cmd.Stdin = bytes.NewReader(raw)
|
||||
|
||||
out, err := cmd.Output()
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
Reference in New Issue
Block a user