93 lines
2.6 KiB
Go
93 lines
2.6 KiB
Go
|
package source
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
"strings"
|
||
|
|
"sync/atomic"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
)
|
||
|
|
|
||
|
|
func TestPublishLatestDropsStaleFrame(t *testing.T) {
|
||
|
|
frames := make(chan Frame, 1)
|
||
|
|
publishLatest(frames, Frame{Sequence: 1})
|
||
|
|
publishLatest(frames, Frame{Sequence: 2})
|
||
|
|
|
||
|
|
if got := <-frames; got.Sequence != 2 {
|
||
|
|
t.Fatalf("sequence = %d, want 2", got.Sequence)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestRetryDelayIsBounded(t *testing.T) {
|
||
|
|
if got := retryDelay(0); got != time.Second {
|
||
|
|
t.Fatalf("first retry delay = %s", got)
|
||
|
|
}
|
||
|
|
if got := retryDelay(3); got != 8*time.Second {
|
||
|
|
t.Fatalf("fourth retry delay = %s", got)
|
||
|
|
}
|
||
|
|
if got := retryDelay(99); got != 8*time.Second {
|
||
|
|
t.Fatalf("bounded retry delay = %s", got)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestStreamRetriesAndPublishesLatestCompletedFrame(t *testing.T) {
|
||
|
|
ctx, cancel := context.WithCancel(context.Background())
|
||
|
|
defer cancel()
|
||
|
|
var decodeCalls int32
|
||
|
|
decodedLatest := make(chan struct{})
|
||
|
|
stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{
|
||
|
|
Probe: func(context.Context, Config) (Metadata, error) {
|
||
|
|
return Metadata{Width: 2, Height: 1, FPS: 30}, nil
|
||
|
|
},
|
||
|
|
Decode: func(ctx context.Context, _ Config, _ Metadata, publish func([]byte)) error {
|
||
|
|
call := atomic.AddInt32(&decodeCalls, 1)
|
||
|
|
if call == 1 {
|
||
|
|
publish([]byte{1, 1, 1, 1, 1, 1})
|
||
|
|
return errors.New("temporary decoder failure")
|
||
|
|
}
|
||
|
|
publish([]byte{2, 2, 2, 2, 2, 2})
|
||
|
|
close(decodedLatest)
|
||
|
|
cancel()
|
||
|
|
return ctx.Err()
|
||
|
|
},
|
||
|
|
Wait: func(ctx context.Context, _ time.Duration) error { return ctx.Err() },
|
||
|
|
})
|
||
|
|
|
||
|
|
select {
|
||
|
|
case <-decodedLatest:
|
||
|
|
case <-time.After(time.Second):
|
||
|
|
t.Fatal("decoder did not publish the second frame")
|
||
|
|
}
|
||
|
|
select {
|
||
|
|
case got := <-stream.Frames():
|
||
|
|
if got.Sequence != 2 || got.BGR[0] != 2 {
|
||
|
|
t.Fatalf("frame = %#v", got)
|
||
|
|
}
|
||
|
|
case <-time.After(time.Second):
|
||
|
|
t.Fatal("did not receive latest decoded frame")
|
||
|
|
}
|
||
|
|
<-stream.Done()
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestStreamNeverPublishesDecoderErrorDetails(t *testing.T) {
|
||
|
|
ctx, cancel := context.WithCancel(context.Background())
|
||
|
|
defer cancel()
|
||
|
|
stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{
|
||
|
|
Probe: func(context.Context, Config) (Metadata, error) {
|
||
|
|
return Metadata{}, errors.New("ffprobe rtsp://operator:secret@example/101 failed")
|
||
|
|
},
|
||
|
|
Decode: func(context.Context, Config, Metadata, func([]byte)) error { return nil },
|
||
|
|
Wait: func(ctx context.Context, _ time.Duration) error {
|
||
|
|
cancel()
|
||
|
|
return ctx.Err()
|
||
|
|
},
|
||
|
|
})
|
||
|
|
|
||
|
|
for update := range stream.Statuses() {
|
||
|
|
if strings.Contains(update.Message, "secret") || strings.Contains(update.Message, "rtsp://") {
|
||
|
|
t.Fatalf("status leaked source details: %#v", update)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|