From a1fc7832efdc998c60ac2f1a2359b34eb69699ae Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Thu, 21 May 2026 15:14:40 +0200 Subject: [PATCH] refactor(event/stream): align source and test files 1:1 by concern MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously stream.go was a 621-line monolith holding the Stream type, SOAP plumbing, pull loop, renew loop, recreate logic and jitter. The test side had grown five orphan files (renew_test.go, reconnect_test.go, soap_test.go, jitter_test.go, coverage_test.go) with no matching source files. The mismatch made it harder than necessary to find the code that backed a given test. This commit splits stream.go by concern so each source file has its own test file alongside it. Files <100 LOC (errors, jitter) were folded into their conceptual parents rather than left as fragments. New layout — 8 source + 8 test + helpers (test utility) + doc: stream.go <-> stream_test.go Stream type, Options, lifecycle soap.go <-> soap_test.go SOAP plumbing + fault detection renew.go <-> renew_test.go Renew loop and absolute time reconnect.go <-> reconnect_test.go Pull loop, recreate, jitter decode.go <-> decode_test.go NotificationMessage -> Event types.go <-> types_test.go Event types + typed errors topics.go <-> topics_test.go Classifier table doc.go Package godoc landing page helpers_test.go waitFor (test-only utility) Mergers ------- * errors.go (typed error wrappers, 42 LOC) -> types.go. ErrPullFailed / ErrRenewFailed / ErrRecreateFailed are part of the type system, not a separate concern. * jitter.go (40 LOC) -> reconnect.go. jitter is an implementation detail of attemptRecreate, used nowhere else. Test distribution ----------------- * coverage_test.go was a catch-all; tests moved to the file matching the function under test: - Close*, NewStream_*, FakeCaller_* -> stream_test.go - DisableReconnect_*, RecreateResets_*, PullPointMutation_* -> reconnect_test.go - Decode_*, ExtractState_* -> decode_test.go * soap_test.go shed the two orphans that did not belong there: - TestRenew_SendsAbsoluteDateTimeNotDuration -> renew_test.go - TestClose_BoundedByTimeoutOnHungUnsubscribe -> stream_test.go * errors_test.go's pure type tests -> types_test.go * errors_test.go's Stream-integration tests -> reconnect_test.go * jitter_test.go -> reconnect_test.go No behaviour change. Test suite passes -race clean. --- event/stream/coverage_test.go | 295 ----------------------------- event/stream/decode_test.go | 101 ++++++++++ event/stream/errors.go | 42 ----- event/stream/errors_test.go | 133 ------------- event/stream/jitter_test.go | 44 ----- event/stream/reconnect.go | 127 +++++++++++++ event/stream/reconnect_test.go | 215 ++++++++++++++++++++- event/stream/renew.go | 64 +++++++ event/stream/renew_test.go | 36 ++++ event/stream/soap.go | 189 +++++++++++++++++++ event/stream/soap_test.go | 89 +-------- event/stream/stream.go | 332 +-------------------------------- event/stream/stream_test.go | 93 +++++++++ event/stream/types.go | 38 ++++ event/stream/types_test.go | 27 +++ 15 files changed, 893 insertions(+), 932 deletions(-) delete mode 100644 event/stream/coverage_test.go delete mode 100644 event/stream/errors.go delete mode 100644 event/stream/errors_test.go delete mode 100644 event/stream/jitter_test.go create mode 100644 event/stream/reconnect.go create mode 100644 event/stream/renew.go create mode 100644 event/stream/soap.go 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) + } + }) + } +}