mirror of
https://github.com/kerberos-io/onvif.git
synced 2026-08-23 15:08:33 +00:00
Merge pull request #5 from sharedjourney/feature/event-stream
feat(event/stream): add channel-based ONVIF event consumer
This commit is contained in:
Cédric Verstraeten
@@ -32,3 +32,23 @@ python3 python/gen_commands.py
|
||||
|
||||
> **Note:** You can also typically run the generator within your IDE thanks to the `//go:generate` lines
|
||||
> towards the top of the `types.go` files.
|
||||
|
||||
## Higher-level helpers
|
||||
|
||||
Some web service directories ship hand-written, higher-level helpers
|
||||
built on top of the wire-layer commands. These are normal Go packages
|
||||
— **not** covered by the `gen_commands.py` workflow above and not
|
||||
expected to be regenerated.
|
||||
|
||||
- [event/stream](../event/stream) — channel-based event consumer that
|
||||
owns the pull-point subscription lifecycle (Create, Pull, Renew,
|
||||
Unsubscribe, reconnect with jittered backoff) and decodes
|
||||
notifications into normalized typed Events. Vendor topic strings
|
||||
(AXIS, Hikvision, Avigilon, Hanwha, Bosch, Dahua) are classified
|
||||
into a small set of `Kind` values. See the package `doc.go` for the
|
||||
public surface and usage.
|
||||
- [event/topic](../event/topic) — topic identifier helpers.
|
||||
|
||||
When adding a similar higher-level helper, place it under the relevant
|
||||
web service directory as a sub-package so consumers find it next to
|
||||
the wire-layer types it builds on.
|
||||
|
||||
96
event/stream/decode.go
Normal file
96
event/stream/decode.go
Normal file
@@ -0,0 +1,96 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/onvif/event"
|
||||
)
|
||||
|
||||
// decode converts a single ONVIF NotificationMessage into a normalized
|
||||
// Event. Topic, Source and Data are always populated even when Kind is
|
||||
// KindUnknown so consumers can fall back to the wire form.
|
||||
func decode(msg event.NotificationMessage, deviceID string, observedAt time.Time) Event {
|
||||
topic := string(msg.Topic.TopicKinds)
|
||||
desc := msg.Message.Message
|
||||
return Event{
|
||||
Kind: Classify(topic),
|
||||
State: extractState(desc.Data.SimpleItem),
|
||||
Operation: parsePropertyOperation(string(desc.PropertyOperation)),
|
||||
DeviceID: deviceID,
|
||||
Source: simpleItemsToMap(desc.Source.SimpleItem),
|
||||
Data: simpleItemsToMap(desc.Data.SimpleItem),
|
||||
Topic: topic,
|
||||
Timestamp: observedAt,
|
||||
DeviceTime: parseDeviceTime(string(desc.UtcTime)),
|
||||
}
|
||||
}
|
||||
|
||||
// simpleItemsToMap returns nil for an empty list so empty notifications
|
||||
// do not allocate.
|
||||
func simpleItemsToMap(items []event.SimpleItem) map[string]string {
|
||||
if len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
m := make(map[string]string, len(items))
|
||||
for _, it := range items {
|
||||
m[string(it.Name)] = string(it.Value)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// extractState scans Data items for a boolean-like value, returning the
|
||||
// first match. Returns StateUnknown for edge-triggered topics like
|
||||
// LineDetector/Crossed whose Data carries only an ObjectId.
|
||||
func extractState(items []event.SimpleItem) State {
|
||||
for _, it := range items {
|
||||
switch strings.ToLower(strings.TrimSpace(string(it.Value))) {
|
||||
case "true", "1", "active":
|
||||
return StateActive
|
||||
case "false", "0", "inactive":
|
||||
return StateInactive
|
||||
}
|
||||
}
|
||||
return StateUnknown
|
||||
}
|
||||
|
||||
// parsePropertyOperation returns PropertyUnknown for absent (optional
|
||||
// per WS-Notification) or unrecognised values.
|
||||
func parsePropertyOperation(s string) PropertyOperation {
|
||||
switch s {
|
||||
case "Initialized":
|
||||
return PropertyInitialized
|
||||
case "Changed":
|
||||
return PropertyChanged
|
||||
case "Deleted":
|
||||
return PropertyDeleted
|
||||
default:
|
||||
return PropertyUnknown
|
||||
}
|
||||
}
|
||||
|
||||
// parseDeviceTime parses wsnt:UtcTime, returning the zero time when
|
||||
// absent or unparseable. Real cameras emit several flavours: with /
|
||||
// without sub-seconds, colon or compact ("+0200") offsets, and some
|
||||
// older Hikvision firmwares omit the timezone entirely (treated as
|
||||
// UTC per WS-BaseNotification which mandates UTC for UtcTime).
|
||||
func parseDeviceTime(s string) time.Time {
|
||||
if s == "" {
|
||||
return time.Time{}
|
||||
}
|
||||
for _, layout := range deviceTimeLayouts {
|
||||
if t, err := time.Parse(layout, s); err == nil {
|
||||
return t.UTC()
|
||||
}
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
|
||||
var deviceTimeLayouts = []string{
|
||||
time.RFC3339Nano,
|
||||
time.RFC3339,
|
||||
"2006-01-02T15:04:05.999-0700", // Geovision
|
||||
"2006-01-02T15:04:05-0700", // some Dahua
|
||||
"2006-01-02T15:04:05.999",
|
||||
"2006-01-02T15:04:05", // older Hikvision (no timezone)
|
||||
}
|
||||
350
event/stream/decode_test.go
Normal file
350
event/stream/decode_test.go
Normal file
@@ -0,0 +1,350 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/onvif/event"
|
||||
"github.com/kerberos-io/onvif/xsd"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// msg builds a NotificationMessage from the topic and a (PropertyOperation,
|
||||
// UtcTime, source items, data items) tuple so tests stay short and intent
|
||||
// is visible at the call site.
|
||||
func msg(topic, propOp, utcTime string, source, data map[string]string) event.NotificationMessage {
|
||||
toItems := func(m map[string]string) []event.SimpleItem {
|
||||
if len(m) == 0 {
|
||||
return nil
|
||||
}
|
||||
items := make([]event.SimpleItem, 0, len(m))
|
||||
for k, v := range m {
|
||||
items = append(items, event.SimpleItem{
|
||||
Name: xsd.AnyType(k),
|
||||
Value: xsd.AnyType(v),
|
||||
})
|
||||
}
|
||||
return items
|
||||
}
|
||||
return event.NotificationMessage{
|
||||
Topic: event.Topic{TopicKinds: xsd.String(topic)},
|
||||
Message: event.MessageBody{
|
||||
Message: event.MessageDescription{
|
||||
PropertyOperation: xsd.AnyType(propOp),
|
||||
UtcTime: xsd.AnyType(utcTime),
|
||||
Source: event.Source{SimpleItem: toItems(source)},
|
||||
Data: event.Data{SimpleItem: toItems(data)},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecode_MotionActive(t *testing.T) {
|
||||
observedAt := time.Date(2026, 5, 21, 10, 30, 1, 0, time.UTC)
|
||||
in := msg(
|
||||
"tns1:RuleEngine/CellMotionDetector/Motion",
|
||||
"Changed",
|
||||
"2026-05-21T10:30:00Z",
|
||||
map[string]string{
|
||||
"VideoSourceConfigurationToken": "VideoSourceConfigToken0",
|
||||
"Rule": "MyMotionRule",
|
||||
},
|
||||
map[string]string{"IsMotion": "true"},
|
||||
)
|
||||
|
||||
ev := decode(in, "axis-cam-01", observedAt)
|
||||
|
||||
assert.Equal(t, KindMotion, ev.Kind)
|
||||
assert.Equal(t, StateActive, ev.State)
|
||||
assert.Equal(t, PropertyChanged, ev.Operation)
|
||||
assert.Equal(t, "axis-cam-01", ev.DeviceID)
|
||||
assert.Equal(t, "VideoSourceConfigToken0", ev.Source["VideoSourceConfigurationToken"])
|
||||
assert.Equal(t, "MyMotionRule", ev.Source["Rule"])
|
||||
assert.Equal(t, "true", ev.Data["IsMotion"])
|
||||
assert.Equal(t, "tns1:RuleEngine/CellMotionDetector/Motion", ev.Topic)
|
||||
assert.True(t, ev.Timestamp.Equal(observedAt))
|
||||
assert.Equal(t, time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC), ev.DeviceTime)
|
||||
}
|
||||
|
||||
func TestDecode_MotionInactive(t *testing.T) {
|
||||
in := msg(
|
||||
"tns1:VideoSource/MotionAlarm",
|
||||
"Changed",
|
||||
"",
|
||||
nil,
|
||||
map[string]string{"State": "false"},
|
||||
)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Equal(t, KindMotion, ev.Kind)
|
||||
assert.Equal(t, StateInactive, ev.State)
|
||||
}
|
||||
|
||||
func TestDecode_HanwhaNumericMotionValue(t *testing.T) {
|
||||
// Hanwha emits xsd:string values "0"/"1" instead of xsd:boolean.
|
||||
in := msg(
|
||||
"tns1:VideoAnalytics/tnssamsung:MotionDetection",
|
||||
"Changed",
|
||||
"",
|
||||
nil,
|
||||
map[string]string{"Motion": "1"},
|
||||
)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Equal(t, KindMotion, ev.Kind)
|
||||
assert.Equal(t, StateActive, ev.State)
|
||||
}
|
||||
|
||||
func TestDecode_AvigilonActiveLiteral(t *testing.T) {
|
||||
// Avigilon and a handful of older firmwares emit "active"/"inactive"
|
||||
// as the Data value rather than a boolean.
|
||||
in := msg(
|
||||
"tns1:Device/tns1:Trigger/tns1:Relay",
|
||||
"Changed",
|
||||
"",
|
||||
map[string]string{"RelayToken": "Relay-1"},
|
||||
map[string]string{"LogicalState": "active"},
|
||||
)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Equal(t, KindDigitalOutput, ev.Kind)
|
||||
assert.Equal(t, StateActive, ev.State)
|
||||
assert.Equal(t, "Relay-1", ev.Source["RelayToken"])
|
||||
}
|
||||
|
||||
func TestDecode_AxisObjectAnalyticsMultiItem(t *testing.T) {
|
||||
// AOA emits active + classType + confidence in the same Data list.
|
||||
// The decoder must preserve every item; State picks the first
|
||||
// boolean-like value, which is 'active'.
|
||||
in := msg(
|
||||
"tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1",
|
||||
"Changed",
|
||||
"",
|
||||
map[string]string{"Source": "device1Scene1"},
|
||||
map[string]string{
|
||||
"active": "1",
|
||||
"classType": "Human",
|
||||
"confidence": "92",
|
||||
},
|
||||
)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Equal(t, KindObjectDetected, ev.Kind)
|
||||
assert.Equal(t, StateActive, ev.State)
|
||||
assert.Equal(t, "Human", ev.Data["classType"])
|
||||
assert.Equal(t, "92", ev.Data["confidence"])
|
||||
assert.Equal(t, "1", ev.Data["active"])
|
||||
}
|
||||
|
||||
func TestDecode_LineDetectorCrossedHasNoState(t *testing.T) {
|
||||
// Edge-triggered topic — Data carries ObjectId, not a boolean. State
|
||||
// must remain Unknown so consumers do not misread it as level-Active.
|
||||
in := msg(
|
||||
"tns1:RuleEngine/LineDetector/Crossed",
|
||||
"Changed",
|
||||
"",
|
||||
map[string]string{"VideoSourceConfigurationToken": "vsct0", "Rule": "LineRule"},
|
||||
map[string]string{"ObjectId": "42"},
|
||||
)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Equal(t, KindObjectDetected, ev.Kind)
|
||||
assert.Equal(t, StateUnknown, ev.State)
|
||||
assert.Equal(t, "42", ev.Data["ObjectId"])
|
||||
}
|
||||
|
||||
func TestDecode_UnknownTopicStillPreservesWireData(t *testing.T) {
|
||||
// Kind unknown does not mean discard: consumers may want to log or
|
||||
// route on the raw topic when classification misses.
|
||||
in := msg(
|
||||
"tns1:UserAlarm/IVA",
|
||||
"",
|
||||
"",
|
||||
nil,
|
||||
map[string]string{"Custom": "true"},
|
||||
)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Equal(t, KindUnknown, ev.Kind)
|
||||
assert.Equal(t, "tns1:UserAlarm/IVA", ev.Topic)
|
||||
assert.Equal(t, "true", ev.Data["Custom"])
|
||||
}
|
||||
|
||||
func TestDecode_PropertyOperationVariants(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
in string
|
||||
want PropertyOperation
|
||||
}{
|
||||
{"initialized", "Initialized", PropertyInitialized},
|
||||
{"changed", "Changed", PropertyChanged},
|
||||
{"deleted", "Deleted", PropertyDeleted},
|
||||
{"absent", "", PropertyUnknown},
|
||||
{"unrecognised", "Bogus", PropertyUnknown},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
in := msg("tns1:VideoSource/MotionAlarm", tc.in, "", nil, nil)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Equal(t, tc.want, ev.Operation)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecode_DeviceTimeParsing(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
in string
|
||||
want time.Time
|
||||
}{
|
||||
{"rfc3339_utc", "2026-05-21T10:30:00Z", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
|
||||
{"rfc3339_with_offset", "2026-05-21T12:30:00+02:00", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
|
||||
{"rfc3339_subsecond", "2026-05-21T10:30:00.500Z", time.Date(2026, 5, 21, 10, 30, 0, 500_000_000, time.UTC)},
|
||||
{"absent", "", time.Time{}},
|
||||
{"unparseable", "not-a-date", time.Time{}},
|
||||
}
|
||||
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())
|
||||
if tc.want.IsZero() {
|
||||
assert.True(t, ev.DeviceTime.IsZero(), "DeviceTime=%v", ev.DeviceTime)
|
||||
} else {
|
||||
assert.True(t, ev.DeviceTime.Equal(tc.want), "got=%v want=%v", ev.DeviceTime, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecode_EmptySourceAndDataYieldNilMaps(t *testing.T) {
|
||||
// Matches the zero-value contract in types_test.go: callers can
|
||||
// safely len() and index into Source/Data without nil-checking, but
|
||||
// we do not allocate an empty map for empty notifications.
|
||||
in := msg("tns1:VideoSource/MotionAlarm", "Changed", "", nil, nil)
|
||||
ev := decode(in, "dev", time.Now())
|
||||
assert.Nil(t, ev.Source)
|
||||
assert.Nil(t, ev.Data)
|
||||
}
|
||||
|
||||
func TestDecode_StateValueIsCaseInsensitive(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
value string
|
||||
want State
|
||||
}{
|
||||
{"true_lower", "true", StateActive},
|
||||
{"true_upper", "TRUE", StateActive},
|
||||
{"true_mixed", "True", StateActive},
|
||||
{"false_lower", "false", StateInactive},
|
||||
{"false_mixed", "False", StateInactive},
|
||||
{"active_mixed", "Active", StateActive},
|
||||
{"inactive_mixed", "Inactive", StateInactive},
|
||||
{"one", "1", StateActive},
|
||||
{"zero", "0", StateInactive},
|
||||
{"empty", "", StateUnknown},
|
||||
{"nonsense", "maybe", 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)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// --- 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")
|
||||
}
|
||||
28
event/stream/doc.go
Normal file
28
event/stream/doc.go
Normal file
@@ -0,0 +1,28 @@
|
||||
// Package stream is a typed, channel-based consumer for ONVIF device
|
||||
// events. It hides the SOAP/XML, pull-point subscription lifecycle,
|
||||
// subscription renewal and vendor-specific topic conventions behind a
|
||||
// single Event stream.
|
||||
//
|
||||
// # Usage
|
||||
//
|
||||
// dev, _ := onvif.NewDevice(onvif.DeviceParams{Xaddr: "...", Username: "...", Password: "..."})
|
||||
// s, err := stream.NewStream(ctx, dev, stream.Options{DeviceID: "front-door"})
|
||||
// if err != nil { /* construction failed: auth, network, or no event support */ }
|
||||
// defer s.Close()
|
||||
//
|
||||
// for ev := range s.Events() {
|
||||
// switch ev.Kind {
|
||||
// case stream.KindMotion:
|
||||
// if ev.State == stream.StateActive { /* start recording */ }
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// NewStream performs network I/O so auth and reachability failures
|
||||
// surface synchronously. Events and Errors close when the Stream stops;
|
||||
// Errors sends are non-blocking so a stalled consumer drops older
|
||||
// errors rather than blocking the pull loop. After a silent reconnect,
|
||||
// the next batch's events carry Event.AfterReconnect=true.
|
||||
//
|
||||
// See topics.go for the verified topic→Kind mapping across AXIS,
|
||||
// Hikvision, Avigilon, Hanwha, Bosch and Dahua.
|
||||
package stream
|
||||
21
event/stream/helpers_test.go
Normal file
21
event/stream/helpers_test.go
Normal file
@@ -0,0 +1,21 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// waitFor polls cond at 10ms intervals up to d. Fails the test with msg
|
||||
// if cond never returns true. Centralises the pattern that appears in
|
||||
// renew/reconnect/stream tests so retries are uniform.
|
||||
func waitFor(t *testing.T, d time.Duration, msg string, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(d)
|
||||
for time.Now().Before(deadline) {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("waitFor timed out after %s: %s", d, msg)
|
||||
}
|
||||
108
event/stream/reconnect.go
Normal file
108
event/stream/reconnect.go
Normal file
@@ -0,0 +1,108 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math/rand"
|
||||
"time"
|
||||
)
|
||||
|
||||
// maxRecreateBackoff caps exponential backoff between recreate
|
||||
// attempts. Sized for fleet deployments: at 30s a 1000-camera setup
|
||||
// recovering from a switch reboot would generate sustained
|
||||
// reconnect traffic; 5 minutes lets the network settle.
|
||||
const maxRecreateBackoff = 5 * time.Minute
|
||||
|
||||
// jitterFraction prevents thundering-herd reconnects when many
|
||||
// cameras drop together (switch reboot, NAT timeout).
|
||||
const jitterFraction = 0.25
|
||||
|
||||
// pullLoop runs PullMessages → decode → Events. After
|
||||
// ReconnectAfterFailures consecutive errors it asks attemptRecreate
|
||||
// to rebuild the subscription. The next batch's events carry
|
||||
// AfterReconnect=true so consumers can suppress the Initialized
|
||||
// replay ONVIF emits on a new subscription.
|
||||
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
|
||||
}
|
||||
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
|
||||
// Clear once the camera transitions past the
|
||||
// Initialized replay to live events.
|
||||
if ev.Operation != PropertyInitialized {
|
||||
afterReconnect = false
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case s.events <- ev:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// attemptRecreate returns (justRecreated, cont). cont is false only
|
||||
// when ctx cancelled during backoff so the caller exits the 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 perturbs d by ±jitterFraction so synchronised drops do not
|
||||
// produce a synchronised reconnect surge.
|
||||
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
|
||||
}
|
||||
315
event/stream/reconnect_test.go
Normal file
315
event/stream/reconnect_test.go
Normal file
@@ -0,0 +1,315 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// createPullPointRespAlt mirrors the first fixture but returns a
|
||||
// different SubscriptionReference Address so a test can prove that
|
||||
// subsequent pulls hit the recreated endpoint.
|
||||
const createPullPointRespAlt = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
|
||||
xmlns:wsa="http://www.w3.org/2005/08/addressing"
|
||||
xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
|
||||
<env:Body>
|
||||
<tev:CreatePullPointSubscriptionResponse>
|
||||
<tev:SubscriptionReference>
|
||||
<wsa:Address>http://camera.local/onvif/Events/PullSub_2</wsa:Address>
|
||||
</tev:SubscriptionReference>
|
||||
<tev:CurrentTime>2026-05-21T10:30:10Z</tev:CurrentTime>
|
||||
<tev:TerminationTime>2026-05-21T10:31:10Z</tev:TerminationTime>
|
||||
</tev:CreatePullPointSubscriptionResponse>
|
||||
</env:Body>
|
||||
</env:Envelope>`
|
||||
|
||||
// --- Recreate after pull failures ------------------------------------
|
||||
|
||||
func TestStream_RecreatesSubscriptionAfterRepeatedPullErrors(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
fc.queueCallMethod(createPullPointRespAlt, nil)
|
||||
|
||||
fc.queueSendSoap("", errors.New("transient failure"))
|
||||
fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{
|
||||
DeviceID: "cam-1",
|
||||
PullTimeout: 50 * time.Millisecond,
|
||||
ReconnectAfterFailures: 1,
|
||||
RetryBackoff: 10 * time.Millisecond,
|
||||
InitialTermination: 30 * time.Second,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
defer s.Close()
|
||||
|
||||
ev := receive(t, s.Events(), 2*time.Second)
|
||||
assert.Equal(t, KindMotion, ev.Kind)
|
||||
|
||||
fc.mu.Lock()
|
||||
defer fc.mu.Unlock()
|
||||
require.Len(t, fc.callMethodCalls, 2,
|
||||
"expected exactly 2 CallMethod calls (initial + recreate)")
|
||||
var newEndpointPulls int
|
||||
for _, c := range fc.sendSoapCalls {
|
||||
if c[0] == "http://camera.local/onvif/Events/PullSub_2" {
|
||||
newEndpointPulls++
|
||||
}
|
||||
}
|
||||
assert.GreaterOrEqual(t, newEndpointPulls, 1,
|
||||
"expected pulls against the recreated subscription endpoint")
|
||||
}
|
||||
|
||||
func TestStream_BackoffWhenRecreateFails(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(2 * time.Second)
|
||||
var calls atomic.Int32
|
||||
for time.Now().Before(deadline) {
|
||||
fc.mu.Lock()
|
||||
calls.Store(int32(len(fc.callMethodCalls)))
|
||||
fc.mu.Unlock()
|
||||
if calls.Load() >= 4 {
|
||||
break
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
assert.GreaterOrEqual(t, calls.Load(), int32(4),
|
||||
"expected stream to retry recreate (>=3 retries on top of the initial create)")
|
||||
}
|
||||
|
||||
func TestStream_ReconnectAfterFailuresDefault(t *testing.T) {
|
||||
o := defaultOptions()
|
||||
assert.Equal(t, 3, o.ReconnectAfterFailures)
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
58
event/stream/renew.go
Normal file
58
event/stream/renew.go
Normal file
@@ -0,0 +1,58 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/xml"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/onvif/event"
|
||||
"github.com/kerberos-io/onvif/xsd"
|
||||
)
|
||||
|
||||
// renewLoop surfaces renew failures and continues. A permanently
|
||||
// failing renew lets the subscription die at the camera; the pull
|
||||
// loop's reconnect path then recreates it — recreate is the only
|
||||
// reliable recovery once a subscription is GC'd.
|
||||
func (s *Stream) renewLoop(ctx context.Context) {
|
||||
interval := s.opts.InitialTermination - s.opts.RenewMargin
|
||||
if interval <= 0 {
|
||||
// Pathological config (margin >= termination): renew at
|
||||
// half termination so we still refresh.
|
||||
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 sends Renew with an absolute UTC TerminationTime.
|
||||
// WS-BaseNotification §6.1.1 also allows xsd:duration but older
|
||||
// Hikvision, some Dahua and some Bosch firmwares reject the
|
||||
// relative form.
|
||||
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
|
||||
}
|
||||
170
event/stream/renew_test.go
Normal file
170
event/stream/renew_test.go
Normal file
@@ -0,0 +1,170 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// countSendSoapMatching counts how many recorded SendSoap calls have a
|
||||
// body containing needle. Safe to call concurrently with the run loop.
|
||||
func countSendSoapMatching(fc *fakeCaller, needle string) int {
|
||||
fc.mu.Lock()
|
||||
defer fc.mu.Unlock()
|
||||
n := 0
|
||||
for _, c := range fc.sendSoapCalls {
|
||||
if strings.Contains(c[1], needle) {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func TestStream_RenewsSubscriptionBeforeExpiry(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
// 100 ms termination with 10 ms margin -> renew every ~90 ms.
|
||||
s, err := newStream(ctx, fc, Options{
|
||||
DeviceID: "cam-1",
|
||||
InitialTermination: 100 * time.Millisecond,
|
||||
RenewMargin: 10 * time.Millisecond,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
defer s.Close()
|
||||
|
||||
deadline := time.Now().Add(500 * time.Millisecond)
|
||||
var renewCount int
|
||||
for time.Now().Before(deadline) {
|
||||
renewCount = countSendSoapMatching(fc, "Renew")
|
||||
if renewCount >= 1 {
|
||||
break
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
assert.GreaterOrEqual(t, renewCount, 1, "expected at least one Renew SendSoap call within 500ms")
|
||||
}
|
||||
|
||||
func TestStream_RenewSendsToSubscriptionEndpoint(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
s, err := newStream(ctx, fc, Options{
|
||||
InitialTermination: 80 * time.Millisecond,
|
||||
RenewMargin: 10 * 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 renewEndpoint string
|
||||
for _, c := range fc.sendSoapCalls {
|
||||
if strings.Contains(c[1], "Renew") {
|
||||
renewEndpoint = c[0]
|
||||
break
|
||||
}
|
||||
}
|
||||
require.NotEmpty(t, renewEndpoint, "no Renew call found")
|
||||
assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", renewEndpoint,
|
||||
"Renew must target the SubscriptionReference Address")
|
||||
}
|
||||
|
||||
func TestStream_RenewMarginAppliesDefault(t *testing.T) {
|
||||
o := defaultOptions()
|
||||
assert.Equal(t, 10*time.Second, o.RenewMargin)
|
||||
}
|
||||
|
||||
func TestStream_RenewErrorSurfacedOnErrorsChannel(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
// Defaults return empty pulls indefinitely so the pull loop is clean.
|
||||
// Override defaultSendSoap on the fly to return a Renew error for
|
||||
// any body that looks like a Renew. We do that by tagging the
|
||||
// default response with an err, then resetting after capturing one.
|
||||
// Simpler: just queue several explicit Renew-error responses; the
|
||||
// fake's queue is consumed in FIFO and the pull body never matches
|
||||
// 'Renew', so queued errors will land on the renew call only if
|
||||
// queued before any pulls. To bias the order we drain via a custom
|
||||
// default.
|
||||
fc.mu.Lock()
|
||||
fc.defaultSendSoap = fakeResp{err: errInjected{}}
|
||||
fc.mu.Unlock()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
s, err := newStream(ctx, fc, Options{
|
||||
InitialTermination: 80 * time.Millisecond,
|
||||
RenewMargin: 10 * time.Millisecond,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
defer s.Close()
|
||||
|
||||
select {
|
||||
case e := <-s.Errors():
|
||||
assert.Contains(t, e.Error(), "injected")
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("expected an error on Errors channel from failing Renew/pull")
|
||||
}
|
||||
}
|
||||
|
||||
// errInjected is a sentinel error type so the test message has a stable
|
||||
// substring without depending on a wrapped string match.
|
||||
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")
|
||||
}
|
||||
178
event/stream/soap.go
Normal file
178
event/stream/soap.go
Normal file
@@ -0,0 +1,178 @@
|
||||
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 SOAP response buffering. 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
|
||||
|
||||
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 returns an empty slice (no 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 is best-effort. Empty endpoint is a no-op
|
||||
// (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
|
||||
}
|
||||
|
||||
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 are wrapped in an
|
||||
// envelope with many namespace prefixes; keying on local name only
|
||||
// sidesteps namespace matching.
|
||||
//
|
||||
// When the camera returns a SOAP Fault, the fault reason is returned
|
||||
// as the error so callers can distinguish auth / expired-subscription
|
||||
// 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: <faultstring>reason</faultstring>
|
||||
soap11FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?faultstring[^>]*>(.*?)</(?:[^:>\s]+:)?faultstring>`)
|
||||
// SOAP 1.2: <Fault>...<Reason><Text>reason</Text></Reason>...</Fault>
|
||||
soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)</(?:[^:>\s]+:)?Text>`)
|
||||
)
|
||||
|
||||
// extractSOAPFault returns the reason text from a SOAP fault or empty
|
||||
// when the body is not a fault. Handles 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 duration as xsd:duration PTnS. Second
|
||||
// precision is sufficient — ONVIF cameras do not honour sub-second
|
||||
// pull timeouts.
|
||||
func durationToXSD(d time.Duration) string {
|
||||
secs := int(d.Round(time.Second).Seconds())
|
||||
if secs <= 0 {
|
||||
secs = 1
|
||||
}
|
||||
return "PT" + strconv.Itoa(secs) + "S"
|
||||
}
|
||||
86
event/stream/soap_test.go
Normal file
86
event/stream/soap_test.go
Normal file
@@ -0,0 +1,86 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// --- SOAP fault detection ---------------------------------------------
|
||||
|
||||
func TestExtractSOAPFault_SOAP11(t *testing.T) {
|
||||
body := `<?xml version="1.0"?>
|
||||
<env:Envelope xmlns:env="http://schemas.xmlsoap.org/soap/envelope/">
|
||||
<env:Body>
|
||||
<env:Fault>
|
||||
<faultcode>env:Client</faultcode>
|
||||
<faultstring>The action requested requires authorization and the sender is not authorized</faultstring>
|
||||
</env:Fault>
|
||||
</env:Body>
|
||||
</env:Envelope>`
|
||||
got := extractSOAPFault(body)
|
||||
assert.Contains(t, got, "not authorized")
|
||||
}
|
||||
|
||||
func TestExtractSOAPFault_SOAP12(t *testing.T) {
|
||||
body := `<?xml version="1.0"?>
|
||||
<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope">
|
||||
<env:Body>
|
||||
<env:Fault>
|
||||
<env:Code><env:Value>env:Sender</env:Value></env:Code>
|
||||
<env:Reason><env:Text xml:lang="en">Subscription has expired</env:Text></env:Reason>
|
||||
</env:Fault>
|
||||
</env:Body>
|
||||
</env:Envelope>`
|
||||
got := extractSOAPFault(body)
|
||||
assert.Contains(t, got, "Subscription has expired")
|
||||
}
|
||||
|
||||
func TestExtractSOAPFault_NotAFault(t *testing.T) {
|
||||
assert.Empty(t, extractSOAPFault(createPullPointResp))
|
||||
}
|
||||
|
||||
func TestExtractSOAPFault_EmptyBody(t *testing.T) {
|
||||
assert.Empty(t, extractSOAPFault(""))
|
||||
}
|
||||
|
||||
func TestUnmarshalNode_ReturnsFaultReasonInsteadOfMissingElement(t *testing.T) {
|
||||
body := `<env:Envelope xmlns:env="http://schemas.xmlsoap.org/soap/envelope/">
|
||||
<env:Body><env:Fault><faultstring>not authorized</faultstring></env:Fault></env:Body>
|
||||
</env:Envelope>`
|
||||
var out struct{}
|
||||
err := unmarshalNode(body, "PullMessagesResponse", &out)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "not authorized")
|
||||
assert.NotContains(t, err.Error(), "missing PullMessagesResponse")
|
||||
}
|
||||
|
||||
// --- Bounded body read -----------------------------------------------
|
||||
|
||||
func TestReadClose_LimitsBodySize(t *testing.T) {
|
||||
if maxResponseBytes < 1024 {
|
||||
t.Skip("limit too small for this test")
|
||||
}
|
||||
big := strings.Repeat("A", maxResponseBytes+1024)
|
||||
body := "<env:Envelope><env:Body>" + big + "</env:Body></env:Envelope>"
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(body, nil)
|
||||
// Construction will fail because the truncated body has no
|
||||
// CreatePullPointSubscriptionResponse — that's fine; what matters is
|
||||
// the read completes without OOM.
|
||||
_, err := newStream(testContext(t), fc, Options{})
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
// 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())
|
||||
t.Cleanup(cancel)
|
||||
return ctx
|
||||
}
|
||||
284
event/stream/stream.go
Normal file
284
event/stream/stream.go
Normal file
@@ -0,0 +1,284 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/onvif"
|
||||
)
|
||||
|
||||
// closeDrainTimeout bounds Close's wait for the pull and renew
|
||||
// goroutines to exit. The loops block in caller.SendSoap which is not
|
||||
// ctx-aware (the underlying http.Client is the only thing that can
|
||||
// unblock them — see caller below). On a hung HTTP transport Close
|
||||
// would otherwise wait forever; instead it returns an error and lets
|
||||
// the calling agent move on.
|
||||
const closeDrainTimeout = 5 * time.Second
|
||||
|
||||
// closeUnsubscribeTimeout bounds the Unsubscribe SOAP call issued by
|
||||
// Close. A subscription expires at the camera once InitialTermination
|
||||
// elapses without a renew, so a missed unsubscribe is at worst
|
||||
// cosmetic.
|
||||
const closeUnsubscribeTimeout = 5 * time.Second
|
||||
|
||||
// Options configures a Stream.
|
||||
//
|
||||
// Zero-value policy: every duration / int field treats zero as "use
|
||||
// the default". To opt out of reconnect set DisableReconnect=true
|
||||
// (ReconnectAfterFailures=0 would otherwise collide with the default
|
||||
// injection). For unbuffered Events / Errors channels set
|
||||
// BufferSize=-1.
|
||||
type Options struct {
|
||||
DeviceID string
|
||||
// RawTopicFilter is the ONVIF ConcreteSet TopicExpression filter
|
||||
// passed verbatim to CreatePullPointSubscription. Callers should
|
||||
// normally leave this empty and rely on Classify for routing —
|
||||
// server-side filtering is fragile across vendors and empty is
|
||||
// required for AXIS.
|
||||
RawTopicFilter string
|
||||
// PullTimeout — zero means default (5s).
|
||||
PullTimeout time.Duration
|
||||
// MessageLimit — zero means default (32). Busy AXIS cameras with
|
||||
// many configured rules can burst beyond 10 per pull.
|
||||
MessageLimit int
|
||||
// InitialTermination — zero means default (60s).
|
||||
InitialTermination time.Duration
|
||||
// RenewMargin — larger margins tolerate slower networks at the
|
||||
// cost of more renew calls. Zero means default (10s).
|
||||
RenewMargin time.Duration
|
||||
// ReconnectAfterFailures — pull-points die for many reasons
|
||||
// (camera reboot, subscription GC after a renew miss, NAT
|
||||
// timeout); rebuilding the subscription is the only reliable
|
||||
// recovery. Zero means default (3). Set DisableReconnect=true
|
||||
// to disable.
|
||||
ReconnectAfterFailures int
|
||||
// DisableReconnect makes the pull loop retry against the
|
||||
// original endpoint until ctx is cancelled.
|
||||
DisableReconnect bool
|
||||
// RetryBackoff is the base sleep between pull/recreate failures.
|
||||
// Recreate failures double this up to maxRecreateBackoff. Zero
|
||||
// means default (1s).
|
||||
RetryBackoff time.Duration
|
||||
// BufferSize — zero means default (16); use -1 for unbuffered.
|
||||
BufferSize int
|
||||
}
|
||||
|
||||
func defaultOptions() Options {
|
||||
return Options{
|
||||
PullTimeout: 5 * time.Second,
|
||||
MessageLimit: 32,
|
||||
InitialTermination: 60 * time.Second,
|
||||
RenewMargin: 10 * time.Second,
|
||||
ReconnectAfterFailures: 3,
|
||||
RetryBackoff: time.Second,
|
||||
BufferSize: 16,
|
||||
}
|
||||
}
|
||||
|
||||
func (o Options) withDefaults() Options {
|
||||
d := defaultOptions()
|
||||
if o.PullTimeout > 0 {
|
||||
d.PullTimeout = o.PullTimeout
|
||||
}
|
||||
if o.MessageLimit > 0 {
|
||||
d.MessageLimit = o.MessageLimit
|
||||
}
|
||||
if o.InitialTermination > 0 {
|
||||
d.InitialTermination = o.InitialTermination
|
||||
}
|
||||
if o.RenewMargin > 0 {
|
||||
d.RenewMargin = o.RenewMargin
|
||||
}
|
||||
if o.ReconnectAfterFailures > 0 {
|
||||
d.ReconnectAfterFailures = o.ReconnectAfterFailures
|
||||
}
|
||||
if o.RetryBackoff > 0 {
|
||||
d.RetryBackoff = o.RetryBackoff
|
||||
}
|
||||
switch {
|
||||
case o.BufferSize > 0:
|
||||
d.BufferSize = o.BufferSize
|
||||
case o.BufferSize < 0:
|
||||
d.BufferSize = 0
|
||||
}
|
||||
d.DeviceID = o.DeviceID
|
||||
d.RawTopicFilter = o.RawTopicFilter
|
||||
d.DisableReconnect = o.DisableReconnect
|
||||
return d
|
||||
}
|
||||
|
||||
// caller is the *onvif.Device subset Stream depends on. Implementations
|
||||
// must:
|
||||
//
|
||||
// - Be safe for concurrent use — pull and renew goroutines call in
|
||||
// from separate goroutines. *onvif.Device satisfies this via
|
||||
// http.Client.
|
||||
// - Enforce a per-request timeout via the underlying HTTP client.
|
||||
// The methods do not take a ctx, so ctx-cancel cannot interrupt a
|
||||
// hung request; only the HTTP client's own timeout can. Close
|
||||
// bounds its drain wait at closeDrainTimeout to survive a misbehaving
|
||||
// caller, but a leaking goroutine remains until the HTTP call
|
||||
// eventually returns.
|
||||
type caller interface {
|
||||
CallMethod(method any) (*http.Response, error)
|
||||
SendSoap(endpoint, body string) (*http.Response, error)
|
||||
}
|
||||
|
||||
type deviceCaller struct{ dev *onvif.Device }
|
||||
|
||||
func (d deviceCaller) CallMethod(m any) (*http.Response, error) {
|
||||
return d.dev.CallMethod(m)
|
||||
}
|
||||
|
||||
func (d deviceCaller) SendSoap(endpoint, body string) (*http.Response, error) {
|
||||
return d.dev.SendSoap(endpoint, body)
|
||||
}
|
||||
|
||||
// Stream owns a single ONVIF pull-point subscription. Safe for Close
|
||||
// from any goroutine while readers consume Events / Errors. Close is
|
||||
// idempotent.
|
||||
type Stream struct {
|
||||
caller caller
|
||||
opts Options
|
||||
|
||||
pullPointMu sync.Mutex
|
||||
pullPoint string
|
||||
|
||||
events chan Event
|
||||
errors chan error
|
||||
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
|
||||
closeOnce sync.Once
|
||||
closeErr error
|
||||
|
||||
// now is overridable so tests can make timestamps deterministic.
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func (s *Stream) getPullPoint() string {
|
||||
s.pullPointMu.Lock()
|
||||
defer s.pullPointMu.Unlock()
|
||||
return s.pullPoint
|
||||
}
|
||||
|
||||
func (s *Stream) setPullPoint(addr string) {
|
||||
s.pullPointMu.Lock()
|
||||
defer s.pullPointMu.Unlock()
|
||||
s.pullPoint = addr
|
||||
}
|
||||
|
||||
// NewStream creates a Stream and performs CreatePullPointSubscription
|
||||
// synchronously so connectivity and authentication failures surface
|
||||
// from NewStream rather than landing on Errors later.
|
||||
//
|
||||
// The returned Stream stops when ctx is cancelled or Close is called.
|
||||
func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, error) {
|
||||
return newStream(ctx, deviceCaller{dev: dev}, opts)
|
||||
}
|
||||
|
||||
func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) {
|
||||
opts = opts.withDefaults()
|
||||
addr, err := createPullPoint(c, opts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create pull point subscription: %w", err)
|
||||
}
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
s := &Stream{
|
||||
caller: c,
|
||||
opts: opts,
|
||||
pullPoint: addr,
|
||||
events: make(chan Event, opts.BufferSize),
|
||||
errors: make(chan error, opts.BufferSize),
|
||||
cancel: cancel,
|
||||
done: make(chan struct{}),
|
||||
now: time.Now,
|
||||
}
|
||||
go s.run(runCtx)
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Events returns the channel of decoded notifications. Closed when
|
||||
// the Stream stops.
|
||||
func (s *Stream) Events() <-chan Event { return s.events }
|
||||
|
||||
// Errors returns the channel of non-fatal errors. Sends are
|
||||
// non-blocking; consumers that fall behind drop older errors. Closed
|
||||
// when the Stream stops.
|
||||
func (s *Stream) Errors() <-chan error { return s.errors }
|
||||
|
||||
// Close stops the background goroutines, waits up to closeDrainTimeout
|
||||
// for them to exit, and then Unsubscribes from the camera (also bounded,
|
||||
// by closeUnsubscribeTimeout). Subsequent calls are no-ops.
|
||||
//
|
||||
// If the drain times out the goroutines are likely wedged inside a
|
||||
// non-ctx-aware caller.SendSoap; they will exit on their own once the
|
||||
// HTTP call returns. Unsubscribe is skipped in that case — the
|
||||
// subscription expires at the camera anyway.
|
||||
func (s *Stream) Close() error {
|
||||
s.closeOnce.Do(func() {
|
||||
s.cancel()
|
||||
|
||||
select {
|
||||
case <-s.done:
|
||||
case <-time.After(closeDrainTimeout):
|
||||
s.closeErr = fmt.Errorf("close: pull/renew loops did not drain within %s (likely stuck in caller HTTP)", closeDrainTimeout)
|
||||
return
|
||||
}
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
errCh <- unsubscribePullPoint(s.caller, s.getPullPoint())
|
||||
}()
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if err != nil {
|
||||
s.closeErr = fmt.Errorf("unsubscribe pull point: %w", err)
|
||||
}
|
||||
case <-time.After(closeUnsubscribeTimeout):
|
||||
s.closeErr = fmt.Errorf("unsubscribe pull point: timeout after %s", closeUnsubscribeTimeout)
|
||||
}
|
||||
})
|
||||
return s.closeErr
|
||||
}
|
||||
|
||||
func (s *Stream) run(ctx context.Context) {
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
s.renewLoop(ctx)
|
||||
}()
|
||||
s.pullLoop(ctx)
|
||||
wg.Wait()
|
||||
|
||||
// Explicit close order after both goroutines have exited so a
|
||||
// future maintainer extending this function does not rely on
|
||||
// defer-ordering for channel-close safety.
|
||||
close(s.errors)
|
||||
close(s.events)
|
||||
close(s.done)
|
||||
}
|
||||
|
||||
func (s *Stream) surfaceError(err error) {
|
||||
select {
|
||||
case s.errors <- err:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// sleepCtx returns false if ctx was cancelled, true if d elapsed.
|
||||
func sleepCtx(ctx context.Context, d time.Duration) bool {
|
||||
t := time.NewTimer(d)
|
||||
defer t.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-t.C:
|
||||
return true
|
||||
}
|
||||
}
|
||||
497
event/stream/stream_test.go
Normal file
497
event/stream/stream_test.go
Normal file
@@ -0,0 +1,497 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// --- fakeCaller --------------------------------------------------------
|
||||
|
||||
// fakeCaller is a test double for the caller interface. Each method
|
||||
// returns the next queued response; when the queue is exhausted it falls
|
||||
// back to a default response so the indefinite pull loop does not
|
||||
// require tests to enumerate every call.
|
||||
//
|
||||
// blockUnsubscribe, when non-nil, causes SendSoap calls whose body
|
||||
// contains "Unsubscribe" to block until the channel is closed.
|
||||
// blockAllSendSoap, when non-nil, blocks every SendSoap call until
|
||||
// closed (simulates a hung HTTP transport).
|
||||
type fakeCaller struct {
|
||||
mu sync.Mutex
|
||||
callMethodResps []fakeResp
|
||||
sendSoapResps []fakeResp
|
||||
defaultSendSoap fakeResp
|
||||
defaultCall fakeResp
|
||||
callMethodCalls []any
|
||||
sendSoapCalls [][2]string
|
||||
blockUnsubscribe chan struct{}
|
||||
blockAllSendSoap chan struct{}
|
||||
}
|
||||
|
||||
type fakeResp struct {
|
||||
body string
|
||||
err error
|
||||
}
|
||||
|
||||
func newFakeCaller() *fakeCaller {
|
||||
return &fakeCaller{
|
||||
// Default: indefinite empty pulls, indefinite OK unsubscribes.
|
||||
defaultSendSoap: fakeResp{body: pullMessagesResp()},
|
||||
defaultCall: fakeResp{err: errors.New("fakeCaller: no default CallMethod response")},
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeCaller) queueCallMethod(body string, err error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.callMethodResps = append(f.callMethodResps, fakeResp{body: body, err: err})
|
||||
}
|
||||
|
||||
func (f *fakeCaller) queueSendSoap(body string, err error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.sendSoapResps = append(f.sendSoapResps, fakeResp{body: body, err: err})
|
||||
}
|
||||
|
||||
func (f *fakeCaller) CallMethod(m any) (*http.Response, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.callMethodCalls = append(f.callMethodCalls, m)
|
||||
r := f.defaultCall
|
||||
if len(f.callMethodResps) > 0 {
|
||||
r = f.callMethodResps[0]
|
||||
f.callMethodResps = f.callMethodResps[1:]
|
||||
}
|
||||
if r.err != nil {
|
||||
return nil, r.err
|
||||
}
|
||||
return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil
|
||||
}
|
||||
|
||||
func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) {
|
||||
f.mu.Lock()
|
||||
f.sendSoapCalls = append(f.sendSoapCalls, [2]string{endpoint, body})
|
||||
r := f.defaultSendSoap
|
||||
if len(f.sendSoapResps) > 0 {
|
||||
r = f.sendSoapResps[0]
|
||||
f.sendSoapResps = f.sendSoapResps[1:]
|
||||
}
|
||||
block := f.blockUnsubscribe
|
||||
blockAll := f.blockAllSendSoap
|
||||
f.mu.Unlock()
|
||||
|
||||
if blockAll != nil {
|
||||
<-blockAll
|
||||
}
|
||||
if block != nil && strings.Contains(body, "Unsubscribe") {
|
||||
<-block
|
||||
}
|
||||
|
||||
if r.err != nil {
|
||||
return nil, r.err
|
||||
}
|
||||
return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil
|
||||
}
|
||||
|
||||
func (f *fakeCaller) sendSoapCallCount() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.sendSoapCalls)
|
||||
}
|
||||
|
||||
// --- fixture SOAP envelopes -------------------------------------------
|
||||
|
||||
// createPullPointResp is the minimal SOAP envelope the lib's existing
|
||||
// xml.Decoder + getXMLNode path can extract a pull-point address from.
|
||||
const createPullPointResp = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
|
||||
xmlns:wsa="http://www.w3.org/2005/08/addressing"
|
||||
xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
|
||||
<env:Body>
|
||||
<tev:CreatePullPointSubscriptionResponse>
|
||||
<tev:SubscriptionReference>
|
||||
<wsa:Address>http://camera.local/onvif/Events/PullSub_1</wsa:Address>
|
||||
</tev:SubscriptionReference>
|
||||
<tev:CurrentTime>2026-05-21T10:30:00Z</tev:CurrentTime>
|
||||
<tev:TerminationTime>2026-05-21T10:31:00Z</tev:TerminationTime>
|
||||
</tev:CreatePullPointSubscriptionResponse>
|
||||
</env:Body>
|
||||
</env:Envelope>`
|
||||
|
||||
func pullMessagesResp(messages ...string) string {
|
||||
return `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
|
||||
xmlns:tev="http://www.onvif.org/ver10/events/wsdl"
|
||||
xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2"
|
||||
xmlns:tt="http://www.onvif.org/ver10/schema">
|
||||
<env:Body>
|
||||
<tev:PullMessagesResponse>
|
||||
<tev:CurrentTime>2026-05-21T10:30:05Z</tev:CurrentTime>
|
||||
<tev:TerminationTime>2026-05-21T10:31:05Z</tev:TerminationTime>
|
||||
` + strings.Join(messages, "\n") + `
|
||||
</tev:PullMessagesResponse>
|
||||
</env:Body>
|
||||
</env:Envelope>`
|
||||
}
|
||||
|
||||
func motionMsg(value string) string {
|
||||
return `<wsnt:NotificationMessage>
|
||||
<wsnt:Topic Dialect="http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet">tns1:RuleEngine/CellMotionDetector/Motion</wsnt:Topic>
|
||||
<wsnt:Message>
|
||||
<tt:Message PropertyOperation="Changed" UtcTime="2026-05-21T10:30:00Z">
|
||||
<tt:Source>
|
||||
<tt:SimpleItem Name="VideoSourceConfigurationToken" Value="VSC0"/>
|
||||
</tt:Source>
|
||||
<tt:Data>
|
||||
<tt:SimpleItem Name="IsMotion" Value="` + value + `"/>
|
||||
</tt:Data>
|
||||
</tt:Message>
|
||||
</wsnt:Message>
|
||||
</wsnt:NotificationMessage>`
|
||||
}
|
||||
|
||||
const unsubscribeResp = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
|
||||
xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2">
|
||||
<env:Body>
|
||||
<wsnt:UnsubscribeResponse/>
|
||||
</env:Body>
|
||||
</env:Envelope>`
|
||||
|
||||
// --- helpers -----------------------------------------------------------
|
||||
|
||||
// receive waits up to d for an event on ch, failing the test if none
|
||||
// arrives.
|
||||
func receive(t *testing.T, ch <-chan Event, d time.Duration) Event {
|
||||
t.Helper()
|
||||
select {
|
||||
case ev, ok := <-ch:
|
||||
if !ok {
|
||||
t.Fatalf("event channel closed before receiving")
|
||||
}
|
||||
return ev
|
||||
case <-time.After(d):
|
||||
t.Fatalf("timed out waiting for event after %s", d)
|
||||
}
|
||||
return Event{} // unreachable
|
||||
}
|
||||
|
||||
// --- tests -------------------------------------------------------------
|
||||
|
||||
func TestNewStream_CreatesPullPointAtConstruction(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
// Queue an empty pull so the run loop can spin without exploding.
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
fc.queueSendSoap(unsubscribeResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, s)
|
||||
require.NoError(t, s.Close())
|
||||
|
||||
// CreatePullPointSubscription was called exactly once.
|
||||
fc.mu.Lock()
|
||||
defer fc.mu.Unlock()
|
||||
require.Len(t, fc.callMethodCalls, 1, "expected one CallMethod call (CreatePullPointSubscription)")
|
||||
}
|
||||
|
||||
func TestStream_DeliversDecodedEvents(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
|
||||
// Provide subsequent empty pulls so the loop doesn't starve before Close.
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
fc.queueSendSoap(unsubscribeResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
|
||||
require.NoError(t, err)
|
||||
defer s.Close()
|
||||
|
||||
ev := receive(t, s.Events(), 2*time.Second)
|
||||
assert.Equal(t, KindMotion, ev.Kind)
|
||||
assert.Equal(t, StateActive, ev.State)
|
||||
assert.Equal(t, "cam-1", ev.DeviceID)
|
||||
assert.Equal(t, "tns1:RuleEngine/CellMotionDetector/Motion", ev.Topic)
|
||||
assert.Equal(t, "VSC0", ev.Source["VideoSourceConfigurationToken"])
|
||||
assert.Equal(t, "true", ev.Data["IsMotion"])
|
||||
}
|
||||
|
||||
func TestStream_PullsAgainstSubscriptionAddress(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
fc.queueSendSoap(unsubscribeResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Wait until at least one pull happened, then close.
|
||||
for i := 0; i < 50 && fc.sendSoapCallCount() == 0; i++ {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
require.NoError(t, s.Close())
|
||||
|
||||
fc.mu.Lock()
|
||||
defer fc.mu.Unlock()
|
||||
require.NotEmpty(t, fc.sendSoapCalls, "expected at least one PullMessages SendSoap call")
|
||||
endpoint := fc.sendSoapCalls[0][0]
|
||||
assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", endpoint,
|
||||
"PullMessages must target the SubscriptionReference Address returned by CreatePullPoint")
|
||||
// Last call (Close) should target the same endpoint with an Unsubscribe body.
|
||||
last := fc.sendSoapCalls[len(fc.sendSoapCalls)-1]
|
||||
assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", last[0])
|
||||
assert.Contains(t, last[1], "Unsubscribe")
|
||||
}
|
||||
|
||||
func TestNewStream_ReturnsErrorWhenCreatePullPointFails(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod("", errors.New("network down"))
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
|
||||
assert.Error(t, err)
|
||||
assert.Nil(t, s)
|
||||
}
|
||||
|
||||
func TestStream_ClosedContextStopsRunLoop(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
// Many empty pulls so the loop is hot when we cancel.
|
||||
for i := 0; i < 20; i++ {
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
}
|
||||
fc.queueSendSoap(unsubscribeResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Wait for at least one pull.
|
||||
for i := 0; i < 50 && fc.sendSoapCallCount() == 0; i++ {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
cancel()
|
||||
|
||||
// Close should still complete cleanly; the goroutine must drain.
|
||||
require.NoError(t, s.Close())
|
||||
|
||||
// Events channel must close so consumers can range-loop safely.
|
||||
select {
|
||||
case _, ok := <-s.Events():
|
||||
assert.False(t, ok, "Events channel should be closed after Close()")
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Events channel was not closed within 1s")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_PullErrorSurfacedOnErrorsChannel(t *testing.T) {
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
fc.queueSendSoap("", errors.New("transient pull failure"))
|
||||
// Then a clean pull so the loop keeps running.
|
||||
fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
fc.queueSendSoap(unsubscribeResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
|
||||
require.NoError(t, err)
|
||||
defer s.Close()
|
||||
|
||||
select {
|
||||
case e := <-s.Errors():
|
||||
assert.Contains(t, e.Error(), "transient pull failure")
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("expected an error on the Errors channel")
|
||||
}
|
||||
// After the transient failure the loop continued and decoded.
|
||||
ev := receive(t, s.Events(), 2*time.Second)
|
||||
assert.Equal(t, KindMotion, ev.Kind)
|
||||
}
|
||||
|
||||
func TestStream_OptionsApplyDefaults(t *testing.T) {
|
||||
o := defaultOptions()
|
||||
assert.Equal(t, 5*time.Second, o.PullTimeout)
|
||||
assert.Equal(t, 32, o.MessageLimit)
|
||||
assert.Equal(t, 60*time.Second, o.InitialTermination)
|
||||
assert.Equal(t, 16, o.BufferSize)
|
||||
}
|
||||
|
||||
func TestStream_DoesNotPanicOnPullExitingDuringClose(t *testing.T) {
|
||||
// Regression guard: Close should not race with the run goroutine
|
||||
// in a way that double-closes the events/errors channels.
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
for i := 0; i < 5; i++ {
|
||||
fc.queueSendSoap(pullMessagesResp(), nil)
|
||||
}
|
||||
fc.queueSendSoap(unsubscribeResp, nil)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
|
||||
require.NoError(t, err)
|
||||
assert.NotPanics(t, func() {
|
||||
require.NoError(t, s.Close())
|
||||
// Double-close should be a no-op, not a panic.
|
||||
_ = s.Close()
|
||||
})
|
||||
}
|
||||
|
||||
// --- 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")
|
||||
}
|
||||
|
||||
func TestClose_BoundedWhenLoopsStuckOnHungHTTP(t *testing.T) {
|
||||
// Simulates a hung HTTP transport: every SendSoap blocks
|
||||
// indefinitely. The pull and renew loops are wedged inside
|
||||
// SendSoap and ctx-cancel cannot unblock them. Close must still
|
||||
// return within its bounded budget so the agent's shutdown does
|
||||
// not hang.
|
||||
fc := newFakeCaller()
|
||||
fc.queueCallMethod(createPullPointResp, nil)
|
||||
blockAll := make(chan struct{})
|
||||
defer close(blockAll)
|
||||
fc.mu.Lock()
|
||||
fc.blockAllSendSoap = blockAll
|
||||
fc.mu.Unlock()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s, err := newStream(ctx, fc, Options{
|
||||
PullTimeout: 100 * time.Millisecond,
|
||||
InitialTermination: 30 * time.Second,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Wait until pullLoop is actually parked inside the blocked
|
||||
// SendSoap. Without this, Close races with the loop's first
|
||||
// iteration and exits via the ctx pre-check instead of
|
||||
// exercising the drain-timeout path.
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for time.Now().Before(deadline) && fc.sendSoapCallCount() == 0 {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
require.GreaterOrEqual(t, fc.sendSoapCallCount(), 1, "pullLoop never reached SendSoap")
|
||||
|
||||
start := time.Now()
|
||||
err = s.Close()
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "drain", "expected a drain-timeout error")
|
||||
// Total budget is closeDrainTimeout for the wait + ~0 for unsubscribe
|
||||
// (which is skipped when drain times out). Give plenty of slack for
|
||||
// scheduling on a loaded CI machine.
|
||||
assert.Less(t, elapsed, closeDrainTimeout+2*time.Second,
|
||||
"Close exceeded bound (%s); expected ~%s", elapsed, closeDrainTimeout)
|
||||
}
|
||||
143
event/stream/topics.go
Normal file
143
event/stream/topics.go
Normal file
@@ -0,0 +1,143 @@
|
||||
package stream
|
||||
|
||||
import "strings"
|
||||
|
||||
// Classify maps an ONVIF topic string to the normalized Kind. Returns
|
||||
// KindUnknown when no rule matches.
|
||||
//
|
||||
// The classifier strips XML-namespace prefixes from each "/"-separated
|
||||
// segment so it is robust to vendor namespaces (tns1:, tnsaxis:,
|
||||
// tnssamsung:, ...). Matching is case-sensitive — ONVIF topics are
|
||||
// case-sensitive per the spec.
|
||||
//
|
||||
// Sources cross-checked when building the rule set below:
|
||||
// - ONVIF Topic Namespace XML
|
||||
// https://www.onvif.org/onvif/ver10/topics/topicns.xml
|
||||
// - ONVIF Analytics Service Spec
|
||||
// https://www.onvif.org/specs/srv/analytics/ONVIF-VideoAnalytics-Service-Spec-v220.pdf
|
||||
// - ONVIF Device IO Service Spec
|
||||
// https://www.onvif.org/specs/srv/io/ONVIF-DeviceIo-Service-Spec.pdf
|
||||
// - openvideolibs/onvif-parsers (Apache-2.0) — empirical topic table
|
||||
// extracted from Home Assistant
|
||||
// https://github.com/openvideolibs/onvif-parsers
|
||||
func Classify(topic string) Kind {
|
||||
if topic == "" {
|
||||
return KindUnknown
|
||||
}
|
||||
canonical := canonicalizeTopic(topic)
|
||||
for _, rule := range topicRules {
|
||||
if strings.Contains(canonical, rule.needle) {
|
||||
return rule.kind
|
||||
}
|
||||
}
|
||||
return KindUnknown
|
||||
}
|
||||
|
||||
// canonicalizeTopic strips the XML-namespace prefix from each
|
||||
// "/"-separated segment, collapsing Avigilon's per-segment-prefixed
|
||||
// form ("tns1:Device/tns1:Trigger/tns1:Relay") and the plain form
|
||||
// ("tns1:Device/Trigger/Relay") to the same matchable path.
|
||||
func canonicalizeTopic(topic string) string {
|
||||
segments := strings.Split(topic, "/")
|
||||
for i, seg := range segments {
|
||||
if idx := strings.Index(seg, ":"); idx >= 0 {
|
||||
segments[i] = seg[idx+1:]
|
||||
}
|
||||
}
|
||||
return strings.Join(segments, "/")
|
||||
}
|
||||
|
||||
// topicRules is evaluated in order — first match wins. Keep more
|
||||
// specific rules ahead of broader ones. LineDetector/Crossed is
|
||||
// edge-triggered (no boolean State); the decoder leaves State as
|
||||
// StateUnknown for it.
|
||||
var topicRules = []struct {
|
||||
needle string
|
||||
kind Kind
|
||||
}{
|
||||
// tns1:VideoSource/MotionAlarm — Profile S basic motion.
|
||||
// https://www.onvif.org/ver10/topics/topicns.xml
|
||||
// https://developer.axis.com/vapix/network-video/event-and-action-services/
|
||||
{"VideoSource/MotionAlarm", KindMotion},
|
||||
|
||||
// tns1:VideoAnalytics/MotionAlarm — Bosch publishes motion under
|
||||
// VideoAnalytics rather than VideoSource.
|
||||
// https://media.boschsecurity.com/fs/media/pb/media/partners_1/integration_tools_1/developer/bosch-metadata-and-iva-events.pdf
|
||||
{"VideoAnalytics/MotionAlarm", KindMotion},
|
||||
|
||||
// tns1:VideoAnalytics/tnssamsung:MotionDetection — Hanwha vendor.
|
||||
// https://github.com/home-assistant/core/issues/66493
|
||||
{"VideoAnalytics/MotionDetection", KindMotion},
|
||||
|
||||
// tns1:RuleEngine/CellMotionDetector/Motion — ONVIF Analytics
|
||||
// standard cell-motion rule.
|
||||
// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.3
|
||||
// https://www.hikvisioneurope.com/eu/portal/portal/Technical%20Materials/24%20How%20To/CCTV/How%20to%20solve%20third%20party%20camera%20motion%20detection%20issue.pdf
|
||||
{"CellMotionDetector/Motion", KindMotion},
|
||||
|
||||
// tns1:RuleEngine/MotionRegionDetector/Motion — AXIS region rule.
|
||||
// https://developer.axis.com/vapix/network-video/event-and-action-services/
|
||||
{"MotionRegionDetector/Motion", KindMotion},
|
||||
|
||||
// AXIS Guard suite — vendor analytics apps with Camera<N>Profile<ID>
|
||||
// suffixes. Treated as motion so they can drive motion-triggered
|
||||
// recording on cameras using these apps instead of basic VMD.
|
||||
// https://developer.axis.com/vapix/applications/motion-guard
|
||||
{"CameraApplicationPlatform/MotionGuard/", KindMotion},
|
||||
{"CameraApplicationPlatform/FenceGuard/", KindMotion},
|
||||
{"CameraApplicationPlatform/LoiteringGuard/", KindMotion},
|
||||
|
||||
// tns1:RuleEngine/TamperDetector/Tamper — standard ONVIF tamper rule.
|
||||
// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.5
|
||||
{"TamperDetector/Tamper", KindTampering},
|
||||
|
||||
// tns1:VideoSource/GlobalSceneChange/ImagingService — the proper
|
||||
// lens-cover signal on firmwares without TamperDetector.
|
||||
// https://www.onvif.org/ver10/topics/topicns.xml
|
||||
{"GlobalSceneChange", KindTampering},
|
||||
|
||||
// tns1:VideoAnalytics/tnssamsung:TamperingDetection — Hanwha.
|
||||
// https://github.com/home-assistant/core/issues/66493
|
||||
{"VideoAnalytics/TamperingDetection", KindTampering},
|
||||
|
||||
// VideoSource/ImageToo* — imaging-quality alarms. See KindImageQuality
|
||||
// for the rationale on splitting these out from KindTampering.
|
||||
// https://www.onvif.org/ver10/topics/topicns.xml
|
||||
{"VideoSource/ImageTooDark", KindImageQuality},
|
||||
{"VideoSource/ImageTooBright", KindImageQuality},
|
||||
{"VideoSource/ImageTooBlurry", KindImageQuality},
|
||||
|
||||
// tns1:Device/Trigger/DigitalInput — standard. Avigilon's per-segment-
|
||||
// prefixed serialisation ("tns1:Device/tns1:Trigger/tns1:DigitalInput")
|
||||
// folds to the same canonical path.
|
||||
// ONVIF-DeviceIo-Service-Spec.pdf §5.2
|
||||
{"Trigger/DigitalInput", KindDigitalInput},
|
||||
// ONVIF-DeviceIo-Service-Spec.pdf §5.3
|
||||
{"Trigger/Relay", KindDigitalOutput},
|
||||
|
||||
// tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario<N>
|
||||
// — Scenario suffixes are numeric per AOA configuration. Prefix-match
|
||||
// because of the dynamic suffix.
|
||||
// https://developer.axis.com/analytics/axis-object-analytics/how-to-guides/axis-object-analytics-counting-data/
|
||||
{"ObjectAnalytics/", KindObjectDetected},
|
||||
|
||||
// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.4
|
||||
{"LineDetector/Crossed", KindObjectDetected},
|
||||
{"FieldDetector/ObjectsInside", KindObjectDetected},
|
||||
|
||||
// tns1:RuleEngine/MyRuleDetector/<RuleName> — vendor rules under the
|
||||
// ONVIF MyRuleDetector container. Explicitly whitelisted because the
|
||||
// same container also carries non-object rules (Bosch Counter,
|
||||
// Occupancy) that must not classify as ObjectDetected.
|
||||
// https://media.boschsecurity.com/fs/media/pb/media/partners_1/integration_tools_1/developer/bosch-metadata-and-iva-events.pdf
|
||||
{"MyRuleDetector/HumanDetect", KindObjectDetected},
|
||||
{"MyRuleDetector/VehicleDetect", KindObjectDetected},
|
||||
{"MyRuleDetector/PeopleDetect", KindObjectDetected},
|
||||
{"MyRuleDetector/ObjectsInside", KindObjectDetected},
|
||||
{"MyRuleDetector/FaceDetect", KindObjectDetected},
|
||||
|
||||
{"Audio/DetectedSound", KindAudioAlarm},
|
||||
// https://developer.axis.com/vapix/network-video/event-and-action-services/
|
||||
{"AudioSource/TriggerLevel", KindAudioAlarm},
|
||||
{"AudioAnalytics/SoundDetection", KindAudioAlarm},
|
||||
}
|
||||
138
event/stream/topics_test.go
Normal file
138
event/stream/topics_test.go
Normal file
@@ -0,0 +1,138 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestClassifyTopic(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
topic string
|
||||
want Kind
|
||||
}{
|
||||
// --- Motion -----------------------------------------------------
|
||||
|
||||
{"video_source_motion_alarm", "tns1:VideoSource/MotionAlarm", KindMotion},
|
||||
{"cell_motion_detector", "tns1:RuleEngine/CellMotionDetector/Motion", KindMotion},
|
||||
{"motion_region_detector", "tns1:RuleEngine/MotionRegionDetector/Motion", KindMotion},
|
||||
{"bosch_video_analytics_motion", "tns1:VideoAnalytics/MotionAlarm", KindMotion},
|
||||
{"hanwha_samsung_motion", "tns1:VideoAnalytics/tnssamsung:MotionDetection", KindMotion},
|
||||
|
||||
// AXIS Guard suite — vendor analytics apps.
|
||||
{"axis_motion_guard", "tnsaxis:CameraApplicationPlatform/MotionGuard/Camera1ProfileANY", KindMotion},
|
||||
{"axis_fence_guard", "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1ProfileANY", KindMotion},
|
||||
{"axis_loitering_guard", "tnsaxis:CameraApplicationPlatform/LoiteringGuard/Camera1ProfileANY", KindMotion},
|
||||
|
||||
// --- Tampering --------------------------------------------------
|
||||
|
||||
{"tamper_detector", "tns1:RuleEngine/TamperDetector/Tamper", KindTampering},
|
||||
{"global_scene_change", "tns1:VideoSource/GlobalSceneChange/ImagingService", KindTampering},
|
||||
{"hanwha_tampering", "tns1:VideoAnalytics/tnssamsung:TamperingDetection", KindTampering},
|
||||
|
||||
// --- Image quality (separated from Tampering) ------------------
|
||||
|
||||
{"image_too_dark", "tns1:VideoSource/ImageTooDark/ImagingService", KindImageQuality},
|
||||
{"image_too_bright", "tns1:VideoSource/ImageTooBright/ImagingService", KindImageQuality},
|
||||
{"image_too_blurry", "tns1:VideoSource/ImageTooBlurry/ImagingService", KindImageQuality},
|
||||
|
||||
// --- Digital input ---------------------------------------------
|
||||
|
||||
{"digital_input", "tns1:Device/Trigger/DigitalInput", KindDigitalInput},
|
||||
{"digital_input_avigilon", "tns1:Device/tns1:Trigger/tns1:DigitalInput", KindDigitalInput},
|
||||
|
||||
// --- Digital output --------------------------------------------
|
||||
|
||||
{"relay", "tns1:Device/Trigger/Relay", KindDigitalOutput},
|
||||
{"relay_avigilon", "tns1:Device/tns1:Trigger/tns1:Relay", KindDigitalOutput},
|
||||
|
||||
// --- Object analytics ------------------------------------------
|
||||
|
||||
// AXIS Object Analytics uses numeric scenario suffixes.
|
||||
{"axis_object_analytics_scenario_1", "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1", KindObjectDetected},
|
||||
{"axis_object_analytics_scenario_2", "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario2", KindObjectDetected},
|
||||
|
||||
// Standard rule-engine analytics topics.
|
||||
{"line_detector_crossed", "tns1:RuleEngine/LineDetector/Crossed", KindObjectDetected},
|
||||
{"field_detector_objects_inside", "tns1:RuleEngine/FieldDetector/ObjectsInside", KindObjectDetected},
|
||||
|
||||
// Whitelisted MyRuleDetector sub-rules.
|
||||
{"my_rule_detector_human", "tns1:RuleEngine/MyRuleDetector/HumanDetect", KindObjectDetected},
|
||||
{"my_rule_detector_vehicle", "tns1:RuleEngine/MyRuleDetector/VehicleDetect", KindObjectDetected},
|
||||
{"my_rule_detector_people", "tns1:RuleEngine/MyRuleDetector/PeopleDetect", KindObjectDetected},
|
||||
{"my_rule_detector_face", "tns1:RuleEngine/MyRuleDetector/FaceDetect", KindObjectDetected},
|
||||
{"my_rule_detector_objects_inside", "tns1:RuleEngine/MyRuleDetector/ObjectsInside", KindObjectDetected},
|
||||
|
||||
// --- Audio -----------------------------------------------------
|
||||
|
||||
{"audio_detected_sound", "tns1:AudioAnalytics/Audio/DetectedSound", KindAudioAlarm},
|
||||
{"axis_audio_trigger_level", "tns1:AudioSource/tnsaxis:TriggerLevel", KindAudioAlarm},
|
||||
{"hanwha_sound_detection", "tns1:AudioAnalytics/tnssamsung:SoundDetection", KindAudioAlarm},
|
||||
|
||||
// --- Negative cases --------------------------------------------
|
||||
|
||||
{"empty", "", KindUnknown},
|
||||
{"unknown_topic", "tns1:UserAlarm/IVA", KindUnknown},
|
||||
{"unrelated_recording_config", "tns1:RecordingConfig/JobState", KindUnknown},
|
||||
|
||||
// MyRuleDetector overmatch guard — Bosch publishes counter and
|
||||
// occupancy under the same container and these must not be
|
||||
// classified as object detection.
|
||||
{"my_rule_detector_counter_not_object", "tns1:RuleEngine/MyRuleDetector/Counter", KindUnknown},
|
||||
{"my_rule_detector_occupancy_not_object", "tns1:RuleEngine/MyRuleDetector/Occupancy", KindUnknown},
|
||||
|
||||
// Substring guards.
|
||||
{"motion_recording_not_motion", "tns1:Recording/MotionRecording/Started", KindUnknown},
|
||||
{"audio_encoder_config_not_audio_alarm", "tns1:Configuration/AudioEncoderConfiguration", KindUnknown},
|
||||
{"relay_failure_not_digital_output", "tns1:Device/HardwareFailure/RelayFailure", KindUnknown},
|
||||
{"digital_input_config_not_digital_input", "tns1:Device/IO/DigitalInputConfiguration", KindUnknown},
|
||||
{"tamper_detector_log_not_tampering", "tns1:Device/Diagnostics/TamperDetectorLog", KindUnknown},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
assert.Equal(t, tc.want, Classify(tc.topic), "topic=%q", tc.topic)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClassifyIsCaseSensitive(t *testing.T) {
|
||||
// ONVIF topic identifiers are case-sensitive per the spec; a
|
||||
// lowercased topic must not match a capitalised pattern.
|
||||
assert.Equal(t, KindUnknown, Classify("tns1:videosource/motionalarm"))
|
||||
}
|
||||
|
||||
func TestCanonicalizeTopicStripsNamespaces(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
in string
|
||||
want string
|
||||
}{
|
||||
{"single_namespace", "tns1:VideoSource/MotionAlarm", "VideoSource/MotionAlarm"},
|
||||
{"per_segment_namespace", "tns1:Device/tns1:Trigger/tns1:Relay", "Device/Trigger/Relay"},
|
||||
{"vendor_namespace_inner", "tns1:VideoAnalytics/tnssamsung:MotionDetection", "VideoAnalytics/MotionDetection"},
|
||||
{"axis_outer_namespace", "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1", "CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1"},
|
||||
{"empty", "", ""},
|
||||
{"no_colon_passthrough", "Foo/Bar", "Foo/Bar"},
|
||||
{"double_slash_keeps_empty_segment", "tns1://Foo", "//Foo"},
|
||||
{"colon_only_segment_collapses_to_empty", "tns1:/Foo", "/Foo"},
|
||||
{"trailing_colon_segment", "tns1:", ""},
|
||||
{"multi_colon_takes_first", "tns1:Foo:Bar/Baz", "Foo:Bar/Baz"},
|
||||
{"leading_slash_kept", "/tns1:Foo", "/Foo"},
|
||||
{"trailing_slash_kept", "tns1:Foo/", "Foo/"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
assert.Equal(t, tc.want, canonicalizeTopic(tc.in), "input=%q", tc.in)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClassifyRuleOrder_ObjectAnalyticsBeforeGenericObjects(t *testing.T) {
|
||||
// Locks the invariant that the prefix rule "ObjectAnalytics/" is
|
||||
// matched before the broader "ObjectsInside" rule. Without this
|
||||
// ordering, AXIS AOA topics that contain neither would still classify
|
||||
// correctly via the ObjectAnalytics/ rule; we encode the dependency.
|
||||
topic := "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1"
|
||||
assert.Equal(t, KindObjectDetected, Classify(topic))
|
||||
}
|
||||
159
event/stream/types.go
Normal file
159
event/stream/types.go
Normal file
@@ -0,0 +1,159 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Kind is the normalized category of an ONVIF event, independent of the
|
||||
// camera vendor's topic naming.
|
||||
type Kind uint8
|
||||
|
||||
const (
|
||||
KindUnknown Kind = iota
|
||||
KindMotion
|
||||
KindTampering
|
||||
// KindImageQuality covers VideoSource imaging alarms. Kept separate
|
||||
// from KindTampering because they fire on legitimate sunset / dawn /
|
||||
// condensation transitions, not on interference.
|
||||
KindImageQuality
|
||||
KindDigitalInput
|
||||
KindDigitalOutput
|
||||
KindObjectDetected
|
||||
KindAudioAlarm
|
||||
)
|
||||
|
||||
func (k Kind) String() string {
|
||||
switch k {
|
||||
case KindUnknown:
|
||||
return "Unknown"
|
||||
case KindMotion:
|
||||
return "Motion"
|
||||
case KindTampering:
|
||||
return "Tampering"
|
||||
case KindImageQuality:
|
||||
return "ImageQuality"
|
||||
case KindDigitalInput:
|
||||
return "DigitalInput"
|
||||
case KindDigitalOutput:
|
||||
return "DigitalOutput"
|
||||
case KindObjectDetected:
|
||||
return "ObjectDetected"
|
||||
case KindAudioAlarm:
|
||||
return "AudioAlarm"
|
||||
default:
|
||||
return fmt.Sprintf("Kind(%d)", uint8(k))
|
||||
}
|
||||
}
|
||||
|
||||
// State is the active/inactive level carried by a boolean ONVIF property
|
||||
// event. StateUnknown is used both when the value cannot be parsed and
|
||||
// when the topic is edge-triggered and carries no boolean state.
|
||||
type State uint8
|
||||
|
||||
const (
|
||||
StateUnknown State = iota
|
||||
StateActive
|
||||
StateInactive
|
||||
)
|
||||
|
||||
func (s State) String() string {
|
||||
switch s {
|
||||
case StateUnknown:
|
||||
return "Unknown"
|
||||
case StateActive:
|
||||
return "Active"
|
||||
case StateInactive:
|
||||
return "Inactive"
|
||||
default:
|
||||
return fmt.Sprintf("State(%d)", uint8(s))
|
||||
}
|
||||
}
|
||||
|
||||
// PropertyOperation mirrors the wsnt:PropertyOperation attribute.
|
||||
// PropertyUnknown covers both "absent on the wire" (the attribute is
|
||||
// optional) and "unrecognised value".
|
||||
type PropertyOperation uint8
|
||||
|
||||
const (
|
||||
PropertyUnknown PropertyOperation = iota
|
||||
PropertyInitialized
|
||||
PropertyChanged
|
||||
PropertyDeleted
|
||||
)
|
||||
|
||||
func (p PropertyOperation) String() string {
|
||||
switch p {
|
||||
case PropertyUnknown:
|
||||
return "Unknown"
|
||||
case PropertyInitialized:
|
||||
return "Initialized"
|
||||
case PropertyChanged:
|
||||
return "Changed"
|
||||
case PropertyDeleted:
|
||||
return "Deleted"
|
||||
default:
|
||||
return fmt.Sprintf("PropertyOperation(%d)", uint8(p))
|
||||
}
|
||||
}
|
||||
|
||||
// Event is a single normalized notification from an ONVIF device.
|
||||
//
|
||||
// Source and Data are maps because ONVIF notifications can carry
|
||||
// multiple SimpleItems — AXIS Object Analytics emits active+classType+
|
||||
// confidence in one Data list, DigitalInput carries InputToken in Source
|
||||
// and LogicalState in Data.
|
||||
type Event struct {
|
||||
Kind Kind
|
||||
State State
|
||||
Operation PropertyOperation
|
||||
DeviceID string
|
||||
Source map[string]string
|
||||
Data map[string]string
|
||||
Topic string
|
||||
Timestamp time.Time
|
||||
// DeviceTime is the camera-reported wsnt:UtcTime. Cameras drift —
|
||||
// prefer Timestamp for ordering and DeviceTime only for forensics or
|
||||
// cross-camera correlation when the caller manages NTP.
|
||||
DeviceTime time.Time
|
||||
// AfterReconnect is true for events delivered after the Stream
|
||||
// silently recreated its subscription. Cameras replay current state
|
||||
// with PropertyInitialized on a new subscription; watch this flag to
|
||||
// suppress duplicate edge-detection. Cleared on the first non-
|
||||
// Initialized event.
|
||||
AfterReconnect bool
|
||||
}
|
||||
|
||||
// Op identifies which Stream operation failed.
|
||||
type Op string
|
||||
|
||||
const (
|
||||
OpPull Op = "pull"
|
||||
OpRenew Op = "renew"
|
||||
OpRecreate Op = "recreate"
|
||||
)
|
||||
|
||||
// ErrPullFailed wraps a transient PullMessages failure. The pull loop
|
||||
// surfaces it and continues.
|
||||
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. Recovered implicitly: a
|
||||
// permanently failing renew lets the subscription die, pull starts
|
||||
// failing, and the reconnect path 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. 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 }
|
||||
143
event/stream/types_test.go
Normal file
143
event/stream/types_test.go
Normal file
@@ -0,0 +1,143 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestKindString(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
kind Kind
|
||||
want string
|
||||
}{
|
||||
{"unknown", KindUnknown, "Unknown"},
|
||||
{"motion", KindMotion, "Motion"},
|
||||
{"tampering", KindTampering, "Tampering"},
|
||||
{"image_quality", KindImageQuality, "ImageQuality"},
|
||||
{"digital_input", KindDigitalInput, "DigitalInput"},
|
||||
{"digital_output", KindDigitalOutput, "DigitalOutput"},
|
||||
{"object_detected", KindObjectDetected, "ObjectDetected"},
|
||||
{"audio_alarm", KindAudioAlarm, "AudioAlarm"},
|
||||
{"out_of_range", Kind(255), "Kind(255)"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
assert.Equal(t, tc.want, tc.kind.String())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestKindStringsAreUnique(t *testing.T) {
|
||||
seen := map[string]Kind{}
|
||||
for k := KindUnknown; k <= KindAudioAlarm; k++ {
|
||||
s := k.String()
|
||||
prev, dup := seen[s]
|
||||
assert.False(t, dup, "duplicate String %q for Kind(%d) and Kind(%d)", s, prev, k)
|
||||
seen[s] = k
|
||||
}
|
||||
}
|
||||
|
||||
func TestStateString(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
state State
|
||||
want string
|
||||
}{
|
||||
{"unknown", StateUnknown, "Unknown"},
|
||||
{"active", StateActive, "Active"},
|
||||
{"inactive", StateInactive, "Inactive"},
|
||||
{"out_of_range", State(255), "State(255)"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
assert.Equal(t, tc.want, tc.state.String())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPropertyOperationString(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
op PropertyOperation
|
||||
want string
|
||||
}{
|
||||
{"unknown", PropertyUnknown, "Unknown"},
|
||||
{"initialized", PropertyInitialized, "Initialized"},
|
||||
{"changed", PropertyChanged, "Changed"},
|
||||
{"deleted", PropertyDeleted, "Deleted"},
|
||||
{"out_of_range", PropertyOperation(255), "PropertyOperation(255)"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
assert.Equal(t, tc.want, tc.op.String())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventZeroValue(t *testing.T) {
|
||||
var e Event
|
||||
assert.Equal(t, KindUnknown, e.Kind)
|
||||
assert.Equal(t, StateUnknown, e.State)
|
||||
assert.Equal(t, PropertyUnknown, e.Operation)
|
||||
assert.Empty(t, e.DeviceID)
|
||||
assert.Nil(t, e.Source)
|
||||
assert.Nil(t, e.Data)
|
||||
assert.Empty(t, e.Topic)
|
||||
assert.True(t, e.Timestamp.IsZero())
|
||||
assert.True(t, e.DeviceTime.IsZero())
|
||||
}
|
||||
|
||||
func TestEventFieldAssignmentRoundTrip(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
deviceTime := now.Add(-2 * time.Second)
|
||||
e := Event{
|
||||
Kind: KindMotion,
|
||||
State: StateActive,
|
||||
Operation: PropertyChanged,
|
||||
DeviceID: "axis-camera-01",
|
||||
Source: map[string]string{"InputToken": "DI1"},
|
||||
Data: map[string]string{"LogicalState": "true"},
|
||||
Topic: "tns1:Device/Trigger/DigitalInput",
|
||||
Timestamp: now,
|
||||
DeviceTime: deviceTime,
|
||||
}
|
||||
assert.Equal(t, KindMotion, e.Kind)
|
||||
assert.Equal(t, StateActive, e.State)
|
||||
assert.Equal(t, PropertyChanged, e.Operation)
|
||||
assert.Equal(t, "axis-camera-01", e.DeviceID)
|
||||
assert.Equal(t, "DI1", e.Source["InputToken"])
|
||||
assert.Equal(t, "true", e.Data["LogicalState"])
|
||||
assert.Equal(t, "tns1:Device/Trigger/DigitalInput", e.Topic)
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
176
examples/event/stream/main.go
Normal file
176
examples/event/stream/main.go
Normal file
@@ -0,0 +1,176 @@
|
||||
// Command streamtest opens an event stream against an ONVIF camera and
|
||||
// prints decoded events as they arrive. Useful for verifying the
|
||||
// classifier against real-camera topics; not intended as a production
|
||||
// tool.
|
||||
//
|
||||
// # Usage
|
||||
//
|
||||
// go run ./examples/event/stream \
|
||||
// -xaddr 192.168.1.10 \
|
||||
// -username root \
|
||||
// -duration 60s
|
||||
//
|
||||
// # Credentials
|
||||
//
|
||||
// The camera password is read, in order of preference:
|
||||
//
|
||||
// 1. The ONVIF_PASSWORD environment variable.
|
||||
// 2. A file pointed at by -password-file (newline stripped).
|
||||
// 3. Interactive prompt when stdin is a tty.
|
||||
//
|
||||
// -password is also accepted but DISCOURAGED — it leaks the credential
|
||||
// into shell history and the system process listing. Use only for
|
||||
// throwaway dev cameras.
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/onvif"
|
||||
"github.com/kerberos-io/onvif/event/stream"
|
||||
)
|
||||
|
||||
func main() {
|
||||
xaddr := flag.String("xaddr", "", "camera host or host:port (required)")
|
||||
username := flag.String("username", "", "ONVIF user (required)")
|
||||
insecurePassword := flag.String("password", "", "INSECURE — leaks into shell history; prefer ONVIF_PASSWORD env or -password-file")
|
||||
passwordFile := flag.String("password-file", "", "read password from this file (newline trimmed)")
|
||||
deviceID := flag.String("device-id", "", "logical name printed with each event (default: xaddr)")
|
||||
filter := flag.String("filter", "", "raw ONVIF ConcreteSet topic filter (empty = all topics, works on AXIS)")
|
||||
pullTimeout := flag.Duration("pull-timeout", 5*time.Second, "server-side wait per PullMessages call")
|
||||
duration := flag.Duration("duration", 0, "stop after this long (0 = run until Ctrl-C)")
|
||||
flag.Parse()
|
||||
|
||||
if *xaddr == "" || *username == "" {
|
||||
flag.Usage()
|
||||
os.Exit(2)
|
||||
}
|
||||
if *deviceID == "" {
|
||||
*deviceID = *xaddr
|
||||
}
|
||||
|
||||
password, err := loadPassword(*insecurePassword, *passwordFile)
|
||||
if err != nil {
|
||||
log.Fatalf("password: %v", err)
|
||||
}
|
||||
|
||||
dev, err := onvif.NewDevice(onvif.DeviceParams{
|
||||
Xaddr: *xaddr,
|
||||
Username: *username,
|
||||
Password: password,
|
||||
AuthMode: onvif.UsernameTokenAuth,
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatalf("connect: %v", err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if *duration > 0 {
|
||||
var done context.CancelFunc
|
||||
ctx, done = context.WithTimeout(ctx, *duration)
|
||||
defer done()
|
||||
}
|
||||
|
||||
sigs := make(chan os.Signal, 1)
|
||||
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
|
||||
go func() {
|
||||
<-sigs
|
||||
cancel()
|
||||
}()
|
||||
|
||||
s, err := stream.NewStream(ctx, dev, stream.Options{
|
||||
DeviceID: *deviceID,
|
||||
RawTopicFilter: *filter,
|
||||
PullTimeout: *pullTimeout,
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatalf("open stream: %v", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := s.Close(); err != nil {
|
||||
log.Printf("stream close: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
log.Printf("streaming from %s (device-id=%s, filter=%q)", *xaddr, *deviceID, *filter)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Printf("done (%v)", ctx.Err())
|
||||
return
|
||||
case ev, ok := <-s.Events():
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
fmt.Printf("%s kind=%-15s state=%-9s op=%-12s topic=%s",
|
||||
ev.Timestamp.Format(time.RFC3339), ev.Kind, ev.State, ev.Operation, ev.Topic)
|
||||
if ev.AfterReconnect {
|
||||
fmt.Print(" [after-reconnect]")
|
||||
}
|
||||
if len(ev.Source) > 0 {
|
||||
fmt.Printf(" source=%v", ev.Source)
|
||||
}
|
||||
if len(ev.Data) > 0 {
|
||||
fmt.Printf(" data=%v", ev.Data)
|
||||
}
|
||||
fmt.Println()
|
||||
case e, ok := <-s.Errors():
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var pull stream.ErrPullFailed
|
||||
var recreate stream.ErrRecreateFailed
|
||||
switch {
|
||||
case errors.As(e, &recreate):
|
||||
log.Printf("RECREATE failed: %v (camera may be offline)", recreate.Err)
|
||||
case errors.As(e, &pull):
|
||||
log.Printf("pull error (will retry): %v", pull.Err)
|
||||
default:
|
||||
log.Printf("stream error: %v", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// loadPassword resolves the camera password from the environment first
|
||||
// (ONVIF_PASSWORD), then -password-file, then an interactive prompt as
|
||||
// a last resort. The insecure -password flag is honoured only if
|
||||
// nothing else is set, and a warning is logged.
|
||||
func loadPassword(insecure, file string) (string, error) {
|
||||
if env := os.Getenv("ONVIF_PASSWORD"); env != "" {
|
||||
return env, nil
|
||||
}
|
||||
if file != "" {
|
||||
b, err := os.ReadFile(file)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("read %s: %w", file, err)
|
||||
}
|
||||
return strings.TrimRight(string(b), "\r\n"), nil
|
||||
}
|
||||
if insecure != "" {
|
||||
log.Print("WARNING: -password leaks into shell history and process listings; prefer ONVIF_PASSWORD env or -password-file")
|
||||
return insecure, nil
|
||||
}
|
||||
// Interactive prompt — works when stdin is a tty. We use a plain
|
||||
// reader (rather than golang.org/x/term hidden input) to keep
|
||||
// this example dependency-free; in production, callers should
|
||||
// integrate term.ReadPassword.
|
||||
fmt.Fprint(os.Stderr, "ONVIF password (visible): ")
|
||||
r := bufio.NewReader(os.Stdin)
|
||||
line, err := r.ReadString('\n')
|
||||
if err != nil {
|
||||
return "", errors.New("no password supplied (set ONVIF_PASSWORD, -password-file, or pipe input)")
|
||||
}
|
||||
return strings.TrimRight(line, "\r\n"), nil
|
||||
}
|
||||
Reference in New Issue
Block a user