diff --git a/event/stream/coverage_test.go b/event/stream/coverage_test.go
deleted file mode 100644
index 29b7ea2..0000000
--- a/event/stream/coverage_test.go
+++ /dev/null
@@ -1,295 +0,0 @@
-package stream
-
-import (
- "context"
- "errors"
- "strings"
- "testing"
- "time"
-
- "github.com/kerberos-io/onvif/event"
- "github.com/kerberos-io/onvif/xsd"
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
-)
-
-// --- Close surfaces unsubscribe error --------------------------------
-
-func TestClose_ReturnsUnsubscribeError(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- // Default empty pulls keep the loop running. Override default
- // SendSoap to fail so Close's Unsubscribe also fails.
- fc.mu.Lock()
- fc.defaultSendSoap = fakeResp{err: errors.New("simulated transport failure")}
- fc.mu.Unlock()
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
- require.NoError(t, err)
-
- err = s.Close()
- require.Error(t, err)
- assert.Contains(t, err.Error(), "unsubscribe pull point")
- assert.Contains(t, err.Error(), "simulated transport failure")
-}
-
-// --- NewStream against already-cancelled context ----------------------
-
-func TestNewStream_CtxAlreadyCancelled(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
-
- ctx, cancel := context.WithCancel(context.Background())
- cancel() // cancel before NewStream
-
- s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
- // Create-pull-point doesn't currently consult ctx (it uses caller
- // directly), so construction succeeds and the run goroutine exits
- // immediately. Close must still work cleanly.
- require.NoError(t, err)
- require.NotNil(t, s)
-
- // Events channel must close promptly because the goroutine exits.
- select {
- case _, ok := <-s.Events():
- assert.False(t, ok, "events channel should be closed when ctx is pre-cancelled")
- case <-time.After(time.Second):
- t.Fatal("events channel was not closed within 1s")
- }
- _ = s.Close()
-}
-
-// --- DisableReconnect honours the opt-out ----------------------------
-
-func TestStream_DisableReconnectKeepsRetryingOriginalEndpoint(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- // All pulls fail; default SendSoap stays as empty-pull (success)
- // only if the fake's queue exhausts — we override default to a
- // failure so EVERY pull errors.
- fc.mu.Lock()
- fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
- fc.mu.Unlock()
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{
- PullTimeout: 10 * time.Millisecond,
- RetryBackoff: 10 * time.Millisecond,
- InitialTermination: 30 * time.Second,
- DisableReconnect: true,
- })
- require.NoError(t, err)
- defer s.Close()
-
- // Let the loop spin for a bit, then assert no second CallMethod
- // (recreate would invoke CallMethod, which we are watching).
- time.Sleep(200 * time.Millisecond)
- fc.mu.Lock()
- calls := len(fc.callMethodCalls)
- fc.mu.Unlock()
- assert.Equal(t, 1, calls, "DisableReconnect must prevent recreate; got %d CallMethod calls", calls)
-}
-
-// --- Recreate resets failures+backoff on success ---------------------
-
-func TestStream_RecreateResetsFailuresAndBackoffOnSuccess(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- fc.queueCallMethod(createPullPointRespAlt, nil)
- // Pull fails once -> triggers recreate -> recreate succeeds ->
- // next pull succeeds. After that we should NOT see another
- // recreate (failures was reset). Provide enough successful empty
- // pulls.
- fc.queueSendSoap("", errors.New("first failure"))
- // Subsequent pulls succeed via default empty pull.
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{
- PullTimeout: 10 * time.Millisecond,
- ReconnectAfterFailures: 1,
- RetryBackoff: 10 * time.Millisecond,
- InitialTermination: 30 * time.Second,
- })
- require.NoError(t, err)
- defer s.Close()
-
- time.Sleep(200 * time.Millisecond)
- fc.mu.Lock()
- calls := len(fc.callMethodCalls)
- fc.mu.Unlock()
- assert.Equal(t, 2, calls,
- "after one failure + successful recreate, no further recreates expected; got %d", calls)
-}
-
-// --- pullPointMu under race ------------------------------------------
-
-func TestStream_PullPointMutationVisibleToRenewLoopUnderRace(t *testing.T) {
- // Drives the pullPoint write-by-pullLoop / read-by-renewLoop race
- // so -race actually exercises the mutex critical sections. With
- // short termination and quick recreate, renew is firing alongside
- // the recreate write.
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- // Queue a stream of alt-response recreates so each retry installs
- // a new pullPoint.
- for i := 0; i < 50; i++ {
- fc.queueCallMethod(createPullPointRespAlt, nil)
- }
- // Default empty pulls.
- // Force pull errors so reconnect path fires repeatedly: override
- // default and queue mostly-failing pulls.
- fc.mu.Lock()
- fc.defaultSendSoap = fakeResp{err: errors.New("recurring pull fail")}
- fc.mu.Unlock()
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{
- PullTimeout: 5 * time.Millisecond,
- ReconnectAfterFailures: 1,
- RetryBackoff: 1 * time.Millisecond,
- InitialTermination: 20 * time.Millisecond,
- RenewMargin: 2 * time.Millisecond,
- })
- require.NoError(t, err)
- defer s.Close()
-
- // Spin for ~300ms; the race detector will fire if either
- // pullPointMu critical section is broken. We don't assert on
- // content here — the value is the -race signal.
- time.Sleep(300 * time.Millisecond)
-}
-
-// --- fakeCaller self-test --------------------------------------------
-
-func TestFakeCaller_QueueThenDefaultFallback(t *testing.T) {
- fc := newFakeCaller()
- fc.queueSendSoap("first", nil)
- fc.queueSendSoap("second", nil)
- // Default already set to an empty pull response.
-
- r1, err := fc.SendSoap("ep", "body")
- require.NoError(t, err)
- b1 := make([]byte, 10)
- n, _ := r1.Body.Read(b1)
- assert.Equal(t, "first", string(b1[:n]))
-
- r2, _ := fc.SendSoap("ep", "body")
- b2 := make([]byte, 10)
- n, _ = r2.Body.Read(b2)
- assert.Equal(t, "second", string(b2[:n]))
-
- // Queue is exhausted; default kicks in.
- r3, err := fc.SendSoap("ep", "body")
- require.NoError(t, err)
- require.NotNil(t, r3)
- b3 := make([]byte, 2048)
- n, _ = r3.Body.Read(b3)
- assert.Contains(t, string(b3[:n]), "PullMessagesResponse",
- "default SendSoap should be an empty PullMessagesResponse envelope")
-}
-
-// --- Decoder coverage gaps -------------------------------------------
-
-func TestDecode_PropertyOperationIsCaseSensitive(t *testing.T) {
- // Per WS-Notification §3.3 PropertyOperation values are
- // 'Initialized' / 'Changed' / 'Deleted'. Lowercased forms in the
- // wild are malformed and should fall through to PropertyUnknown.
- in := msg("tns1:VideoSource/MotionAlarm", "changed", "", nil, nil)
- ev := decode(in, "dev", time.Now())
- assert.Equal(t, PropertyUnknown, ev.Operation)
-}
-
-func TestDecode_StateValueTrimsWhitespace(t *testing.T) {
- tests := []struct {
- name string
- value string
- want State
- }{
- {"leading_trailing", " true ", StateActive},
- {"tab_newline", "\ttrue\n", StateActive},
- {"only_spaces", " ", StateUnknown},
- }
- for _, tc := range tests {
- t.Run(tc.name, func(t *testing.T) {
- in := msg("tns1:VideoSource/MotionAlarm", "Changed", "",
- nil, map[string]string{"State": tc.value})
- ev := decode(in, "dev", time.Now())
- assert.Equal(t, tc.want, ev.State)
- })
- }
-}
-
-func TestDecode_SimpleItemEmptyValueIsUnknownState(t *testing.T) {
- in := msg("tns1:VideoSource/MotionAlarm", "Changed", "",
- nil, map[string]string{"State": ""})
- ev := decode(in, "dev", time.Now())
- assert.Equal(t, StateUnknown, ev.State)
- // Empty value still preserved in the Data map.
- v, ok := ev.Data["State"]
- assert.True(t, ok)
- assert.Equal(t, "", v)
-}
-
-func TestDecode_DeviceTimeAdditionalLayouts(t *testing.T) {
- tests := []struct {
- name string
- in string
- want time.Time
- }{
- {"compact_offset", "2026-05-21T12:30:00+0200", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
- {"compact_offset_subsec", "2026-05-21T12:30:00.500+0200", time.Date(2026, 5, 21, 10, 30, 0, 500_000_000, time.UTC)},
- {"naked_no_tz", "2026-05-21T10:30:00", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
- }
- for _, tc := range tests {
- t.Run(tc.name, func(t *testing.T) {
- in := msg("tns1:VideoSource/MotionAlarm", "Changed", tc.in, nil, nil)
- ev := decode(in, "dev", time.Now())
- assert.True(t, ev.DeviceTime.Equal(tc.want),
- "input=%q got=%v want=%v", tc.in, ev.DeviceTime, tc.want)
- })
- }
-}
-
-// --- extractState deterministic order with explicit slice ------------
-
-func TestExtractState_FirstBooleanLikeWins(t *testing.T) {
- // Verifies the documented behaviour: when multiple Data items have
- // boolean-like values, the first by slice order wins.
- in := msg("tns1:VideoSource/MotionAlarm", "Changed", "", nil, nil)
- in.Message.Message.Data.SimpleItem = simpleItemsFromPairs([]pair{
- {"ObjectId", "42"},
- {"State", "true"},
- {"Trailer", "false"},
- })
- ev := decode(in, "dev", time.Now())
- assert.Equal(t, StateActive, ev.State,
- "first boolean-like value (State=true) must win, not Trailer=false")
-}
-
-type pair struct{ k, v string }
-
-func simpleItemsFromPairs(pairs []pair) []event.SimpleItem {
- out := make([]event.SimpleItem, len(pairs))
- for i, p := range pairs {
- out[i] = event.SimpleItem{
- Name: xsd.AnyType(p.k),
- Value: xsd.AnyType(p.v),
- }
- }
- return out
-}
-
-// --- ensure the new layouts don't accept unrelated junk --------------
-
-func TestDecode_DeviceTimeStillRejectsNonsense(t *testing.T) {
- for _, s := range []string{"hello", "2026-13-45T99:99:99", strings.Repeat("9", 50)} {
- in := msg("tns1:VideoSource/MotionAlarm", "Changed", s, nil, nil)
- ev := decode(in, "dev", time.Now())
- assert.True(t, ev.DeviceTime.IsZero(), "input=%q should yield zero, got %v", s, ev.DeviceTime)
- }
-}
diff --git a/event/stream/decode_test.go b/event/stream/decode_test.go
index 2d52800..f30eec7 100644
--- a/event/stream/decode_test.go
+++ b/event/stream/decode_test.go
@@ -1,6 +1,7 @@
package stream
import (
+ "strings"
"testing"
"time"
@@ -247,3 +248,103 @@ func TestDecode_StateValueIsCaseInsensitive(t *testing.T) {
})
}
}
+
+// --- Edge cases for state extraction and time parsing ----------------
+
+func TestDecode_PropertyOperationIsCaseSensitive(t *testing.T) {
+ // Per WS-Notification §3.3 PropertyOperation values are
+ // 'Initialized' / 'Changed' / 'Deleted'. Lowercased forms are
+ // malformed and should fall through to PropertyUnknown.
+ in := msg("tns1:VideoSource/MotionAlarm", "changed", "", nil, nil)
+ ev := decode(in, "dev", time.Now())
+ assert.Equal(t, PropertyUnknown, ev.Operation)
+}
+
+func TestDecode_StateValueTrimsWhitespace(t *testing.T) {
+ tests := []struct {
+ name string
+ value string
+ want State
+ }{
+ {"leading_trailing", " true ", StateActive},
+ {"tab_newline", "\ttrue\n", StateActive},
+ {"only_spaces", " ", StateUnknown},
+ }
+ for _, tc := range tests {
+ t.Run(tc.name, func(t *testing.T) {
+ in := msg("tns1:VideoSource/MotionAlarm", "Changed", "",
+ nil, map[string]string{"State": tc.value})
+ ev := decode(in, "dev", time.Now())
+ assert.Equal(t, tc.want, ev.State)
+ })
+ }
+}
+
+func TestDecode_SimpleItemEmptyValueIsUnknownState(t *testing.T) {
+ in := msg("tns1:VideoSource/MotionAlarm", "Changed", "",
+ nil, map[string]string{"State": ""})
+ ev := decode(in, "dev", time.Now())
+ assert.Equal(t, StateUnknown, ev.State)
+ v, ok := ev.Data["State"]
+ assert.True(t, ok)
+ assert.Equal(t, "", v)
+}
+
+func TestDecode_DeviceTimeAdditionalLayouts(t *testing.T) {
+ tests := []struct {
+ name string
+ in string
+ want time.Time
+ }{
+ {"compact_offset", "2026-05-21T12:30:00+0200", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
+ {"compact_offset_subsec", "2026-05-21T12:30:00.500+0200", time.Date(2026, 5, 21, 10, 30, 0, 500_000_000, time.UTC)},
+ {"naked_no_tz", "2026-05-21T10:30:00", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
+ }
+ for _, tc := range tests {
+ t.Run(tc.name, func(t *testing.T) {
+ in := msg("tns1:VideoSource/MotionAlarm", "Changed", tc.in, nil, nil)
+ ev := decode(in, "dev", time.Now())
+ assert.True(t, ev.DeviceTime.Equal(tc.want),
+ "input=%q got=%v want=%v", tc.in, ev.DeviceTime, tc.want)
+ })
+ }
+}
+
+func TestDecode_DeviceTimeStillRejectsNonsense(t *testing.T) {
+ for _, s := range []string{"hello", "2026-13-45T99:99:99", strings.Repeat("9", 50)} {
+ in := msg("tns1:VideoSource/MotionAlarm", "Changed", s, nil, nil)
+ ev := decode(in, "dev", time.Now())
+ assert.True(t, ev.DeviceTime.IsZero(), "input=%q should yield zero, got %v", s, ev.DeviceTime)
+ }
+}
+
+// --- First-boolean-wins with explicit slice order --------------------
+
+type pair struct{ k, v string }
+
+func simpleItemsFromPairs(pairs []pair) []event.SimpleItem {
+ out := make([]event.SimpleItem, len(pairs))
+ for i, p := range pairs {
+ out[i] = event.SimpleItem{
+ Name: xsd.AnyType(p.k),
+ Value: xsd.AnyType(p.v),
+ }
+ }
+ return out
+}
+
+func TestExtractState_FirstBooleanLikeWins(t *testing.T) {
+ // Documented behaviour: when multiple Data items have boolean-like
+ // values, the first by slice order wins. Use explicit slice
+ // construction so the assertion does not depend on map iteration
+ // order.
+ in := msg("tns1:VideoSource/MotionAlarm", "Changed", "", nil, nil)
+ in.Message.Message.Data.SimpleItem = simpleItemsFromPairs([]pair{
+ {"ObjectId", "42"},
+ {"State", "true"},
+ {"Trailer", "false"},
+ })
+ ev := decode(in, "dev", time.Now())
+ assert.Equal(t, StateActive, ev.State,
+ "first boolean-like value (State=true) must win, not Trailer=false")
+}
diff --git a/event/stream/errors.go b/event/stream/errors.go
deleted file mode 100644
index bd80674..0000000
--- a/event/stream/errors.go
+++ /dev/null
@@ -1,42 +0,0 @@
-package stream
-
-import "fmt"
-
-// Op identifies which Stream operation failed. Used by ErrPullFailed,
-// ErrRenewFailed and ErrRecreateFailed so consumers can branch with
-// errors.As without parsing the wrapped message.
-type Op string
-
-const (
- OpPull Op = "pull"
- OpRenew Op = "renew"
- OpRecreate Op = "recreate"
-)
-
-// ErrPullFailed wraps a transient PullMessages failure. The pull loop
-// surfaces it on the Errors channel and continues. Consumers can match
-// with errors.As(err, &stream.ErrPullFailed{}) — see
-// TestErrors_TypedAssertion.
-type ErrPullFailed struct{ Err error }
-
-func (e ErrPullFailed) Error() string { return fmt.Sprintf("pull messages: %v", e.Err) }
-func (e ErrPullFailed) Unwrap() error { return e.Err }
-func (ErrPullFailed) Op() Op { return OpPull }
-
-// ErrRenewFailed wraps a Renew SOAP failure. Renew errors are usually
-// recovered implicitly: the subscription dies, pull starts failing, and
-// the reconnect logic recreates it.
-type ErrRenewFailed struct{ Err error }
-
-func (e ErrRenewFailed) Error() string { return fmt.Sprintf("renew pull point: %v", e.Err) }
-func (e ErrRenewFailed) Unwrap() error { return e.Err }
-func (ErrRenewFailed) Op() Op { return OpRenew }
-
-// ErrRecreateFailed wraps a failed CreatePullPointSubscription during
-// the reconnect path. The loop continues with exponential backoff;
-// consumers seeing this repeatedly should consider the camera offline.
-type ErrRecreateFailed struct{ Err error }
-
-func (e ErrRecreateFailed) Error() string { return fmt.Sprintf("recreate pull point: %v", e.Err) }
-func (e ErrRecreateFailed) Unwrap() error { return e.Err }
-func (ErrRecreateFailed) Op() Op { return OpRecreate }
diff --git a/event/stream/errors_test.go b/event/stream/errors_test.go
deleted file mode 100644
index d68e347..0000000
--- a/event/stream/errors_test.go
+++ /dev/null
@@ -1,133 +0,0 @@
-package stream
-
-import (
- "context"
- "errors"
- "testing"
- "time"
-
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
-)
-
-func TestTypedErrors_UnwrapAndOp(t *testing.T) {
- inner := errors.New("boom")
- tests := []struct {
- name string
- err error
- op Op
- }{
- {"pull", ErrPullFailed{Err: inner}, OpPull},
- {"renew", ErrRenewFailed{Err: inner}, OpRenew},
- {"recreate", ErrRecreateFailed{Err: inner}, OpRecreate},
- }
- for _, tc := range tests {
- t.Run(tc.name, func(t *testing.T) {
- assert.True(t, errors.Is(tc.err, inner), "errors.Is should unwrap to inner")
- assert.Contains(t, tc.err.Error(), "boom")
-
- // Each typed error exposes Op() for branch-without-string-parse.
- if e, ok := tc.err.(interface{ Op() Op }); ok {
- assert.Equal(t, tc.op, e.Op())
- } else {
- t.Fatalf("%T does not expose Op()", tc.err)
- }
- })
- }
-}
-
-func TestStream_PullErrorIsTypedErrPullFailed(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- fc.queueSendSoap("", errors.New("transient"))
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{
- PullTimeout: 50 * time.Millisecond,
- RetryBackoff: 10 * time.Millisecond,
- InitialTermination: 30 * time.Second,
- })
- require.NoError(t, err)
- defer s.Close()
-
- select {
- case e := <-s.Errors():
- var pullErr ErrPullFailed
- require.True(t, errors.As(e, &pullErr), "expected ErrPullFailed, got %T: %v", e, e)
- assert.Contains(t, pullErr.Err.Error(), "transient")
- case <-time.After(time.Second):
- t.Fatal("expected ErrPullFailed on Errors channel")
- }
-}
-
-func TestStream_RecreateErrorIsTypedErrRecreateFailed(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- fc.mu.Lock()
- fc.defaultCall = fakeResp{err: errors.New("recreate fail")}
- fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
- fc.mu.Unlock()
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{
- PullTimeout: 10 * time.Millisecond,
- ReconnectAfterFailures: 1,
- RetryBackoff: 10 * time.Millisecond,
- InitialTermination: 30 * time.Second,
- })
- require.NoError(t, err)
- defer s.Close()
-
- deadline := time.Now().Add(time.Second)
- var sawRecreate bool
- for time.Now().Before(deadline) && !sawRecreate {
- select {
- case e := <-s.Errors():
- var rec ErrRecreateFailed
- if errors.As(e, &rec) {
- sawRecreate = true
- assert.Contains(t, rec.Err.Error(), "recreate fail")
- }
- case <-time.After(50 * time.Millisecond):
- }
- }
- assert.True(t, sawRecreate, "expected at least one ErrRecreateFailed on Errors")
-}
-
-func TestStream_AfterReconnectFlagSetOnFirstPostRecreateBatch(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- // Second create is the recreate.
- fc.queueCallMethod(createPullPointRespAlt, nil)
-
- // First pull fails -> triggers recreate with ReconnectAfterFailures=1.
- fc.queueSendSoap("", errors.New("transient"))
- // First pull after recreate: a Changed motion event. The flag
- // should be true, and should clear (because we received a
- // non-Initialized event).
- fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
- // Second pull after recreate: another motion event. Flag should
- // now be false.
- fc.queueSendSoap(pullMessagesResp(motionMsg("false")), nil)
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{
- PullTimeout: 50 * time.Millisecond,
- ReconnectAfterFailures: 1,
- RetryBackoff: 10 * time.Millisecond,
- InitialTermination: 30 * time.Second,
- })
- require.NoError(t, err)
- defer s.Close()
-
- ev1 := receive(t, s.Events(), 2*time.Second)
- assert.True(t, ev1.AfterReconnect, "first event after recreate must carry AfterReconnect=true")
- assert.Equal(t, StateActive, ev1.State)
-
- ev2 := receive(t, s.Events(), 2*time.Second)
- assert.False(t, ev2.AfterReconnect, "subsequent events should not carry AfterReconnect")
- assert.Equal(t, StateInactive, ev2.State)
-}
diff --git a/event/stream/jitter_test.go b/event/stream/jitter_test.go
deleted file mode 100644
index 3e5d839..0000000
--- a/event/stream/jitter_test.go
+++ /dev/null
@@ -1,44 +0,0 @@
-package stream
-
-import (
- "testing"
- "time"
-
- "github.com/stretchr/testify/assert"
-)
-
-func TestJitter_StaysWithinFraction(t *testing.T) {
- const base = time.Second
- low := time.Duration(float64(base) * (1 - jitterFraction))
- high := time.Duration(float64(base) * (1 + jitterFraction))
- for i := 0; i < 200; i++ {
- got := jitter(base)
- assert.GreaterOrEqual(t, got, low, "iteration %d", i)
- assert.LessOrEqual(t, got, high, "iteration %d", i)
- }
-}
-
-func TestJitter_ZeroAndNegativeReturnPositive(t *testing.T) {
- assert.Greater(t, jitter(0), time.Duration(0))
- assert.Greater(t, jitter(-time.Second), time.Duration(0))
-}
-
-func TestJitter_VariesAcrossCalls(t *testing.T) {
- // Sanity check that we're not returning a constant. Vanishingly
- // unlikely to flake (probability ~ (1/uint64-space)^9).
- first := jitter(time.Second)
- allEqual := true
- for i := 0; i < 10; i++ {
- if jitter(time.Second) != first {
- allEqual = false
- break
- }
- }
- assert.False(t, allEqual, "jitter is producing a constant; rand seed not working")
-}
-
-func TestMaxRecreateBackoff_Is5Minutes(t *testing.T) {
- // Document the policy choice in a test so a future maintainer
- // changing this notices.
- assert.Equal(t, 5*time.Minute, maxRecreateBackoff)
-}
diff --git a/event/stream/reconnect.go b/event/stream/reconnect.go
new file mode 100644
index 0000000..52cb047
--- /dev/null
+++ b/event/stream/reconnect.go
@@ -0,0 +1,127 @@
+package stream
+
+import (
+ "context"
+ "math/rand"
+ "time"
+)
+
+// maxRecreateBackoff caps exponential backoff between recreate attempts.
+// Sized for fleet deployments: a 1000-camera setup recovering from a
+// switch reboot would otherwise hammer the network with one recreate
+// attempt per camera per 30s; 5 minutes gives the network time to
+// settle while still recovering promptly when a single camera comes
+// back.
+const maxRecreateBackoff = 5 * time.Minute
+
+// jitterFraction is the symmetric jitter applied to recreate backoff:
+// the actual sleep is sampled from [backoff*(1-jitter), backoff*(1+jitter)].
+// Prevents thundering-herd reconnects when many cameras drop together
+// (switch reboot, NAT timeout).
+const jitterFraction = 0.25
+
+// pullLoop is the main pull goroutine of a Stream. It calls
+// PullMessages in a tight loop, decodes results into Events and feeds
+// the Events channel.
+//
+// After ReconnectAfterFailures consecutive pull errors it asks
+// attemptRecreate to recreate the pull-point subscription, marking the
+// next batch's events with AfterReconnect so consumers can suppress
+// duplicate handling of the ONVIF Initialized-replay that follows a
+// new subscription.
+//
+// Exits when ctx is cancelled.
+func (s *Stream) pullLoop(ctx context.Context) {
+ var failures int
+ recreateBackoff := s.opts.RetryBackoff
+ var afterReconnect bool
+
+ for {
+ if ctx.Err() != nil {
+ return
+ }
+ msgs, err := pullMessages(s.caller, s.getPullPoint(), s.opts)
+ if err != nil {
+ s.surfaceError(ErrPullFailed{Err: err})
+ failures++
+ if !s.opts.DisableReconnect && failures >= s.opts.ReconnectAfterFailures {
+ justRecreated, cont := s.attemptRecreate(ctx, &failures, &recreateBackoff)
+ if !cont {
+ return
+ }
+ if justRecreated {
+ afterReconnect = true
+ }
+ continue
+ }
+ if !sleepCtx(ctx, s.opts.RetryBackoff) {
+ return
+ }
+ continue
+ }
+ // Successful pull resets failure tracking.
+ failures = 0
+ recreateBackoff = s.opts.RetryBackoff
+ observedAt := s.now()
+ for _, m := range msgs {
+ ev := decode(m, s.opts.DeviceID, observedAt)
+ if afterReconnect {
+ ev.AfterReconnect = true
+ // ONVIF replays current state with
+ // PropertyInitialized on a new subscription.
+ // Clear the flag as soon as we see anything
+ // other than Initialized — at that point we
+ // have transitioned to live events.
+ if ev.Operation != PropertyInitialized {
+ afterReconnect = false
+ }
+ }
+ select {
+ case <-ctx.Done():
+ return
+ case s.events <- ev:
+ }
+ }
+ }
+}
+
+// attemptRecreate calls CreatePullPointSubscription and on success
+// installs the new endpoint atomically. The first return is true when
+// recreate succeeded just now (caller flags the next batch with
+// AfterReconnect). The second return is false only if ctx was cancelled
+// during backoff (caller should exit the run loop).
+func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *time.Duration) (justRecreated, cont bool) {
+ addr, err := createPullPoint(s.caller, s.opts)
+ if err != nil {
+ s.surfaceError(ErrRecreateFailed{Err: err})
+ if !sleepCtx(ctx, jitter(*backoff)) {
+ return false, false
+ }
+ *backoff *= 2
+ if *backoff > maxRecreateBackoff {
+ *backoff = maxRecreateBackoff
+ }
+ return false, true
+ }
+ s.setPullPoint(addr)
+ *failures = 0
+ *backoff = s.opts.RetryBackoff
+ return true, true
+}
+
+// jitter returns d perturbed by ±jitterFraction. Used to spread
+// recreate attempts across a fleet so a synchronised drop (switch
+// reboot, DHCP storm) does not cause a synchronised reconnect surge.
+// Returns at least 1ns to keep sleepCtx happy.
+func jitter(d time.Duration) time.Duration {
+ if d <= 0 {
+ return time.Nanosecond
+ }
+ spread := float64(d) * jitterFraction
+ delta := (rand.Float64()*2 - 1) * spread
+ out := time.Duration(float64(d) + delta)
+ if out <= 0 {
+ out = time.Nanosecond
+ }
+ return out
+}
diff --git a/event/stream/reconnect_test.go b/event/stream/reconnect_test.go
index 9162768..4dfbb5e 100644
--- a/event/stream/reconnect_test.go
+++ b/event/stream/reconnect_test.go
@@ -29,15 +29,13 @@ const createPullPointRespAlt = `
`
+// --- Recreate after pull failures ------------------------------------
+
func TestStream_RecreatesSubscriptionAfterRepeatedPullErrors(t *testing.T) {
fc := newFakeCaller()
- // Initial subscription.
fc.queueCallMethod(createPullPointResp, nil)
- // Recreated subscription returns a *different* endpoint.
fc.queueCallMethod(createPullPointRespAlt, nil)
- // First pull fails. With ReconnectAfterFailures=1 this triggers a
- // recreate; subsequent pulls go to PullSub_2 which we'll observe.
fc.queueSendSoap("", errors.New("transient failure"))
fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
@@ -48,7 +46,7 @@ func TestStream_RecreatesSubscriptionAfterRepeatedPullErrors(t *testing.T) {
PullTimeout: 50 * time.Millisecond,
ReconnectAfterFailures: 1,
RetryBackoff: 10 * time.Millisecond,
- InitialTermination: 30 * time.Second, // keep renew quiet
+ InitialTermination: 30 * time.Second,
})
require.NoError(t, err)
defer s.Close()
@@ -60,8 +58,6 @@ func TestStream_RecreatesSubscriptionAfterRepeatedPullErrors(t *testing.T) {
defer fc.mu.Unlock()
require.Len(t, fc.callMethodCalls, 2,
"expected exactly 2 CallMethod calls (initial + recreate)")
- // The PullMessages call that delivered the motion event must
- // target the new endpoint.
var newEndpointPulls int
for _, c := range fc.sendSoapCalls {
if c[0] == "http://camera.local/onvif/Events/PullSub_2" {
@@ -75,9 +71,6 @@ func TestStream_RecreatesSubscriptionAfterRepeatedPullErrors(t *testing.T) {
func TestStream_BackoffWhenRecreateFails(t *testing.T) {
fc := newFakeCaller()
fc.queueCallMethod(createPullPointResp, nil)
- // After the initial successful create, every CallMethod (recreate)
- // and SendSoap (pull) fails. The loop should keep retrying with
- // exponential backoff rather than blocking forever or spinning.
fc.mu.Lock()
fc.defaultCall = fakeResp{err: errors.New("recreate fail")}
fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
@@ -118,3 +111,205 @@ func TestStream_RetryBackoffDefault(t *testing.T) {
o := defaultOptions()
assert.Equal(t, time.Second, o.RetryBackoff)
}
+
+func TestStream_DisableReconnectKeepsRetryingOriginalEndpoint(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ fc.mu.Lock()
+ fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
+ fc.mu.Unlock()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{
+ PullTimeout: 10 * time.Millisecond,
+ RetryBackoff: 10 * time.Millisecond,
+ InitialTermination: 30 * time.Second,
+ DisableReconnect: true,
+ })
+ require.NoError(t, err)
+ defer s.Close()
+
+ time.Sleep(200 * time.Millisecond)
+ fc.mu.Lock()
+ calls := len(fc.callMethodCalls)
+ fc.mu.Unlock()
+ assert.Equal(t, 1, calls, "DisableReconnect must prevent recreate; got %d CallMethod calls", calls)
+}
+
+func TestStream_RecreateResetsFailuresAndBackoffOnSuccess(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ fc.queueCallMethod(createPullPointRespAlt, nil)
+ fc.queueSendSoap("", errors.New("first failure"))
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{
+ PullTimeout: 10 * time.Millisecond,
+ ReconnectAfterFailures: 1,
+ RetryBackoff: 10 * time.Millisecond,
+ InitialTermination: 30 * time.Second,
+ })
+ require.NoError(t, err)
+ defer s.Close()
+
+ time.Sleep(200 * time.Millisecond)
+ fc.mu.Lock()
+ calls := len(fc.callMethodCalls)
+ fc.mu.Unlock()
+ assert.Equal(t, 2, calls,
+ "after one failure + successful recreate, no further recreates expected; got %d", calls)
+}
+
+func TestStream_PullPointMutationVisibleToRenewLoopUnderRace(t *testing.T) {
+ // Drives the pullPoint write-by-pullLoop / read-by-renewLoop race
+ // so -race actually exercises the mutex critical sections.
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ for i := 0; i < 50; i++ {
+ fc.queueCallMethod(createPullPointRespAlt, nil)
+ }
+ fc.mu.Lock()
+ fc.defaultSendSoap = fakeResp{err: errors.New("recurring pull fail")}
+ fc.mu.Unlock()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{
+ PullTimeout: 5 * time.Millisecond,
+ ReconnectAfterFailures: 1,
+ RetryBackoff: 1 * time.Millisecond,
+ InitialTermination: 20 * time.Millisecond,
+ RenewMargin: 2 * time.Millisecond,
+ })
+ require.NoError(t, err)
+ defer s.Close()
+
+ time.Sleep(300 * time.Millisecond)
+}
+
+// --- Typed errors from the reconnect path ----------------------------
+
+func TestStream_PullErrorIsTypedErrPullFailed(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ fc.queueSendSoap("", errors.New("transient"))
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{
+ PullTimeout: 50 * time.Millisecond,
+ RetryBackoff: 10 * time.Millisecond,
+ InitialTermination: 30 * time.Second,
+ })
+ require.NoError(t, err)
+ defer s.Close()
+
+ select {
+ case e := <-s.Errors():
+ var pullErr ErrPullFailed
+ require.True(t, errors.As(e, &pullErr), "expected ErrPullFailed, got %T: %v", e, e)
+ assert.Contains(t, pullErr.Err.Error(), "transient")
+ case <-time.After(time.Second):
+ t.Fatal("expected ErrPullFailed on Errors channel")
+ }
+}
+
+func TestStream_RecreateErrorIsTypedErrRecreateFailed(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ fc.mu.Lock()
+ fc.defaultCall = fakeResp{err: errors.New("recreate fail")}
+ fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
+ fc.mu.Unlock()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{
+ PullTimeout: 10 * time.Millisecond,
+ ReconnectAfterFailures: 1,
+ RetryBackoff: 10 * time.Millisecond,
+ InitialTermination: 30 * time.Second,
+ })
+ require.NoError(t, err)
+ defer s.Close()
+
+ deadline := time.Now().Add(time.Second)
+ var sawRecreate bool
+ for time.Now().Before(deadline) && !sawRecreate {
+ select {
+ case e := <-s.Errors():
+ var rec ErrRecreateFailed
+ if errors.As(e, &rec) {
+ sawRecreate = true
+ assert.Contains(t, rec.Err.Error(), "recreate fail")
+ }
+ case <-time.After(50 * time.Millisecond):
+ }
+ }
+ assert.True(t, sawRecreate, "expected at least one ErrRecreateFailed on Errors")
+}
+
+func TestStream_AfterReconnectFlagSetOnFirstPostRecreateBatch(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ fc.queueCallMethod(createPullPointRespAlt, nil)
+
+ fc.queueSendSoap("", errors.New("transient"))
+ fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
+ fc.queueSendSoap(pullMessagesResp(motionMsg("false")), nil)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{
+ PullTimeout: 50 * time.Millisecond,
+ ReconnectAfterFailures: 1,
+ RetryBackoff: 10 * time.Millisecond,
+ InitialTermination: 30 * time.Second,
+ })
+ require.NoError(t, err)
+ defer s.Close()
+
+ ev1 := receive(t, s.Events(), 2*time.Second)
+ assert.True(t, ev1.AfterReconnect, "first event after recreate must carry AfterReconnect=true")
+ assert.Equal(t, StateActive, ev1.State)
+
+ ev2 := receive(t, s.Events(), 2*time.Second)
+ assert.False(t, ev2.AfterReconnect, "subsequent events should not carry AfterReconnect")
+ assert.Equal(t, StateInactive, ev2.State)
+}
+
+// --- Jitter ----------------------------------------------------------
+
+func TestJitter_StaysWithinFraction(t *testing.T) {
+ const base = time.Second
+ low := time.Duration(float64(base) * (1 - jitterFraction))
+ high := time.Duration(float64(base) * (1 + jitterFraction))
+ for i := 0; i < 200; i++ {
+ got := jitter(base)
+ assert.GreaterOrEqual(t, got, low, "iteration %d", i)
+ assert.LessOrEqual(t, got, high, "iteration %d", i)
+ }
+}
+
+func TestJitter_ZeroAndNegativeReturnPositive(t *testing.T) {
+ assert.Greater(t, jitter(0), time.Duration(0))
+ assert.Greater(t, jitter(-time.Second), time.Duration(0))
+}
+
+func TestJitter_VariesAcrossCalls(t *testing.T) {
+ first := jitter(time.Second)
+ allEqual := true
+ for i := 0; i < 10; i++ {
+ if jitter(time.Second) != first {
+ allEqual = false
+ break
+ }
+ }
+ assert.False(t, allEqual, "jitter is producing a constant; rand seed not working")
+}
+
+func TestMaxRecreateBackoff_Is5Minutes(t *testing.T) {
+ assert.Equal(t, 5*time.Minute, maxRecreateBackoff)
+}
diff --git a/event/stream/renew.go b/event/stream/renew.go
new file mode 100644
index 0000000..6a8763a
--- /dev/null
+++ b/event/stream/renew.go
@@ -0,0 +1,64 @@
+package stream
+
+import (
+ "context"
+ "encoding/xml"
+ "fmt"
+ "time"
+
+ "github.com/kerberos-io/onvif/event"
+ "github.com/kerberos-io/onvif/xsd"
+)
+
+// renewLoop refreshes the subscription before InitialTermination expires.
+// Exits when ctx is cancelled. Renew failures surface as ErrRenewFailed
+// on the Errors channel; the loop continues because a permanently
+// failing renew will eventually drop the subscription and the pull
+// loop's reconnect path will recover (recreate is the only reliable
+// recovery once a subscription is GC'd at the camera).
+func (s *Stream) renewLoop(ctx context.Context) {
+ interval := s.opts.InitialTermination - s.opts.RenewMargin
+ if interval <= 0 {
+ // Pathological config (margin >= termination): fall back to
+ // renewing at half the termination so we still refresh,
+ // rather than busy-looping or never renewing.
+ interval = s.opts.InitialTermination / 2
+ if interval <= 0 {
+ interval = time.Second
+ }
+ }
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ if err := renewPullPoint(s.caller, s.getPullPoint(), s.opts); err != nil {
+ s.surfaceError(ErrRenewFailed{Err: err})
+ }
+ }
+ }
+}
+
+// renewPullPoint issues a wsnt:Renew SOAP against the given
+// subscription endpoint with an absolute TerminationTime.
+//
+// WS-BaseNotification §6.1.1 declares TerminationTime as xsd:dateTime
+// OR xsd:duration, but older Hikvision, some Dahua and some Bosch
+// firmwares reject the relative-duration form. We send an absolute
+// UTC datetime to match what production NVRs do.
+func renewPullPoint(c caller, endpoint string, opts Options) error {
+ absoluteEnd := time.Now().UTC().Add(opts.InitialTermination).Format("2006-01-02T15:04:05Z")
+ req := event.Renew{TerminationTime: xsd.String(absoluteEnd)}
+ body, err := xml.Marshal(req)
+ if err != nil {
+ return fmt.Errorf("marshal Renew: %w", err)
+ }
+ resp, err := c.SendSoap(endpoint, string(body))
+ if err != nil {
+ return err
+ }
+ _, err = readClose(resp)
+ return err
+}
diff --git a/event/stream/renew_test.go b/event/stream/renew_test.go
index 20e2146..cc758bb 100644
--- a/event/stream/renew_test.go
+++ b/event/stream/renew_test.go
@@ -132,3 +132,39 @@ func TestStream_RenewErrorSurfacedOnErrorsChannel(t *testing.T) {
type errInjected struct{}
func (errInjected) Error() string { return "injected fake error" }
+
+func TestRenew_SendsAbsoluteDateTimeNotDuration(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{
+ InitialTermination: 30 * time.Millisecond,
+ RenewMargin: 5 * time.Millisecond,
+ })
+ require.NoError(t, err)
+ defer s.Close()
+
+ deadline := time.Now().Add(500 * time.Millisecond)
+ for time.Now().Before(deadline) {
+ if countSendSoapMatching(fc, "Renew") >= 1 {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+
+ fc.mu.Lock()
+ defer fc.mu.Unlock()
+ var renewBody string
+ for _, c := range fc.sendSoapCalls {
+ if strings.Contains(c[1], "Renew") {
+ renewBody = c[1]
+ break
+ }
+ }
+ require.NotEmpty(t, renewBody, "no Renew call observed")
+ // Absolute form is "YYYY-MM-DDTHH:MM:SSZ" not "PTnS".
+ assert.NotContains(t, renewBody, "PT", "Renew should not send relative duration; some firmwares reject it")
+ assert.Regexp(t, `\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z`, renewBody, "Renew should send absolute RFC3339 UTC")
+}
diff --git a/event/stream/soap.go b/event/stream/soap.go
new file mode 100644
index 0000000..4d9bcf6
--- /dev/null
+++ b/event/stream/soap.go
@@ -0,0 +1,189 @@
+package stream
+
+import (
+ "bytes"
+ "encoding/xml"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "regexp"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/kerberos-io/onvif/event"
+ "github.com/kerberos-io/onvif/xsd"
+)
+
+// maxResponseBytes caps the size of a SOAP response we will buffer in
+// memory. ONVIF PullMessages bodies are normally <100KB even with dense
+// analytics payloads; 10 MiB is comfortably above legitimate traffic
+// while keeping a hostile or buggy camera from OOMing the process.
+const maxResponseBytes = 10 << 20
+
+// createPullPoint issues a CreatePullPointSubscription against the
+// device service. Returns the SubscriptionReference Address, which is
+// the endpoint subsequent PullMessages / Renew / Unsubscribe calls
+// target.
+func createPullPoint(c caller, opts Options) (string, error) {
+ term := xsd.String(durationToXSD(opts.InitialTermination))
+ req := event.CreatePullPointSubscription{InitialTerminationTime: &term}
+ if opts.RawTopicFilter != "" {
+ req.Filter = &event.FilterType{
+ TopicExpression: &event.TopicExpressionType{
+ Dialect: xsd.String("http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"),
+ TopicKinds: xsd.String(opts.RawTopicFilter),
+ },
+ }
+ }
+ 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
+}
+
+// pullMessages issues PullMessages against an active subscription
+// endpoint and returns the decoded NotificationMessage list. Empty
+// slice (not error) when the camera had nothing within PullTimeout.
+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
+}
+
+// unsubscribePullPoint sends a best-effort Unsubscribe to release the
+// subscription server-side. Empty endpoint is a no-op (the construction
+// failed before installing one).
+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
+}
+
+// readClose reads at most maxResponseBytes from resp.Body and closes
+// it. LimitReader prevents a hostile or buggy camera from OOMing the
+// agent by streaming an unbounded response.
+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(io.LimitReader(resp.Body, maxResponseBytes))
+ 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.
+//
+// When the camera returns a SOAP Fault instead of the expected
+// response, the fault reason is surfaced as the error so callers can
+// distinguish "auth failed" / "subscription expired" from "unparseable
+// response".
+func unmarshalNode(body, localName string, out any) error {
+ if reason := extractSOAPFault(body); reason != "" {
+ return fmt.Errorf("ONVIF SOAP fault: %s", reason)
+ }
+ 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
+ }
+}
+
+var (
+ // SOAP 1.1: reason
+ soap11FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?faultstring[^>]*>(.*?)\s]+:)?faultstring>`)
+ // SOAP 1.2: ...reason...
+ soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)\s]+:)?Text>`)
+)
+
+// extractSOAPFault returns the human-readable reason text from a SOAP
+// fault, or empty string when the body is not a fault. Handles both
+// SOAP 1.1 (faultstring) and SOAP 1.2 (Reason/Text) shapes.
+func extractSOAPFault(body string) string {
+ if !strings.Contains(body, "Fault") {
+ return ""
+ }
+ if m := soap11FaultRE.FindStringSubmatch(body); len(m) > 1 {
+ return strings.TrimSpace(m[1])
+ }
+ if m := soap12FaultRE.FindStringSubmatch(body); len(m) > 1 {
+ return strings.TrimSpace(m[1])
+ }
+ return ""
+}
+
+// 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/soap_test.go b/event/stream/soap_test.go
index e8c48f5..7c5b6c2 100644
--- a/event/stream/soap_test.go
+++ b/event/stream/soap_test.go
@@ -4,7 +4,6 @@ import (
"context"
"strings"
"testing"
- "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -59,97 +58,29 @@ func TestUnmarshalNode_ReturnsFaultReasonInsteadOfMissingElement(t *testing.T) {
assert.NotContains(t, err.Error(), "missing PullMessagesResponse")
}
-// --- Renew sends absolute datetime -----------------------------------
-
-func TestRenew_SendsAbsoluteDateTimeNotDuration(t *testing.T) {
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
-
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{
- InitialTermination: 30 * time.Millisecond,
- RenewMargin: 5 * time.Millisecond,
- })
- require.NoError(t, err)
- defer s.Close()
-
- deadline := time.Now().Add(500 * time.Millisecond)
- for time.Now().Before(deadline) {
- if countSendSoapMatching(fc, "Renew") >= 1 {
- break
- }
- time.Sleep(10 * time.Millisecond)
- }
-
- fc.mu.Lock()
- defer fc.mu.Unlock()
- var renewBody string
- for _, c := range fc.sendSoapCalls {
- if strings.Contains(c[1], "Renew") {
- renewBody = c[1]
- break
- }
- }
- require.NotEmpty(t, renewBody, "no Renew call observed")
- // Absolute form is "YYYY-MM-DDTHH:MM:SSZ" not "PTnS".
- assert.NotContains(t, renewBody, "PT", "Renew should not send relative duration; some firmwares reject it")
- assert.Regexp(t, `\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z`, renewBody, "Renew should send absolute RFC3339 UTC")
-}
-
// --- Bounded body read -----------------------------------------------
func TestReadClose_LimitsBodySize(t *testing.T) {
- // Build a response with a body just over the limit. readClose must
- // not return more than the limit even if the camera pretends to
- // send more.
if maxResponseBytes < 1024 {
t.Skip("limit too small for this test")
}
big := strings.Repeat("A", maxResponseBytes+1024)
- // Wrap in a minimal SOAP envelope so the body is at least
- // well-formed shape-wise.
body := "" + big + ""
fc := newFakeCaller()
fc.queueCallMethod(body, nil)
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
// Construction will fail because the truncated body has no
- // CreatePullPointSubscriptionResponse — that's fine; what matters
- // is the read completes without OOM.
- _, err := newStream(ctx, fc, Options{})
+ // CreatePullPointSubscriptionResponse — that's fine; what matters is
+ // the read completes without OOM.
+ _, err := newStream(testContext(t), fc, Options{})
assert.Error(t, err)
}
-// --- Close timeout ---------------------------------------------------
-
-func TestClose_BoundedByTimeoutOnHungUnsubscribe(t *testing.T) {
- // Patch closeUnsubscribeTimeout for the duration of the test so the
- // assertion completes promptly. We can't change the const at runtime
- // so we use a short InitialTermination and verify Close still
- // returns within closeUnsubscribeTimeout + slack rather than
- // blocking forever.
- fc := newFakeCaller()
- fc.queueCallMethod(createPullPointResp, nil)
- block := make(chan struct{})
- defer close(block) // release the hung Unsubscribe so the fake's goroutine exits
- fc.mu.Lock()
- fc.blockUnsubscribe = block
- fc.mu.Unlock()
-
+// testContext returns a Background context already wired to cancel via
+// t.Cleanup so the test does not need to manage the cancellation
+// goroutine inline.
+func testContext(t *testing.T) context.Context {
+ t.Helper()
ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
- s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
- require.NoError(t, err)
-
- start := time.Now()
- err = s.Close()
- elapsed := time.Since(start)
- // Unsubscribe is hung, so Close must surface a timeout error from
- // the bounded wait rather than block forever. closeUnsubscribeTimeout
- // is 5s; allow 1s slack for scheduling.
- require.Error(t, err)
- assert.Contains(t, err.Error(), "timeout")
- assert.Less(t, elapsed, closeUnsubscribeTimeout+time.Second,
- "Close exceeded bound (%s); expected ~%s", elapsed, closeUnsubscribeTimeout)
+ t.Cleanup(cancel)
+ return ctx
}
diff --git a/event/stream/stream.go b/event/stream/stream.go
index b1f22d2..4f11415 100644
--- a/event/stream/stream.go
+++ b/event/stream/stream.go
@@ -1,31 +1,15 @@
package stream
import (
- "bytes"
"context"
- "encoding/xml"
- "errors"
"fmt"
- "io"
- "math/rand"
"net/http"
- "regexp"
- "strconv"
- "strings"
"sync"
"time"
"github.com/kerberos-io/onvif"
- "github.com/kerberos-io/onvif/event"
- "github.com/kerberos-io/onvif/xsd"
)
-// maxResponseBytes caps the size of a SOAP response we will buffer in
-// memory. ONVIF PullMessages bodies are normally <100KB even with dense
-// analytics payloads; 10 MiB is comfortably above legitimate traffic
-// while keeping a hostile or buggy camera from OOMing the process.
-const maxResponseBytes = 10 << 20
-
// closeUnsubscribeTimeout bounds the SOAP Unsubscribe call issued by
// Close so a hung camera connection cannot wedge the caller. The
// subscription expires at the camera anyway once InitialTermination
@@ -138,20 +122,6 @@ func (o Options) withDefaults() Options {
return d
}
-// maxRecreateBackoff caps exponential backoff between recreate attempts.
-// Sized for fleet deployments: a 1000-camera setup recovering from a
-// switch reboot would otherwise hammer the network with one recreate
-// attempt per camera per 30s; 5 minutes gives the network time to
-// settle while still recovering promptly when a single camera comes
-// back.
-const maxRecreateBackoff = 5 * time.Minute
-
-// jitterFraction is the symmetric jitter applied to recreate backoff:
-// the actual sleep is sampled from [backoff*(1-jitter), backoff*(1+jitter)].
-// Prevents thundering-herd reconnects when many cameras drop together
-// (switch reboot, NAT timeout).
-const jitterFraction = 0.25
-
// caller is the subset of *onvif.Device the Stream depends on. Tests
// substitute a fake; production code uses the device adapter.
//
@@ -282,6 +252,8 @@ func (s *Stream) Close() error {
return s.closeErr
}
+// run orchestrates the pull and renew goroutines and closes the
+// emission channels once both have exited.
func (s *Stream) run(ctx context.Context) {
var wg sync.WaitGroup
wg.Add(1)
@@ -300,130 +272,8 @@ func (s *Stream) run(ctx context.Context) {
close(s.done)
}
-func (s *Stream) pullLoop(ctx context.Context) {
- var failures int
- recreateBackoff := s.opts.RetryBackoff
- var afterReconnect bool
-
- for {
- if ctx.Err() != nil {
- return
- }
- msgs, err := pullMessages(s.caller, s.getPullPoint(), s.opts)
- if err != nil {
- s.surfaceError(ErrPullFailed{Err: err})
- failures++
- if !s.opts.DisableReconnect && failures >= s.opts.ReconnectAfterFailures {
- justRecreated, cont := s.attemptRecreate(ctx, &failures, &recreateBackoff)
- if !cont {
- return
- }
- if justRecreated {
- afterReconnect = true
- }
- continue
- }
- if !sleepCtx(ctx, s.opts.RetryBackoff) {
- return
- }
- continue
- }
- // Successful pull resets failure tracking.
- failures = 0
- recreateBackoff = s.opts.RetryBackoff
- observedAt := s.now()
- for _, m := range msgs {
- ev := decode(m, s.opts.DeviceID, observedAt)
- if afterReconnect {
- ev.AfterReconnect = true
- // ONVIF replays current state with
- // PropertyInitialized on a new subscription.
- // Clear the flag as soon as we see anything
- // other than Initialized — at that point we
- // have transitioned to live events.
- if ev.Operation != PropertyInitialized {
- afterReconnect = false
- }
- }
- select {
- case <-ctx.Done():
- return
- case s.events <- ev:
- }
- }
- }
-}
-
-// attemptRecreate calls CreatePullPointSubscription and on success
-// installs the new endpoint atomically. The first return is true when
-// recreate succeeded just now (caller flags the next batch with
-// AfterReconnect). The second return is false only if ctx was cancelled
-// during backoff (caller should exit the run loop).
-func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *time.Duration) (justRecreated, cont bool) {
- addr, err := createPullPoint(s.caller, s.opts)
- if err != nil {
- s.surfaceError(ErrRecreateFailed{Err: err})
- if !sleepCtx(ctx, jitter(*backoff)) {
- return false, false
- }
- *backoff *= 2
- if *backoff > maxRecreateBackoff {
- *backoff = maxRecreateBackoff
- }
- return false, true
- }
- s.setPullPoint(addr)
- *failures = 0
- *backoff = s.opts.RetryBackoff
- return true, true
-}
-
-// jitter returns d perturbed by ±jitterFraction. Used to spread
-// recreate attempts across a fleet so a synchronised drop (switch
-// reboot, DHCP storm) does not cause a synchronised reconnect surge.
-// Returns at least 1ns to keep sleepCtx happy.
-func jitter(d time.Duration) time.Duration {
- if d <= 0 {
- return time.Nanosecond
- }
- spread := float64(d) * jitterFraction
- delta := (rand.Float64()*2 - 1) * spread
- out := time.Duration(float64(d) + delta)
- if out <= 0 {
- out = time.Nanosecond
- }
- return out
-}
-
-// renewLoop refreshes the subscription before InitialTermination expires.
-// Exits when ctx is cancelled.
-func (s *Stream) renewLoop(ctx context.Context) {
- interval := s.opts.InitialTermination - s.opts.RenewMargin
- if interval <= 0 {
- // Pathological config (margin >= termination): fall back to
- // renewing at half the termination so we still refresh,
- // rather than busy-looping or never renewing.
- interval = s.opts.InitialTermination / 2
- if interval <= 0 {
- interval = time.Second
- }
- }
- ticker := time.NewTicker(interval)
- defer ticker.Stop()
- for {
- select {
- case <-ctx.Done():
- return
- case <-ticker.C:
- if err := renewPullPoint(s.caller, s.getPullPoint(), s.opts); err != nil {
- s.surfaceError(ErrRenewFailed{Err: err})
- }
- }
- }
-}
-
// surfaceError sends err on the errors channel non-blockingly so a
-// stalled consumer cannot block the pull loop.
+// stalled consumer cannot block the pull or renew loop.
func (s *Stream) surfaceError(err error) {
select {
case s.errors <- err:
@@ -443,179 +293,3 @@ func sleepCtx(ctx context.Context, d time.Duration) bool {
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.RawTopicFilter != "" {
- req.Filter = &event.FilterType{
- TopicExpression: &event.TopicExpressionType{
- Dialect: xsd.String("http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"),
- TopicKinds: xsd.String(opts.RawTopicFilter),
- },
- }
- }
- 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 renewPullPoint(c caller, endpoint string, opts Options) error {
- // WS-BaseNotification §6.1.1 declares TerminationTime as
- // xsd:dateTime OR xsd:duration, but older Hikvision, some Dahua
- // and some Bosch firmwares reject the relative-duration form. Send
- // an absolute UTC datetime to match what production NVRs do.
- absoluteEnd := time.Now().UTC().Add(opts.InitialTermination).Format("2006-01-02T15:04:05Z")
- req := event.Renew{TerminationTime: xsd.String(absoluteEnd)}
- body, err := xml.Marshal(req)
- if err != nil {
- return fmt.Errorf("marshal Renew: %w", err)
- }
- resp, err := c.SendSoap(endpoint, string(body))
- if err != nil {
- return err
- }
- _, err = readClose(resp)
- return err
-}
-
-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()
- // LimitReader prevents a hostile or buggy camera from OOMing the
- // agent by streaming an unbounded response body.
- b, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
- 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.
-//
-// When the camera returns a SOAP Fault instead of the expected
-// response, the fault reason is surfaced as the error so callers can
-// distinguish "auth failed" / "subscription expired" from "unparseable
-// response".
-func unmarshalNode(body, localName string, out any) error {
- if reason := extractSOAPFault(body); reason != "" {
- return fmt.Errorf("ONVIF SOAP fault: %s", reason)
- }
- 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
- }
-}
-
-var (
- // SOAP 1.1: reason
- soap11FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?faultstring[^>]*>(.*?)\s]+:)?faultstring>`)
- // SOAP 1.2: ...reason...
- soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)\s]+:)?Text>`)
-)
-
-// extractSOAPFault returns the human-readable reason text from a SOAP
-// fault, or empty string when the body is not a fault. Handles both
-// SOAP 1.1 (faultstring) and SOAP 1.2 (Reason/Text) shapes.
-func extractSOAPFault(body string) string {
- if !strings.Contains(body, "Fault") {
- return ""
- }
- if m := soap11FaultRE.FindStringSubmatch(body); len(m) > 1 {
- return strings.TrimSpace(m[1])
- }
- if m := soap12FaultRE.FindStringSubmatch(body); len(m) > 1 {
- return strings.TrimSpace(m[1])
- }
- return ""
-}
-
-// 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
index b7343fe..062cac7 100644
--- a/event/stream/stream_test.go
+++ b/event/stream/stream_test.go
@@ -351,3 +351,96 @@ func TestStream_DoesNotPanicOnPullExitingDuringClose(t *testing.T) {
_ = s.Close()
})
}
+
+// --- Close error / timeout paths -------------------------------------
+
+func TestClose_ReturnsUnsubscribeError(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ fc.mu.Lock()
+ fc.defaultSendSoap = fakeResp{err: errors.New("simulated transport failure")}
+ fc.mu.Unlock()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
+ require.NoError(t, err)
+
+ err = s.Close()
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "unsubscribe pull point")
+ assert.Contains(t, err.Error(), "simulated transport failure")
+}
+
+func TestClose_BoundedByTimeoutOnHungUnsubscribe(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+ block := make(chan struct{})
+ defer close(block) // release the hung Unsubscribe so the fake's goroutine exits
+ fc.mu.Lock()
+ fc.blockUnsubscribe = block
+ fc.mu.Unlock()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
+ require.NoError(t, err)
+
+ start := time.Now()
+ err = s.Close()
+ elapsed := time.Since(start)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "timeout")
+ assert.Less(t, elapsed, closeUnsubscribeTimeout+time.Second,
+ "Close exceeded bound (%s); expected ~%s", elapsed, closeUnsubscribeTimeout)
+}
+
+// --- NewStream edge cases --------------------------------------------
+
+func TestNewStream_CtxAlreadyCancelled(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueCallMethod(createPullPointResp, nil)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel() // cancel before NewStream
+
+ s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
+ require.NoError(t, err)
+ require.NotNil(t, s)
+
+ select {
+ case _, ok := <-s.Events():
+ assert.False(t, ok, "events channel should be closed when ctx is pre-cancelled")
+ case <-time.After(time.Second):
+ t.Fatal("events channel was not closed within 1s")
+ }
+ _ = s.Close()
+}
+
+// --- fakeCaller self-test --------------------------------------------
+
+func TestFakeCaller_QueueThenDefaultFallback(t *testing.T) {
+ fc := newFakeCaller()
+ fc.queueSendSoap("first", nil)
+ fc.queueSendSoap("second", nil)
+
+ r1, err := fc.SendSoap("ep", "body")
+ require.NoError(t, err)
+ b1 := make([]byte, 10)
+ n, _ := r1.Body.Read(b1)
+ assert.Equal(t, "first", string(b1[:n]))
+
+ r2, _ := fc.SendSoap("ep", "body")
+ b2 := make([]byte, 10)
+ n, _ = r2.Body.Read(b2)
+ assert.Equal(t, "second", string(b2[:n]))
+
+ // Queue is exhausted; default kicks in.
+ r3, err := fc.SendSoap("ep", "body")
+ require.NoError(t, err)
+ require.NotNil(t, r3)
+ b3 := make([]byte, 2048)
+ n, _ = r3.Body.Read(b3)
+ assert.Contains(t, string(b3[:n]), "PullMessagesResponse",
+ "default SendSoap should be an empty PullMessagesResponse envelope")
+}
diff --git a/event/stream/types.go b/event/stream/types.go
index 9f1557d..a52f314 100644
--- a/event/stream/types.go
+++ b/event/stream/types.go
@@ -169,3 +169,41 @@ type Event struct {
// first event whose Operation is not PropertyInitialized.
AfterReconnect bool
}
+
+// Op identifies which Stream operation failed. Used by ErrPullFailed,
+// ErrRenewFailed and ErrRecreateFailed so consumers can branch with
+// errors.As without parsing the wrapped message.
+type Op string
+
+const (
+ OpPull Op = "pull"
+ OpRenew Op = "renew"
+ OpRecreate Op = "recreate"
+)
+
+// ErrPullFailed wraps a transient PullMessages failure. The pull loop
+// surfaces it on the Errors channel and continues. Consumers can match
+// with errors.As(err, &stream.ErrPullFailed{}).
+type ErrPullFailed struct{ Err error }
+
+func (e ErrPullFailed) Error() string { return fmt.Sprintf("pull messages: %v", e.Err) }
+func (e ErrPullFailed) Unwrap() error { return e.Err }
+func (ErrPullFailed) Op() Op { return OpPull }
+
+// ErrRenewFailed wraps a Renew SOAP failure. Renew errors are usually
+// recovered implicitly: the subscription dies, pull starts failing,
+// and the reconnect logic recreates it.
+type ErrRenewFailed struct{ Err error }
+
+func (e ErrRenewFailed) Error() string { return fmt.Sprintf("renew pull point: %v", e.Err) }
+func (e ErrRenewFailed) Unwrap() error { return e.Err }
+func (ErrRenewFailed) Op() Op { return OpRenew }
+
+// ErrRecreateFailed wraps a failed CreatePullPointSubscription during
+// the reconnect path. The loop continues with exponential backoff;
+// consumers seeing this repeatedly should consider the camera offline.
+type ErrRecreateFailed struct{ Err error }
+
+func (e ErrRecreateFailed) Error() string { return fmt.Sprintf("recreate pull point: %v", e.Err) }
+func (e ErrRecreateFailed) Unwrap() error { return e.Err }
+func (ErrRecreateFailed) Op() Op { return OpRecreate }
diff --git a/event/stream/types_test.go b/event/stream/types_test.go
index ed2399d..8b295bb 100644
--- a/event/stream/types_test.go
+++ b/event/stream/types_test.go
@@ -1,6 +1,7 @@
package stream
import (
+ "errors"
"testing"
"time"
@@ -114,3 +115,29 @@ func TestEventFieldAssignmentRoundTrip(t *testing.T) {
assert.True(t, e.Timestamp.Equal(now))
assert.True(t, e.DeviceTime.Equal(deviceTime))
}
+
+// --- Typed errors -----------------------------------------------------
+
+func TestTypedErrors_UnwrapAndOp(t *testing.T) {
+ inner := errors.New("boom")
+ tests := []struct {
+ name string
+ err error
+ op Op
+ }{
+ {"pull", ErrPullFailed{Err: inner}, OpPull},
+ {"renew", ErrRenewFailed{Err: inner}, OpRenew},
+ {"recreate", ErrRecreateFailed{Err: inner}, OpRecreate},
+ }
+ for _, tc := range tests {
+ t.Run(tc.name, func(t *testing.T) {
+ assert.True(t, errors.Is(tc.err, inner), "errors.Is should unwrap to inner")
+ assert.Contains(t, tc.err.Error(), "boom")
+ if e, ok := tc.err.(interface{ Op() Op }); ok {
+ assert.Equal(t, tc.op, e.Op())
+ } else {
+ t.Fatalf("%T does not expose Op()", tc.err)
+ }
+ })
+ }
+}