kerberos-io/onvif
1
0
Fork 0
mirror of https://github.com/kerberos-io/onvif.git synced 2026-08-23 15:08:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs(event/stream): trim comments to WHY, drop noise

Browse Source 
Audit against the standard 'default to no comments; only add one when
the WHY is non-obvious'. Net: 238 lines removed across 8 files, no
behaviour change, tests still pass -race.

What went
---------
* Section banners (// ---------- Motion ----------): noise once
  per-rule citations exist.
* Per-rule 'Data: IsMotion (xsd:boolean)' wire-format lines in
  topics.go: that's WHAT; the spec citation carries WHY.
* Per-field doc on Event struct restating each field name (// Kind
  is the normalized event category) and the type-doc preamble.
* Stringer doc comments ('// String implements fmt.Stringer.') and
  similar conventional-method noise.
* 'Used by ErrPullFailed / ErrRenewFailed / ErrRecreateFailed' in
  the Op doc — the rule-named anti-pattern.
* doc.go Invariants and Reconnect sections duplicating per-function
  docs.
* Internal helper doc-comments restating what the function does
  (surfaceError, run, simpleItemsToMap first sentence, etc.).

What stayed
-----------
* Every spec / vendor-doc citation in topics.go.
* Race-condition WHY in stream.go run() close ordering.
* Workaround WHY in renew.go (absolute datetime vs duration).
* WS-BaseNotification UTC rationale + vendor format list in
  decode.go.
* Fleet-sizing and thundering-herd rationale in reconnect.go.
* Stream consumer invariants (NewStream synchronous I/O, Errors
  non-blocking, Close idempotent + bounded).

The change matches the codebase's stated style (CLAUDE.md): WHY only,
no WHAT, no cross-file references, no current-task narration.
This commit is contained in:
Sebastian Norling
2026-05-21 18:43:08 +02:00
parent badcc8fba2
commit fcc3a90f9b
8 changed files with 199 additions and 437 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
event/stream/decode.go
View File

@@ -7,20 +7,9 @@ import (
"github.com/kerberos-io/onvif/event"
)
// decode converts a single ONVIF NotificationMessage into the package's
// normalized Event representation. Unexported because the only intended
// caller is the Stream; downstream consumers receive decoded Events on
// the Events channel. Tests reach decode directly because they're in
// the same package.
//
// deviceID is supplied by the caller because the message itself does not
// identify the originating camera. observedAt is recorded verbatim as
// Event.Timestamp; the camera-reported wsnt:UtcTime attribute (when
// present and parseable) populates Event.DeviceTime.
//
// When the Topic does not match any classifier rule the returned Event
// has Kind == KindUnknown but Source, Data and Topic are still populated
// so consumers can fall back to inspecting the wire form.
// 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
@@ -37,9 +26,8 @@ func decode(msg event.NotificationMessage, deviceID string, observedAt time.Time
}
}
// simpleItemsToMap collapses ONVIF SimpleItem lists to a Name->Value map.
// Returns nil for an empty list so empty notifications do not allocate
// and match the Event zero-value contract.
// 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
@@ -51,14 +39,9 @@ func simpleItemsToMap(items []event.SimpleItem) map[string]string {
return m
}
// extractState scans Data items for a boolean-like value and returns the
// first one as a State. Returns StateUnknown when no item parses — this
// is the correct outcome for edge-triggered topics such as
// 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.
//
// Iteration order over the original []SimpleItem is preserved so the
// behaviour stays deterministic per notification. (Map iteration is not
// involved; simpleItemsToMap is a separate path.)
func extractState(items []event.SimpleItem) State {
for _, it := range items {
switch strings.ToLower(strings.TrimSpace(string(it.Value))) {
@@ -71,9 +54,8 @@ func extractState(items []event.SimpleItem) State {
return StateUnknown
}
// parsePropertyOperation parses the wsnt:PropertyOperation attribute.
// The attribute is optional per WS-Notification; an empty or unrecognised
// value yields PropertyUnknown.
// parsePropertyOperation returns PropertyUnknown for absent (optional
// per WS-Notification) or unrecognised values.
func parsePropertyOperation(s string) PropertyOperation {
switch s {
case "Initialized":
@@ -87,15 +69,11 @@ func parsePropertyOperation(s string) PropertyOperation {
}
}
// parseDeviceTime parses the wsnt:UtcTime attribute, returning the zero
// time when the attribute is absent or unparseable. The result is
// normalised to UTC so equality comparisons across timezones work.
//
// xsd:dateTime in ONVIF messages is RFC 3339 in practice but real
// cameras emit several flavours: with/without sub-seconds, with colon
// or compact ("+0200") timezone offsets, and some older Hikvision
// firmwares omit the timezone entirely (treated as UTC per
// WS-BaseNotification which mandates UTC for UtcTime).
// 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{}
@@ -108,14 +86,11 @@ func parseDeviceTime(s string) time.Time {
return time.Time{}
}
// deviceTimeLayouts lists the wsnt:UtcTime forms observed across vendor
// firmwares. Ordered from most-precise / most-common first so the
// happy path hits early.
var deviceTimeLayouts = []string{
time.RFC3339Nano, // 2026-05-21T10:30:00.500Z, ...+02:00
time.RFC3339, // 2026-05-21T10:30:00Z, ...+02:00
"2006-01-02T15:04:05.999-0700", // sub-second + compact offset (Geovision)
"2006-01-02T15:04:05-0700", // compact offset (some Dahua)
"2006-01-02T15:04:05.999", // no timezone, sub-second (rare)
"2006-01-02T15:04:05", // naked, no TZ (older Hikvision)
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)
}

47
event/stream/doc.go
View File

@@ -7,7 +7,7 @@
//
// 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 camera does not advertise events */ }
// if err != nil { /* construction failed: auth, network, or no event support */ }
// defer s.Close()
//
// for ev := range s.Events() {
@@ -17,43 +17,12 @@
// }
// }
//
// # Invariants
// 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.
//
// NewStream performs network I/O. It returns once the
// CreatePullPointSubscription call has succeeded; auth and reachability
// failures surface as an error from NewStream rather than landing on
// the Errors channel later.
//
// Two goroutines back each Stream: a pull loop and a renew loop. Both
// exit when the context passed to NewStream is cancelled or when Close
// is called. Close is idempotent and bounded — see Stream.Close.
//
// Events is closed exactly when the Stream stops. Ranging over Events
// is safe; a closed channel terminates the loop without a Close call.
// Errors is also closed at stop time. Both channels are buffered (16
// slots by default); sends to Errors are non-blocking so a stalled
// consumer drops older errors rather than the pull loop blocking on
// log output.
//
// The decoded Event preserves the wire form (Topic, raw Source and
// Data maps) so callers can fall back to inspecting non-standard
// payloads when Kind is KindUnknown.
//
// # Reconnect
//
// On ReconnectAfterFailures consecutive PullMessages failures the
// Stream silently recreates its pull-point subscription. ONVIF cameras
// replay each property's current value with PropertyInitialized on a
// new subscription; Events delivered between recreate and the first
// non-Initialized event carry Event.AfterReconnect=true so consumers
// can suppress duplicate handling.
//
// Set Options.DisableReconnect=true to opt out of recreate; the pull
// loop will retry against the original subscription until ctx cancel.
//
// # Topic classification
//
// Classify maps ONVIF topic strings to a small set of normalized Kind
// values across AXIS, Hikvision, Avigilon, Hanwha, Bosch and Dahua. See
// topics.go for the verified mapping table with public-doc citations.
// See topics.go for the verified topic→Kind mapping across AXIS,
// Hikvision, Avigilon, Hanwha, Bosch and Dahua.
package stream

53
event/stream/reconnect.go
View File

@@ -6,31 +6,21 @@ import (
"time"
)
// maxRecreateBackoff caps exponential backoff between recreate attempts.
// Sized for fleet deployments: a 1000-camera setup recovering from a
// switch reboot would otherwise hammer the network with one recreate
// attempt per camera per 30s; 5 minutes gives the network time to
// settle while still recovering promptly when a single camera comes
// back.
// 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 is the symmetric jitter applied to recreate backoff:
// the actual sleep is sampled from [backoff*(1-jitter), backoff*(1+jitter)].
// Prevents thundering-herd reconnects when many cameras drop together
// (switch reboot, NAT timeout).
// jitterFraction prevents thundering-herd reconnects when many
// cameras drop together (switch reboot, NAT timeout).
const jitterFraction = 0.25
// pullLoop is the main pull goroutine of a Stream. It calls
// PullMessages in a tight loop, decodes results into Events and feeds
// the Events channel.
//
// After ReconnectAfterFailures consecutive pull errors it asks
// attemptRecreate to recreate the pull-point subscription, marking the
// next batch's events with AfterReconnect so consumers can suppress
// duplicate handling of the ONVIF Initialized-replay that follows a
// new subscription.
//
// Exits when ctx is cancelled.
// 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
@@ -59,7 +49,6 @@ func (s *Stream) pullLoop(ctx context.Context) {
}
continue
}
// Successful pull resets failure tracking.
failures = 0
recreateBackoff = s.opts.RetryBackoff
observedAt := s.now()
@@ -67,11 +56,8 @@ func (s *Stream) pullLoop(ctx context.Context) {
ev := decode(m, s.opts.DeviceID, observedAt)
if afterReconnect {
ev.AfterReconnect = true
// ONVIF replays current state with
// PropertyInitialized on a new subscription.
// Clear the flag as soon as we see anything
// other than Initialized — at that point we
// have transitioned to live events.
// Clear once the camera transitions past the
// Initialized replay to live events.
if ev.Operation != PropertyInitialized {
afterReconnect = false
}
@@ -85,11 +71,8 @@ func (s *Stream) pullLoop(ctx context.Context) {
}
}
// attemptRecreate calls CreatePullPointSubscription and on success
// installs the new endpoint atomically. The first return is true when
// recreate succeeded just now (caller flags the next batch with
// AfterReconnect). The second return is false only if ctx was cancelled
// during backoff (caller should exit the run loop).
// 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 {
@@ -109,10 +92,8 @@ func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *ti
return true, true
}
// jitter returns d perturbed by ±jitterFraction. Used to spread
// recreate attempts across a fleet so a synchronised drop (switch
// reboot, DHCP storm) does not cause a synchronised reconnect surge.
// Returns at least 1ns to keep sleepCtx happy.
// 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

26
event/stream/renew.go
View File

@@ -10,18 +10,15 @@ import (
"github.com/kerberos-io/onvif/xsd"
)
// renewLoop refreshes the subscription before InitialTermination expires.
// Exits when ctx is cancelled. Renew failures surface as ErrRenewFailed
// on the Errors channel; the loop continues because a permanently
// failing renew will eventually drop the subscription and the pull
// loop's reconnect path will recover (recreate is the only reliable
// recovery once a subscription is GC'd at the camera).
// 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): fall back to
// renewing at half the termination so we still refresh,
// rather than busy-looping or never renewing.
// Pathological config (margin >= termination): renew at
// half termination so we still refresh.
interval = s.opts.InitialTermination / 2
if interval <= 0 {
interval = time.Second
@@ -41,13 +38,10 @@ func (s *Stream) renewLoop(ctx context.Context) {
}
}
// renewPullPoint issues a wsnt:Renew SOAP against the given
// subscription endpoint with an absolute TerminationTime.
//
// WS-BaseNotification §6.1.1 declares TerminationTime as xsd:dateTime
// OR xsd:duration, but older Hikvision, some Dahua and some Bosch
// firmwares reject the relative-duration form. We send an absolute
// UTC datetime to match what production NVRs do.
// 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)}

51
event/stream/soap.go
View File

@@ -16,16 +16,12 @@ import (
"github.com/kerberos-io/onvif/xsd"
)
// maxResponseBytes caps the size of a SOAP response we will buffer in
// memory. ONVIF PullMessages bodies are normally <100KB even with dense
// analytics payloads; 10 MiB is comfortably above legitimate traffic
// while keeping a hostile or buggy camera from OOMing the process.
// 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
// createPullPoint issues a CreatePullPointSubscription against the
// device service. Returns the SubscriptionReference Address, which is
// the endpoint subsequent PullMessages / Renew / Unsubscribe calls
// target.
func createPullPoint(c caller, opts Options) (string, error) {
term := xsd.String(durationToXSD(opts.InitialTermination))
req := event.CreatePullPointSubscription{InitialTerminationTime: &term}
@@ -56,9 +52,8 @@ func createPullPoint(c caller, opts Options) (string, error) {
return addr, nil
}
// pullMessages issues PullMessages against an active subscription
// endpoint and returns the decoded NotificationMessage list. Empty
// slice (not error) when the camera had nothing within PullTimeout.
// 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)),
@@ -83,9 +78,8 @@ func pullMessages(c caller, endpoint string, opts Options) ([]event.Notification
return decoded.NotificationMessage, nil
}
// unsubscribePullPoint sends a best-effort Unsubscribe to release the
// subscription server-side. Empty endpoint is a no-op (the construction
// failed before installing one).
// 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
@@ -102,9 +96,6 @@ func unsubscribePullPoint(c caller, endpoint string) error {
return err
}
// readClose reads at most maxResponseBytes from resp.Body and closes
// it. LimitReader prevents a hostile or buggy camera from OOMing the
// agent by streaming an unbounded response.
func readClose(resp *http.Response) (string, error) {
if resp == nil || resp.Body == nil {
return "", errors.New("nil HTTP response")
@@ -118,14 +109,13 @@ func readClose(resp *http.Response) (string, error) {
}
// unmarshalNode finds the first XML start element with the given local
// name and decodes it into out. ONVIF SOAP responses come wrapped in an
// envelope with multiple namespace prefixes; this helper sidesteps
// namespace matching by keying on local name only.
// 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 instead of the expected
// response, the fault reason is surfaced as the error so callers can
// distinguish "auth failed" / "subscription expired" from "unparseable
// response".
// 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)
@@ -160,9 +150,9 @@ var (
soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)</(?:[^:>\s]+:)?Text>`)
)
// extractSOAPFault returns the human-readable reason text from a SOAP
// fault, or empty string when the body is not a fault. Handles both
// SOAP 1.1 (faultstring) and SOAP 1.2 (Reason/Text) shapes.
// 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 ""
@@ -176,10 +166,9 @@ func extractSOAPFault(body string) string {
return ""
}
// durationToXSD formats a Go time.Duration as an xsd:duration string in
// PTnS form. Second precision is sufficient — ONVIF cameras do not
// honour sub-second pull timeouts and intermediate routers may round in
// any case.
// 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 {

143
event/stream/stream.go
View File

@@ -10,70 +10,51 @@ import (
"github.com/kerberos-io/onvif"
)
// closeUnsubscribeTimeout bounds the SOAP Unsubscribe call issued by
// Close so a hung camera connection cannot wedge the caller. The
// subscription expires at the camera anyway once InitialTermination
// elapses, so a missed unsubscribe is at worst cosmetic.
// 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 entirely set DisableReconnect=true
// (sentinel `ReconnectAfterFailures=0` would otherwise collide with the
// default-injection policy). To get a synchronous (unbuffered) channel
// pair set BufferSize=-1.
// 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 identifies the camera in emitted Events. Recommended so
// a single channel can fan in multiple cameras. Empty is allowed.
DeviceID string
// RawTopicFilter is the raw ONVIF ConcreteSet TopicExpression
// filter passed to CreatePullPointSubscription. Empty means no
// filter — required for AXIS, accepted by every other vendor we
// support. The name carries 'Raw' because the value is fed verbatim
// into the SOAP envelope: callers should normally leave it empty
// and rely on Classify for routing rather than ask the camera to
// filter server-side, which is fragile across vendors.
// 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 is the server-side wait time in each PullMessages
// call (xsd:duration). The camera returns early when messages are
// available; otherwise it returns empty after this timeout. Zero
// means default (5s).
// PullTimeout — zero means default (5s).
PullTimeout time.Duration
// MessageLimit caps the number of NotificationMessage entries
// returned per PullMessages call. Zero means default (32). A busy
// AXIS with many configured inputs can burst beyond 10 per pull;
// 32 covers that without significantly enlarging quiet pulls.
// MessageLimit — zero means default (32). Busy AXIS cameras with
// many configured rules can burst beyond 10 per pull.
MessageLimit int
// InitialTermination is the requested subscription lifetime passed
// to CreatePullPointSubscription. The renew loop refreshes well
// before this expires. Zero means default (60s).
// InitialTermination — zero means default (60s).
InitialTermination time.Duration
// RenewMargin is how long before InitialTermination expiry the
// renew loop fires. Larger margins tolerate slower networks at the
// cost of more renew SOAP calls. Zero means default (10s).
// RenewMargin — larger margins tolerate slower networks at the
// cost of more renew calls. Zero means default (10s).
RenewMargin time.Duration
// ReconnectAfterFailures is the consecutive PullMessages failure
// count that triggers a CreatePullPointSubscription recreate. The
// camera or pull-point can die for many reasons (camera reboot,
// subscription garbage-collected after a renew miss, intermediate
// NAT timeout); rebuilding the subscription is the only reliable
// recovery. Zero means default (3). To disable reconnect entirely
// set DisableReconnect=true.
// 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 skips automatic CreatePullPointSubscription
// recreate. The pull loop will continue retrying against the
// original endpoint until ctx is cancelled. Useful for tests or
// callers managing recovery externally.
// DisableReconnect makes the pull loop retry against the
// original endpoint until ctx is cancelled.
DisableReconnect bool
// RetryBackoff is the initial sleep between a pull/recreate failure
// and the next attempt. Recreate failures double this up to a 30s
// ceiling. Zero means default (1s).
// RetryBackoff is the base sleep between pull/recreate failures.
// Recreate failures double this up to maxRecreateBackoff. Zero
// means default (1s).
RetryBackoff time.Duration
// BufferSize is the buffer size of the Events and Errors channels.
// Larger buffers absorb consumer hiccups at the cost of memory.
// Zero means default (16); use -1 for unbuffered (synchronous)
// channels.
// BufferSize — zero means default (16); use -1 for unbuffered.
BufferSize int
}
@@ -109,7 +90,6 @@ func (o Options) withDefaults() Options {
if o.RetryBackoff > 0 {
d.RetryBackoff = o.RetryBackoff
}
// BufferSize: zero -> default; negative -> 0 (unbuffered).
switch {
case o.BufferSize > 0:
d.BufferSize = o.BufferSize
@@ -122,13 +102,9 @@ func (o Options) withDefaults() Options {
return d
}
// caller is the subset of *onvif.Device the Stream depends on. Tests
// substitute a fake; production code uses the device adapter.
//
// Implementations must be safe for concurrent use: the pull loop and
// renew loop call into caller from separate goroutines. *onvif.Device
// satisfies this because its HTTP client is the goroutine-safe
// http.Client.
// 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.
type caller interface {
CallMethod(method any) (*http.Response, error)
SendSoap(endpoint, body string) (*http.Response, error)
@@ -144,12 +120,9 @@ func (d deviceCaller) SendSoap(endpoint, body string) (*http.Response, error) {
return d.dev.SendSoap(endpoint, body)
}
// Stream owns a single ONVIF pull-point subscription and surfaces the
// decoded notifications on a typed channel. Close stops the background
// goroutine and unsubscribes from the camera.
//
// A Stream is safe for concurrent use by Close from any goroutine while
// readers consume Events / Errors; Close is idempotent.
// 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
@@ -166,7 +139,7 @@ type Stream struct {
closeOnce sync.Once
closeErr error
// now is overridable in tests to make timestamps deterministic.
// now is overridable so tests can make timestamps deterministic.
now func() time.Time
}
@@ -182,14 +155,11 @@ func (s *Stream) setPullPoint(addr string) {
s.pullPoint = addr
}
// NewStream creates a Stream against an ONVIF device. It performs the
// CreatePullPointSubscription call synchronously so connectivity and
// authentication problems surface immediately as an error rather than
// landing on the Errors channel later. The background pull loop starts
// before NewStream returns.
// 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 when Close is
// called.
// 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)
}
@@ -215,22 +185,20 @@ func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) {
return s, nil
}
// Events returns the channel of decoded notifications. The channel is
// closed when the Stream stops.
// 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 encountered while
// pulling. Sends are non-blocking, so consumers that fall behind drop
// older errors. The channel is closed when the Stream stops.
// 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 goroutine, waits for it to exit, and
// unsubscribes from the camera. Subsequent calls are no-ops.
// Close stops the background goroutines, waits for them to exit and
// Unsubscribes from the camera. Subsequent calls are no-ops.
//
// Unsubscribe is bounded by closeUnsubscribeTimeout so a hung camera
// connection cannot wedge the caller. On timeout Close still returns
// promptly; the subscription will expire at the camera once
// InitialTermination + RenewMargin elapses without a renew.
// connection cannot wedge the caller.
func (s *Stream) Close() error {
s.closeOnce.Do(func() {
s.cancel()
@@ -252,8 +220,6 @@ func (s *Stream) Close() error {
return s.closeErr
}
// run orchestrates the pull and renew goroutines and closes the
// emission channels once both have exited.
func (s *Stream) run(ctx context.Context) {
var wg sync.WaitGroup
wg.Add(1)
@@ -265,15 +231,13 @@ func (s *Stream) run(ctx context.Context) {
wg.Wait()
// Explicit close order after both goroutines have exited so a
// future maintainer extending this function does not accidentally
// rely on defer-ordering for channel-close safety.
// future maintainer extending this function does not rely on
// defer-ordering for channel-close safety.
close(s.errors)
close(s.events)
close(s.done)
}
// surfaceError sends err on the errors channel non-blockingly so a
// stalled consumer cannot block the pull or renew loop.
func (s *Stream) surfaceError(err error) {
select {
case s.errors <- err:
@@ -281,8 +245,7 @@ func (s *Stream) surfaceError(err error) {
}
}
// sleepCtx blocks for d or until ctx is cancelled. Returns true if d
// elapsed, false if ctx was cancelled.
// 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()

139
event/stream/topics.go
View File

@@ -2,24 +2,23 @@ package stream
import "strings"
// Classify maps an ONVIF topic string (e.g. "tns1:VideoSource/MotionAlarm")
// to the normalized Kind that callers should switch on. Returns
// Classify maps an ONVIF topic string to the normalized Kind. Returns
// KindUnknown when no rule matches.
//
// The classifier strips XML-namespace prefixes (tns1:, tnsaxis:,
// tnssamsung:, ...) from each "/"-separated segment of the topic so it is
// robust to vendor namespace variants. Matching is case-sensitive because
// ONVIF topic identifiers are case-sensitive per the spec.
// 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 (RuleEngine topics)
// - ONVIF Analytics Service Spec
// https://www.onvif.org/specs/srv/analytics/ONVIF-VideoAnalytics-Service-Spec-v220.pdf
// - ONVIF Device IO Service Spec (DigitalInput, Relay)
// - 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 ONVIF integration
// extracted from Home Assistant
// https://github.com/openvideolibs/onvif-parsers
func Classify(topic string) Kind {
if topic == "" {
@@ -34,15 +33,10 @@ func Classify(topic string) Kind {
return KindUnknown
}
// canonicalizeTopic strips the XML-namespace prefix (anything up to and
// including the first ':') from each "/"-separated segment. This collapses
// vendor variants like "tns1:Device/tns1:Trigger/tns1:Relay" (Avigilon
// serialisation) and "tns1:Device/Trigger/Relay" (everyone else) to a
// single matchable form.
//
// A segment that is only a prefix (e.g. "tns1:") canonicalizes to the
// empty string. Multiple colons in one segment are not expected in real
// ONVIF topics; the first colon wins.
// 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 {
@@ -53,133 +47,88 @@ func canonicalizeTopic(topic string) string {
return strings.Join(segments, "/")
}
// topicRules is evaluated in order; first match wins. Keep more specific
// rules ahead of broader ones — e.g. "ObjectAnalytics/" must precede any
// future bare "Analytics" rule, and "MyRuleDetector/HumanDetect" must
// precede a hypothetical broader "MyRuleDetector" entry. Each rule cites
// the documentation that supports including it.
//
// Substring matching is intentional so vendor-specific path prefixes
// outside the standard tns1: namespace (e.g.
// tnsaxis:CameraApplicationPlatform/...) still match.
//
// Note on edge-triggered topics: tns1:RuleEngine/LineDetector/Crossed
// carries an ObjectId rather than a State boolean. Consumers of Crossed
// must not expect a level-triggered Active/Inactive semantic — the Stream
// decoder will leave State as StateUnknown for these.
// 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
}{
// ---------- Motion -------------------------------------------------
// tns1:VideoSource/MotionAlarm — Profile S basic motion. Emitted by
// AXIS (basic VMD), Bosch, Dahua, Hikvision (newer firmware) and
// Hanwha as a fallback. Data SimpleItem: State (xsd:boolean).
// 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. Data: State.
// 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/Samsung
// Wisenet vendor-namespaced motion. Data: Motion ("0"/"1").
// 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. Emitted by AXIS (VMD3+), Hikvision,
// Avigilon analytics, others. Data: IsMotion (xsd:boolean).
// 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-specific region
// motion rule. Data: IsMotion (xsd:boolean).
// 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 that fire motion-like
// events with Camera<N>Profile<ID> suffixes. Treated as motion so
// they can drive motion-triggered recording on cameras configured
// with these apps instead of basic VMD.
// 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},
// ---------- Tampering ---------------------------------------------
// tns1:RuleEngine/TamperDetector/Tamper — standard ONVIF tamper
// rule. Data: IsTamper (xsd:boolean). Anchored on the rule-name
// segment so "TamperDetectorLog" (hypothetical) does not match.
// tns1:RuleEngine/TamperDetector/Tamper — standard ONVIF tamper rule.
// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.5
{"TamperDetector/Tamper", KindTampering},
// tns1:VideoSource/GlobalSceneChange/ImagingService — Hikvision (and
// others) emit this on real lens-cover / scene substitution. This is
// the proper tamper signal on firmwares without TamperDetector.
// 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 vendor.
// tns1:VideoAnalytics/tnssamsung:TamperingDetection — Hanwha.
// https://github.com/home-assistant/core/issues/66493
{"VideoAnalytics/TamperingDetection", KindTampering},
// ---------- Image quality -----------------------------------------
// tns1:VideoSource/ImageTooDark|ImageTooBright|ImageTooBlurry —
// imaging-quality alarms. Integrators (Milestone, Genetec, Frigate)
// route these separately from tamper because they fire on legitimate
// sunset/dawn/condensation transitions, not on actual interference.
// 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},
// ---------- Digital I/O -------------------------------------------
// tns1:Device/Trigger/DigitalInput — standard ONVIF DeviceIO topic.
// Avigilon emits the per-segment-prefixed variant
// "tns1:Device/tns1:Trigger/tns1:DigitalInput"; canonicalization
// folds both to the same path. Data: LogicalState (xsd:boolean),
// Source: InputToken.
// 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},
// tns1:Device/Trigger/Relay — standard ONVIF DeviceIO topic. Same
// canonicalisation note as DigitalInput. Data: LogicalState,
// Source: RelayToken.
// ONVIF-DeviceIo-Service-Spec.pdf §5.3
{"Trigger/Relay", KindDigitalOutput},
// ---------- Object analytics --------------------------------------
// tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario<N>
// — AXIS Object Analytics. Scenario suffixes are numeric per the
// AOA configuration (Device1Scenario1, Device1Scenario2, ...). Data:
// active ("0"/"1") plus classType / confidence when configured.
// — 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},
// tns1:RuleEngine/LineDetector/Crossed — line crossing (Hikvision,
// Bosch IVA, others). Data: ObjectId (xsd:int); edge-triggered, no
// State boolean.
// https://www.onvif.org/specs/srv/analytics/ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.4
// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.4
{"LineDetector/Crossed", KindObjectDetected},
// tns1:RuleEngine/FieldDetector/ObjectsInside — intrusion / region
// detector (Hikvision, Bosch, Dahua). Data: IsInside (xsd:boolean).
{"FieldDetector/ObjectsInside", KindObjectDetected},
// tns1:RuleEngine/MyRuleDetector/<RuleName> — vendor-defined rule
// names under the ONVIF MyRuleDetector container. We whitelist
// object-class rules emitted by Bosch IVA, Dahua SMD and Hikvision
// AcuSense so non-object rules under the same container (Bosch
// Counter, Occupancy) do not get mis-classified.
// 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},
@@ -187,16 +136,8 @@ var topicRules = []struct {
{"MyRuleDetector/ObjectsInside", KindObjectDetected},
{"MyRuleDetector/FaceDetect", KindObjectDetected},
// ---------- Audio --------------------------------------------------
// tns1:AudioAnalytics/Audio/DetectedSound — standard ONVIF audio
// detection. Data: State (xsd:boolean).
{"Audio/DetectedSound", KindAudioAlarm},
// tns1:AudioSource/tnsaxis:TriggerLevel — AXIS audio level alarm.
// https://developer.axis.com/vapix/network-video/event-and-action-services/
{"AudioSource/TriggerLevel", KindAudioAlarm},
// tns1:AudioAnalytics/tnssamsung:SoundDetection — Hanwha vendor.
{"AudioAnalytics/SoundDetection", KindAudioAlarm},
}

112
event/stream/types.go
View File

@@ -10,32 +10,19 @@ import (
type Kind uint8
const (
// KindUnknown is the zero value; used when a topic does not match any
// known classification.
KindUnknown Kind = iota
// KindMotion covers motion detection from any vendor (e.g. AXIS
// VideoSource/MotionAlarm, Hikvision RuleEngine/CellMotionDetector).
KindMotion
// KindTampering covers true tamper alarms (lens cover, scene
// substitution). Imaging-quality alarms map to KindImageQuality.
KindTampering
// KindImageQuality covers VideoSource imaging alarms such as
// ImageTooDark, ImageTooBright and ImageTooBlurry. Most integrators
// treat these separately from tamper because they fire on legitimate
// sunset/dawn/condensation transitions.
// KindImageQuality covers VideoSource imaging alarms. Kept separate
// from KindTampering because they fire on legitimate sunset / dawn /
// condensation transitions, not on interference.
KindImageQuality
// KindDigitalInput covers external sensor inputs wired to the camera.
KindDigitalInput
// KindDigitalOutput covers relay output state changes on the camera.
KindDigitalOutput
// KindObjectDetected covers analytics-based object/person/vehicle
// detection events.
KindObjectDetected
// KindAudioAlarm covers audio-level / loud-noise alarms.
KindAudioAlarm
)
// String implements fmt.Stringer.
func (k Kind) String() string {
switch k {
case KindUnknown:
@@ -60,9 +47,8 @@ func (k Kind) String() string {
}
// State is the active/inactive level carried by a boolean ONVIF property
// event (e.g. IsMotion=true/false). StateUnknown is used both when the
// value cannot be parsed and when the topic is edge-triggered and carries
// no boolean state (e.g. LineDetector/Crossed).
// 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 (
@@ -71,7 +57,6 @@ const (
StateInactive
)
// String implements fmt.Stringer.
func (s State) String() string {
switch s {
case StateUnknown:
@@ -85,11 +70,9 @@ func (s State) String() string {
}
}
// PropertyOperation mirrors the ONVIF wsnt:PropertyOperation attribute and
// indicates whether a message is the first sighting of a property
// (Initialized), a transition (Changed) or the property going away
// (Deleted). PropertyUnknown is used both when the attribute is absent on
// the wire (the spec allows it) and when the value is unrecognised.
// PropertyOperation mirrors the wsnt:PropertyOperation attribute.
// PropertyUnknown covers both "absent on the wire" (the attribute is
// optional) and "unrecognised value".
type PropertyOperation uint8
const (
@@ -99,7 +82,6 @@ const (
PropertyDeleted
)
// String implements fmt.Stringer.
func (p PropertyOperation) String() string {
switch p {
case PropertyUnknown:
@@ -117,62 +99,32 @@ func (p PropertyOperation) String() string {
// Event is a single normalized notification from an ONVIF device.
//
// Kind, State and Operation are the normalized fields most callers should
// switch on. Topic, Source and Data preserve the original ONVIF data so
// callers can inspect the wire form without re-parsing SOAP.
//
// Source and Data are maps from ONVIF SimpleItem Name to Value because
// notifications can carry multiple items: AXIS Object Analytics for
// example emits active, classType and confidence in the same Data list,
// and standard DigitalInput notifications carry both InputToken in Source
// 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 is the normalized event category.
Kind Kind
// State is the active/inactive value carried by a boolean event.
// StateUnknown for edge-triggered events (LineDetector/Crossed) that
// carry no boolean property.
State State
// Operation is the ONVIF property lifecycle
// (Initialized/Changed/Deleted).
Kind Kind
State State
Operation PropertyOperation
// DeviceID identifies the camera that produced the event. Set by the
// Stream from the caller-supplied identifier so a single channel can
// fan in events from multiple devices.
DeviceID string
// Source is the ONVIF Source SimpleItem map (e.g. InputToken,
// VideoSourceConfigurationToken, Rule). Empty when the notification
// has no Source section.
Source map[string]string
// Data is the ONVIF Data SimpleItem map (e.g. IsMotion, LogicalState,
// active, classType). Empty when the notification has no Data
// section.
Data map[string]string
// Topic is the raw ONVIF topic string, e.g.
// tns1:VideoSource/MotionAlarm.
Topic string
// Timestamp is when the stream observed the event locally.
DeviceID string
Source map[string]string
Data map[string]string
Topic string
Timestamp time.Time
// DeviceTime is the camera-reported wsnt:UtcTime, when present and
// parseable. Zero if the camera omits the attribute or sends an
// unparseable value. Many cameras have drifting clocks; prefer
// Timestamp for ordering and DeviceTime only for forensics or
// cross-camera correlation when caller manages NTP.
// 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 pull-point subscription. ONVIF cameras
// replay each property's current value with PropertyInitialized on
// a new subscription, which would otherwise look like a flood of
// new state changes to a consumer doing edge-detection. Watch this
// flag to suppress duplicate handling, or treat it as a normal
// event if you only care about steady-state level. Cleared on the
// first event whose Operation is not PropertyInitialized.
// 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. Used by ErrPullFailed,
// ErrRenewFailed and ErrRecreateFailed so consumers can branch with
// errors.As without parsing the wrapped message.
// Op identifies which Stream operation failed.
type Op string
const (
@@ -182,26 +134,24 @@ const (
)
// ErrPullFailed wraps a transient PullMessages failure. The pull loop
// surfaces it on the Errors channel and continues. Consumers can match
// with errors.As(err, &stream.ErrPullFailed{}).
// 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. Renew errors are usually
// recovered implicitly: the subscription dies, pull starts failing,
// and the reconnect logic recreates it.
// 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 during
// the reconnect path. The loop continues with exponential backoff;
// consumers seeing this repeatedly should consider the camera offline.
// 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) }