diff --git a/event/stream/stream.go b/event/stream/stream.go new file mode 100644 index 0000000..8ed8a44 --- /dev/null +++ b/event/stream/stream.go @@ -0,0 +1,353 @@ +package stream + +import ( + "bytes" + "context" + "encoding/xml" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "sync" + "time" + + "github.com/kerberos-io/onvif" + "github.com/kerberos-io/onvif/event" + "github.com/kerberos-io/onvif/xsd" +) + +// Options configures a Stream. The zero value is usable; defaultOptions +// fills in production-sensible defaults for any unset field. +type Options struct { + // DeviceID identifies the camera in emitted Events. Recommended so a + // single channel can fan in multiple cameras. Empty is allowed. + DeviceID string + // TopicFilter is the raw ONVIF ConcreteSet TopicExpression filter + // passed to CreatePullPointSubscription. The empty string means no + // filter — required for AXIS, accepted by every other vendor we + // support. Callers should normally leave this empty and rely on + // Classify for routing. + TopicFilter string + // PullTimeout is the server-side wait time in each PullMessages call + // (xsd:duration). The camera returns early when messages are + // available; otherwise it returns empty after this timeout. Default: + // 5s. + PullTimeout time.Duration + // MessageLimit caps the number of NotificationMessage entries + // returned per PullMessages call. Default: 10. + MessageLimit int + // InitialTermination is the requested subscription lifetime passed + // to CreatePullPointSubscription. The renew loop (added in a later + // commit) will refresh well before this expires. Default: 60s. + InitialTermination time.Duration + // BufferSize is the buffer size of the Events and Errors channels. + // Larger buffers absorb consumer hiccups at the cost of memory. + // Default: 16. + BufferSize int +} + +func defaultOptions() Options { + return Options{ + PullTimeout: 5 * time.Second, + MessageLimit: 10, + InitialTermination: 60 * time.Second, + BufferSize: 16, + } +} + +func (o Options) withDefaults() Options { + d := defaultOptions() + if o.PullTimeout > 0 { + d.PullTimeout = o.PullTimeout + } + if o.MessageLimit > 0 { + d.MessageLimit = o.MessageLimit + } + if o.InitialTermination > 0 { + d.InitialTermination = o.InitialTermination + } + if o.BufferSize > 0 { + d.BufferSize = o.BufferSize + } + d.DeviceID = o.DeviceID + d.TopicFilter = o.TopicFilter + return d +} + +// caller is the subset of *onvif.Device the Stream depends on. Tests +// substitute a fake; production code uses the device adapter. +type caller interface { + CallMethod(method any) (*http.Response, error) + SendSoap(endpoint, body string) (*http.Response, error) +} + +type deviceCaller struct{ dev *onvif.Device } + +func (d deviceCaller) CallMethod(m any) (*http.Response, error) { + return d.dev.CallMethod(m) +} + +func (d deviceCaller) SendSoap(endpoint, body string) (*http.Response, error) { + return d.dev.SendSoap(endpoint, body) +} + +// Stream owns a single ONVIF pull-point subscription and surfaces the +// decoded notifications on a typed channel. Close stops the background +// goroutine and unsubscribes from the camera. +// +// A Stream is safe for concurrent use by Close from any goroutine while +// readers consume Events / Errors; Close is idempotent. +type Stream struct { + caller caller + opts Options + pullPoint string + + events chan Event + errors chan error + + cancel context.CancelFunc + done chan struct{} + + closeOnce sync.Once + closeErr error + + // now is overridable in tests to make timestamps deterministic. + now func() time.Time +} + +// NewStream creates a Stream against an ONVIF device. It performs the +// CreatePullPointSubscription call synchronously so connectivity and +// authentication problems surface immediately as an error rather than +// landing on the Errors channel later. The background pull loop starts +// before NewStream returns. +// +// The returned Stream stops when ctx is cancelled or when Close is +// called. +func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, error) { + return newStream(ctx, deviceCaller{dev: dev}, opts) +} + +func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) { + opts = opts.withDefaults() + addr, err := createPullPoint(c, opts) + if err != nil { + return nil, fmt.Errorf("create pull point subscription: %w", err) + } + runCtx, cancel := context.WithCancel(ctx) + s := &Stream{ + caller: c, + opts: opts, + pullPoint: addr, + events: make(chan Event, opts.BufferSize), + errors: make(chan error, opts.BufferSize), + cancel: cancel, + done: make(chan struct{}), + now: time.Now, + } + go s.run(runCtx) + return s, nil +} + +// Events returns the channel of decoded notifications. The channel is +// closed when the Stream stops. +func (s *Stream) Events() <-chan Event { return s.events } + +// Errors returns the channel of non-fatal errors encountered while +// pulling. Sends are non-blocking, so consumers that fall behind drop +// older errors. The channel is closed when the Stream stops. +func (s *Stream) Errors() <-chan error { return s.errors } + +// Close stops the background goroutine, waits for it to exit, and +// unsubscribes from the camera. Subsequent calls are no-ops. +func (s *Stream) Close() error { + s.closeOnce.Do(func() { + s.cancel() + <-s.done + // Unsubscribe is best-effort: if the camera is unreachable + // the subscription will expire on its own at + // InitialTermination + Renew interval anyway. + if err := unsubscribePullPoint(s.caller, s.pullPoint); err != nil { + s.closeErr = fmt.Errorf("unsubscribe pull point: %w", err) + } + }) + return s.closeErr +} + +func (s *Stream) run(ctx context.Context) { + defer close(s.done) + defer close(s.events) + defer close(s.errors) + + for { + if ctx.Err() != nil { + return + } + msgs, err := pullMessages(s.caller, s.pullPoint, s.opts) + if err != nil { + s.surfaceError(err) + // Brief backoff before retrying; reconnect-on-error + // lands in a follow-up commit and replaces this with + // proper subscription recreation. + if !sleepCtx(ctx, time.Second) { + return + } + continue + } + observedAt := s.now() + for _, m := range msgs { + ev := Decode(m, s.opts.DeviceID, observedAt) + select { + case <-ctx.Done(): + return + case s.events <- ev: + } + } + } +} + +// surfaceError sends err on the errors channel non-blockingly so a +// stalled consumer cannot block the pull loop. +func (s *Stream) surfaceError(err error) { + select { + case s.errors <- err: + default: + } +} + +// sleepCtx blocks for d or until ctx is cancelled. Returns true if d +// elapsed, false if ctx was cancelled. +func sleepCtx(ctx context.Context, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} + +// --- SOAP helpers (unexported) ---------------------------------------- + +func createPullPoint(c caller, opts Options) (string, error) { + term := xsd.String(durationToXSD(opts.InitialTermination)) + req := event.CreatePullPointSubscription{InitialTerminationTime: &term} + if opts.TopicFilter != "" { + req.Filter = &event.FilterType{ + TopicExpression: &event.TopicExpressionType{ + Dialect: xsd.String("http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"), + TopicKinds: xsd.String(opts.TopicFilter), + }, + } + } + resp, err := c.CallMethod(req) + if err != nil { + return "", err + } + body, err := readClose(resp) + if err != nil { + return "", err + } + var decoded event.CreatePullPointSubscriptionResponse + if err := unmarshalNode(body, "CreatePullPointSubscriptionResponse", &decoded); err != nil { + return "", err + } + addr := string(decoded.SubscriptionReference.Address) + if addr == "" { + return "", errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address") + } + return addr, nil +} + +func pullMessages(c caller, endpoint string, opts Options) ([]event.NotificationMessage, error) { + req := event.PullMessages{ + Timeout: xsd.Duration(durationToXSD(opts.PullTimeout)), + MessageLimit: xsd.Int(opts.MessageLimit), + } + body, err := xml.Marshal(req) + if err != nil { + return nil, fmt.Errorf("marshal PullMessages: %w", err) + } + resp, err := c.SendSoap(endpoint, string(body)) + if err != nil { + return nil, err + } + respBody, err := readClose(resp) + if err != nil { + return nil, err + } + var decoded event.PullMessagesResponse + if err := unmarshalNode(respBody, "PullMessagesResponse", &decoded); err != nil { + return nil, err + } + return decoded.NotificationMessage, nil +} + +func unsubscribePullPoint(c caller, endpoint string) error { + if endpoint == "" { + return nil + } + body, err := xml.Marshal(event.Unsubscribe{}) + if err != nil { + return fmt.Errorf("marshal Unsubscribe: %w", err) + } + resp, err := c.SendSoap(endpoint, string(body)) + if err != nil { + return err + } + _, err = readClose(resp) + return err +} + +func readClose(resp *http.Response) (string, error) { + if resp == nil || resp.Body == nil { + return "", errors.New("nil HTTP response") + } + defer resp.Body.Close() + b, err := io.ReadAll(resp.Body) + if err != nil { + return "", fmt.Errorf("read response body: %w", err) + } + return string(b), nil +} + +// unmarshalNode finds the first XML start element with the given local +// name and decodes it into out. ONVIF SOAP responses come wrapped in an +// envelope with multiple namespace prefixes; this helper sidesteps +// namespace matching by keying on local name only. +func unmarshalNode(body, localName string, out any) error { + dec := xml.NewDecoder(bytes.NewBufferString(body)) + for { + tok, err := dec.Token() + if err != nil { + if errors.Is(err, io.EOF) { + return fmt.Errorf("ONVIF response missing %s element", localName) + } + return fmt.Errorf("scan ONVIF response: %w", err) + } + start, ok := tok.(xml.StartElement) + if !ok { + continue + } + if start.Name.Local != localName { + continue + } + if err := dec.DecodeElement(out, &start); err != nil { + return fmt.Errorf("decode %s: %w", localName, err) + } + return nil + } +} + +// durationToXSD formats a Go time.Duration as an xsd:duration string in +// PTnS form. Second precision is sufficient — ONVIF cameras do not +// honour sub-second pull timeouts and intermediate routers may round in +// any case. +func durationToXSD(d time.Duration) string { + secs := int(d.Round(time.Second).Seconds()) + if secs <= 0 { + secs = 1 + } + return "PT" + strconv.Itoa(secs) + "S" +} diff --git a/event/stream/stream_test.go b/event/stream/stream_test.go new file mode 100644 index 0000000..698d07e --- /dev/null +++ b/event/stream/stream_test.go @@ -0,0 +1,342 @@ +package stream + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- fakeCaller -------------------------------------------------------- + +// fakeCaller is a test double for the caller interface. Each method +// returns the next queued response; when the queue is exhausted it falls +// back to a default response so the indefinite pull loop does not +// require tests to enumerate every call. +type fakeCaller struct { + mu sync.Mutex + callMethodResps []fakeResp + sendSoapResps []fakeResp + defaultSendSoap fakeResp + defaultCall fakeResp + callMethodCalls []any + sendSoapCalls [][2]string +} + +type fakeResp struct { + body string + err error +} + +func newFakeCaller() *fakeCaller { + return &fakeCaller{ + // Default: indefinite empty pulls, indefinite OK unsubscribes. + defaultSendSoap: fakeResp{body: pullMessagesResp()}, + defaultCall: fakeResp{err: errors.New("fakeCaller: no default CallMethod response")}, + } +} + +func (f *fakeCaller) queueCallMethod(body string, err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.callMethodResps = append(f.callMethodResps, fakeResp{body: body, err: err}) +} + +func (f *fakeCaller) queueSendSoap(body string, err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.sendSoapResps = append(f.sendSoapResps, fakeResp{body: body, err: err}) +} + +func (f *fakeCaller) CallMethod(m any) (*http.Response, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.callMethodCalls = append(f.callMethodCalls, m) + r := f.defaultCall + if len(f.callMethodResps) > 0 { + r = f.callMethodResps[0] + f.callMethodResps = f.callMethodResps[1:] + } + if r.err != nil { + return nil, r.err + } + return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil +} + +func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.sendSoapCalls = append(f.sendSoapCalls, [2]string{endpoint, body}) + r := f.defaultSendSoap + if len(f.sendSoapResps) > 0 { + r = f.sendSoapResps[0] + f.sendSoapResps = f.sendSoapResps[1:] + } + if r.err != nil { + return nil, r.err + } + return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil +} + +func (f *fakeCaller) sendSoapCallCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.sendSoapCalls) +} + +// --- fixture SOAP envelopes ------------------------------------------- + +// createPullPointResp is the minimal SOAP envelope the lib's existing +// xml.Decoder + getXMLNode path can extract a pull-point address from. +const createPullPointResp = ` + + + + + http://camera.local/onvif/Events/PullSub_1 + + 2026-05-21T10:30:00Z + 2026-05-21T10:31:00Z + + +` + +func pullMessagesResp(messages ...string) string { + return ` + + + + 2026-05-21T10:30:05Z + 2026-05-21T10:31:05Z + ` + strings.Join(messages, "\n") + ` + + +` +} + +func motionMsg(value string) string { + return ` + tns1:RuleEngine/CellMotionDetector/Motion + + + + + + + + + + +` +} + +const unsubscribeResp = ` + + + + +` + +// --- helpers ----------------------------------------------------------- + +// receive waits up to d for an event on ch, failing the test if none +// arrives. +func receive(t *testing.T, ch <-chan Event, d time.Duration) Event { + t.Helper() + select { + case ev, ok := <-ch: + if !ok { + t.Fatalf("event channel closed before receiving") + } + return ev + case <-time.After(d): + t.Fatalf("timed out waiting for event after %s", d) + } + return Event{} // unreachable +} + +// --- tests ------------------------------------------------------------- + +func TestNewStream_CreatesPullPointAtConstruction(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + // Queue an empty pull so the run loop can spin without exploding. + fc.queueSendSoap(pullMessagesResp(), nil) + fc.queueSendSoap(unsubscribeResp, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"}) + require.NoError(t, err) + require.NotNil(t, s) + require.NoError(t, s.Close()) + + // CreatePullPointSubscription was called exactly once. + fc.mu.Lock() + defer fc.mu.Unlock() + require.Len(t, fc.callMethodCalls, 1, "expected one CallMethod call (CreatePullPointSubscription)") +} + +func TestStream_DeliversDecodedEvents(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil) + // Provide subsequent empty pulls so the loop doesn't starve before Close. + fc.queueSendSoap(pullMessagesResp(), nil) + fc.queueSendSoap(pullMessagesResp(), nil) + fc.queueSendSoap(unsubscribeResp, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"}) + require.NoError(t, err) + defer s.Close() + + ev := receive(t, s.Events(), 2*time.Second) + assert.Equal(t, KindMotion, ev.Kind) + assert.Equal(t, StateActive, ev.State) + assert.Equal(t, "cam-1", ev.DeviceID) + assert.Equal(t, "tns1:RuleEngine/CellMotionDetector/Motion", ev.Topic) + assert.Equal(t, "VSC0", ev.Source["VideoSourceConfigurationToken"]) + assert.Equal(t, "true", ev.Data["IsMotion"]) +} + +func TestStream_PullsAgainstSubscriptionAddress(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + fc.queueSendSoap(pullMessagesResp(), nil) + fc.queueSendSoap(pullMessagesResp(), nil) + fc.queueSendSoap(unsubscribeResp, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"}) + require.NoError(t, err) + + // Wait until at least one pull happened, then close. + for i := 0; i < 50 && fc.sendSoapCallCount() == 0; i++ { + time.Sleep(10 * time.Millisecond) + } + require.NoError(t, s.Close()) + + fc.mu.Lock() + defer fc.mu.Unlock() + require.NotEmpty(t, fc.sendSoapCalls, "expected at least one PullMessages SendSoap call") + endpoint := fc.sendSoapCalls[0][0] + assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", endpoint, + "PullMessages must target the SubscriptionReference Address returned by CreatePullPoint") + // Last call (Close) should target the same endpoint with an Unsubscribe body. + last := fc.sendSoapCalls[len(fc.sendSoapCalls)-1] + assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", last[0]) + assert.Contains(t, last[1], "Unsubscribe") +} + +func TestNewStream_ReturnsErrorWhenCreatePullPointFails(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod("", errors.New("network down")) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"}) + assert.Error(t, err) + assert.Nil(t, s) +} + +func TestStream_ClosedContextStopsRunLoop(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + // Many empty pulls so the loop is hot when we cancel. + for i := 0; i < 20; i++ { + fc.queueSendSoap(pullMessagesResp(), nil) + } + fc.queueSendSoap(unsubscribeResp, nil) + + ctx, cancel := context.WithCancel(context.Background()) + s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"}) + require.NoError(t, err) + + // Wait for at least one pull. + for i := 0; i < 50 && fc.sendSoapCallCount() == 0; i++ { + time.Sleep(10 * time.Millisecond) + } + cancel() + + // Close should still complete cleanly; the goroutine must drain. + require.NoError(t, s.Close()) + + // Events channel must close so consumers can range-loop safely. + select { + case _, ok := <-s.Events(): + assert.False(t, ok, "Events channel should be closed after Close()") + case <-time.After(time.Second): + t.Fatal("Events channel was not closed within 1s") + } +} + +func TestStream_PullErrorSurfacedOnErrorsChannel(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + fc.queueSendSoap("", errors.New("transient pull failure")) + // Then a clean pull so the loop keeps running. + fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil) + fc.queueSendSoap(pullMessagesResp(), nil) + fc.queueSendSoap(unsubscribeResp, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"}) + require.NoError(t, err) + defer s.Close() + + select { + case e := <-s.Errors(): + assert.Contains(t, e.Error(), "transient pull failure") + case <-time.After(2 * time.Second): + t.Fatal("expected an error on the Errors channel") + } + // After the transient failure the loop continued and decoded. + ev := receive(t, s.Events(), 2*time.Second) + assert.Equal(t, KindMotion, ev.Kind) +} + +func TestStream_OptionsApplyDefaults(t *testing.T) { + o := defaultOptions() + assert.Equal(t, 5*time.Second, o.PullTimeout) + assert.Equal(t, 10, o.MessageLimit) + assert.Equal(t, 60*time.Second, o.InitialTermination) + assert.Equal(t, 16, o.BufferSize) +} + +func TestStream_DoesNotPanicOnPullExitingDuringClose(t *testing.T) { + // Regression guard: Close should not race with the run goroutine + // in a way that double-closes the events/errors channels. + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + for i := 0; i < 5; i++ { + fc.queueSendSoap(pullMessagesResp(), nil) + } + fc.queueSendSoap(unsubscribeResp, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"}) + require.NoError(t, err) + assert.NotPanics(t, func() { + require.NoError(t, s.Close()) + // Double-close should be a no-op, not a panic. + _ = s.Close() + }) +}