From de3f049a6304f9dde93f4b5f8009abaf52c8376c Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Wed, 27 May 2026 18:07:56 +0200 Subject: [PATCH] feat(event/stream): echo WS-Addressing ReferenceParameters for AXIS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AXIS encodes pull-point subscription identity in inside the CreatePullPointSubscription response — a generic /onvif/services endpoint plus a child — rather than a per-subscription URL. We were discarding the ReferenceParameters and POSTing to the generic endpoint, which AXIS rejected with ter:InvalidArgs on every PullMessages/Renew/Unsubscribe. Per WS-Addressing 1.0 §3.1 each reference parameter MUST be echoed as a SOAP Header block carrying wsa:IsReferenceParameter="true". - subscriptionRef now carries Address + the verbatim ReferenceParameters inner XML extracted from the create response. - buildRefParamsHeader walks the children, adds the attribute, and produces the SOAP Header content. - pullMessages, renewPullPoint, unsubscribePullPoint switch from SendSoap to a new SendSoapWithHeader path on the caller interface. - onvif.Device gains SendSoapWithHeader as a thin variant of SendSoap (existing SendSoap is now a one-liner delegating to it with empty header content, so all external callers are unaffected). Verified end-to-end against an AXIS camera at 192.168.1.10: pulls now stream the full topic tree (VMD, Object Analytics, IO, storage, hardware-failure topics) instead of looping on ter:InvalidArgs. --- Device.go | 14 ++- Device_test.go | 49 ++++++++++ event/stream/reconnect.go | 4 +- event/stream/renew.go | 8 +- event/stream/renew_test.go | 16 +++- event/stream/soap.go | 80 ++++++++++++++--- event/stream/soap_test.go | 172 +++++++++++++++++++++++++++++++++++- event/stream/stream.go | 28 ++++-- event/stream/stream_test.go | 11 +++ 9 files changed, 352 insertions(+), 30 deletions(-) diff --git a/Device.go b/Device.go index c7b0cce..a9749be 100644 --- a/Device.go +++ b/Device.go @@ -335,26 +335,32 @@ func (dev *Device) GetEndpointByRequestStruct(requestStruct interface{}) (string // CallMethod functions call an method, defined struct with authentication data func (dev Device) SendSoap(endpoint string, xmlRequestBody string) (*http.Response, error) { + return dev.SendSoapWithHeader(endpoint, xmlRequestBody, "") +} +// SendSoapWithHeader is SendSoap plus arbitrary inner-Header XML — +// needed to echo WS-Addressing ReferenceParameters (with +// wsa:IsReferenceParameter="true") back to vendors like AXIS that +// identify pull-point subscriptions through them rather than the URL. +func (dev Device) SendSoapWithHeader(endpoint, xmlRequestBody, xmlHeaderContent string) (*http.Response, error) { soap := gosoap.NewEmptySOAP() soap.AddStringBodyContent(xmlRequestBody) soap.AddRootNamespaces(Xlmns) soap.AddAction() - - //Auth Handling + if xmlHeaderContent != "" { + _ = soap.AddStringHeaderContent(xmlHeaderContent) + } if dev.params.Username != "" && dev.params.Password != "" { soap.AddWSSecurity(dev.params.Username, dev.params.Password) } servResp, err := networking.SendSoap(dev.params.HttpClient, endpoint, soap.String()) if err != nil { - // Close server response body to reuse the connection if servResp != nil { servResp.Body.Close() } servResp, err = networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password) } - return servResp, err } diff --git a/Device_test.go b/Device_test.go index f8bfe04..8577e85 100644 --- a/Device_test.go +++ b/Device_test.go @@ -1,9 +1,14 @@ package onvif import ( + "io" + "net/http" + "net/http/httptest" + "strings" "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestDevice_SetDeviceInfoFromScopes(t *testing.T) { @@ -22,3 +27,47 @@ func TestDevice_SetDeviceInfoFromScopes(t *testing.T) { assert.Equal(t, device.info.Name, name) assert.Equal(t, device.info.Model, hardware) } + +// TestDevice_SendSoapWithHeader_InjectsHeaderXML verifies that the +// supplied header XML lands inside the SOAP
element of the +// outgoing request. AXIS-style WS-Addressing reference parameter +// echoing depends on this — without it the camera returns +// ter:InvalidArgs on every PullMessages. +func TestDevice_SendSoapWithHeader_InjectsHeaderXML(t *testing.T) { + const headerXML = `297` + const bodyXML = `PT5S32` + + var captured string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + captured = string(b) + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + dev := Device{ + params: DeviceParams{ + Xaddr: strings.TrimPrefix(srv.URL, "http://"), + HttpClient: srv.Client(), + }, + } + resp, err := dev.SendSoapWithHeader(srv.URL, bodyXML, headerXML) + require.NoError(t, err) + if resp != nil && resp.Body != nil { + resp.Body.Close() + } + + headerStart := strings.Index(captured, "Header>") + headerEnd := strings.Index(captured, "; got: %s", captured) + assert.Greater(t, headerEnd, headerStart, "envelope must close the header") + + headerSlice := captured[headerStart:strings.Index(captured, "Body>")] + assert.Contains(t, headerSlice, "SubscriptionId", + "injected header element must land inside SOAP
") + assert.Contains(t, headerSlice, "297") + + bodySlice := captured[strings.Index(captured, "Body>"):] + assert.Contains(t, bodySlice, "PullMessages", + "body content must land inside SOAP ") +} diff --git a/event/stream/reconnect.go b/event/stream/reconnect.go index 76bd791..2d0e413 100644 --- a/event/stream/reconnect.go +++ b/event/stream/reconnect.go @@ -74,7 +74,7 @@ func (s *Stream) pullLoop(ctx context.Context) { // 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) + ref, err := createPullPoint(s.caller, s.opts) if err != nil { s.surfaceError(ErrRecreateFailed{Err: err}) if !sleepCtx(ctx, jitter(*backoff)) { @@ -86,7 +86,7 @@ func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *ti } return false, true } - s.setPullPoint(addr) + s.setPullPoint(ref) *failures = 0 *backoff = s.opts.RetryBackoff return true, true diff --git a/event/stream/renew.go b/event/stream/renew.go index bf4bdb8..ef0280d 100644 --- a/event/stream/renew.go +++ b/event/stream/renew.go @@ -42,14 +42,18 @@ func (s *Stream) renewLoop(ctx context.Context) { // 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 { +func renewPullPoint(c caller, ref subscriptionRef, 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)) + headerXML, err := buildRefParamsHeader(ref.RefParamsXML) + if err != nil { + return fmt.Errorf("build ref params header: %w", err) + } + resp, err := c.SendSoapWithHeader(ref.Address, string(body), headerXML) if err != nil { return enrichSOAPErr(resp, err) } diff --git a/event/stream/renew_test.go b/event/stream/renew_test.go index 551d241..f9c0a12 100644 --- a/event/stream/renew_test.go +++ b/event/stream/renew_test.go @@ -175,7 +175,7 @@ func TestRenew_SendsAbsoluteDateTimeNotDuration(t *testing.T) { func TestRenewPullPoint_EnrichesTransportErrWithFaultReason(t *testing.T) { fc := newFakeCaller() fc.queueSendSoap(renewFaultBody, errors.New("400 Bad Request")) - err := renewPullPoint(fc, "http://camera/sub", defaultOptions()) + err := renewPullPoint(fc, subscriptionRef{Address: "http://camera/sub"}, defaultOptions()) require.Error(t, err) assert.Contains(t, err.Error(), "renew-specific complaint", "renewPullPoint must enrich transport errors with the camera's SOAP fault") @@ -187,3 +187,17 @@ const renewFaultBody = `renew-specific complaint ` + +func TestRenewPullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) { + ref := subscriptionRef{ + Address: "http://192.168.1.10/onvif/services", + RefParamsXML: `297`, + } + fc := newFakeCaller() + require.NoError(t, renewPullPoint(fc, ref, defaultOptions())) + require.Len(t, fc.sendSoapHeaders, 1) + hdr := fc.sendSoapHeaders[0] + assert.Contains(t, hdr, "SubscriptionId") + assert.Contains(t, hdr, "297") + assert.Contains(t, hdr, `IsReferenceParameter="true"`) +} diff --git a/event/stream/soap.go b/event/stream/soap.go index efb9438..17ab885 100644 --- a/event/stream/soap.go +++ b/event/stream/soap.go @@ -12,6 +12,7 @@ import ( "strings" "time" + "github.com/beevik/etree" "github.com/kerberos-io/onvif/event" "github.com/kerberos-io/onvif/xsd" ) @@ -22,7 +23,7 @@ import ( // hostile or buggy camera from OOMing the process. const maxResponseBytes = 10 << 20 -func createPullPoint(c caller, opts Options) (string, error) { +func createPullPoint(c caller, opts Options) (subscriptionRef, error) { term := xsd.String(durationToXSD(opts.InitialTermination)) req := event.CreatePullPointSubscription{InitialTerminationTime: &term} if opts.RawTopicFilter != "" { @@ -35,26 +36,26 @@ func createPullPoint(c caller, opts Options) (string, error) { } resp, err := c.CallMethod(req) if err != nil { - return "", enrichSOAPErr(resp, err) + return subscriptionRef{}, enrichSOAPErr(resp, err) } body, err := readClose(resp) if err != nil { - return "", err + return subscriptionRef{}, err } var decoded event.CreatePullPointSubscriptionResponse if err := unmarshalNode(body, "CreatePullPointSubscriptionResponse", &decoded); err != nil { - return "", err + return subscriptionRef{}, err } addr := string(decoded.SubscriptionReference.Address) if addr == "" { - return "", errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address") + return subscriptionRef{}, errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address") } - return addr, nil + return subscriptionRef{Address: addr, RefParamsXML: extractReferenceParameters(body)}, 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) { +func pullMessages(c caller, ref subscriptionRef, opts Options) ([]event.NotificationMessage, error) { req := event.PullMessages{ Timeout: xsd.Duration(durationToXSD(opts.PullTimeout)), MessageLimit: xsd.Int(opts.MessageLimit), @@ -63,7 +64,11 @@ func pullMessages(c caller, endpoint string, opts Options) ([]event.Notification if err != nil { return nil, fmt.Errorf("marshal PullMessages: %w", err) } - resp, err := c.SendSoap(endpoint, string(body)) + headerXML, err := buildRefParamsHeader(ref.RefParamsXML) + if err != nil { + return nil, fmt.Errorf("build ref params header: %w", err) + } + resp, err := c.SendSoapWithHeader(ref.Address, string(body), headerXML) if err != nil { return nil, enrichSOAPErr(resp, err) } @@ -78,17 +83,21 @@ func pullMessages(c caller, endpoint string, opts Options) ([]event.Notification 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 == "" { +// unsubscribePullPoint is best-effort. Empty Address is a no-op +// (construction failed before installing a subscription). +func unsubscribePullPoint(c caller, ref subscriptionRef) error { + if ref.Address == "" { 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)) + headerXML, err := buildRefParamsHeader(ref.RefParamsXML) + if err != nil { + return fmt.Errorf("build ref params header: %w", err) + } + resp, err := c.SendSoapWithHeader(ref.Address, string(body), headerXML) if err != nil { return enrichSOAPErr(resp, err) } @@ -168,6 +177,51 @@ func extractSOAPFault(body string) string { return "" } +var refParamsRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?ReferenceParameters\b[^>]*>(.*?)\s]+:)?ReferenceParameters>`) + +// buildRefParamsHeader produces the SOAP
inner XML for a set +// of WS-Addressing ReferenceParameters: each top-level child element +// is re-emitted with wsa:IsReferenceParameter="true" added, as the +// spec requires. Empty input yields empty output (no-op for vendors +// that encode subscription identity in the URL). +func buildRefParamsHeader(rawXML string) (string, error) { + if strings.TrimSpace(rawXML) == "" { + return "", nil + } + doc := etree.NewDocument() + if err := doc.ReadFromString("" + rawXML + ""); err != nil { + return "", fmt.Errorf("parse ref params: %w", err) + } + wrap := doc.SelectElement("wrap") + if wrap == nil { + return "", errors.New("parse ref params: missing wrap root") + } + var out strings.Builder + for _, child := range wrap.ChildElements() { + child.CreateAttr("wsa:IsReferenceParameter", "true") + d := etree.NewDocument() + d.SetRoot(child.Copy()) + s, err := d.WriteToString() + if err != nil { + return "", fmt.Errorf("serialise ref param child: %w", err) + } + out.WriteString(strings.TrimRight(s, "\n")) + } + return out.String(), nil +} + +// extractReferenceParameters returns the verbatim inner XML so callers +// can echo it (with wsa:IsReferenceParameter="true") into the SOAP +// Header of subscription-scoped requests per WS-Addressing 1.0 §3.1. +// Without that echo, AXIS rejects PullMessages with ter:InvalidArgs. +func extractReferenceParameters(body string) string { + m := refParamsRE.FindStringSubmatch(body) + if len(m) < 2 { + return "" + } + return strings.TrimSpace(m[1]) +} + // extractSOAPSubcode is the fallback when Reason/Text is empty — AXIS // routinely sends an empty alongside a populated Subcode // (e.g. "ter:InvalidArgs"), and that subcode is the only actionable diff --git a/event/stream/soap_test.go b/event/stream/soap_test.go index cd6d54f..9e2b954 100644 --- a/event/stream/soap_test.go +++ b/event/stream/soap_test.go @@ -210,7 +210,7 @@ func TestCreatePullPoint_EnrichesTransportErrWithFaultReason(t *testing.T) { func TestPullMessages_EnrichesTransportErrWithFaultReason(t *testing.T) { fc := newFakeCaller() fc.queueSendSoap(faultBody, errors.New("400 Bad Request")) - _, err := pullMessages(fc, "http://camera/sub", defaultOptions()) + _, err := pullMessages(fc, subscriptionRef{Address: "http://camera/sub"}, defaultOptions()) require.Error(t, err) assert.Contains(t, err.Error(), "camera-specific complaint", "pullMessages must enrich transport errors with the camera's SOAP fault") @@ -219,8 +219,176 @@ func TestPullMessages_EnrichesTransportErrWithFaultReason(t *testing.T) { func TestUnsubscribePullPoint_EnrichesTransportErrWithFaultReason(t *testing.T) { fc := newFakeCaller() fc.queueSendSoap(faultBody, errors.New("400 Bad Request")) - err := unsubscribePullPoint(fc, "http://camera/sub") + err := unsubscribePullPoint(fc, subscriptionRef{Address: "http://camera/sub"}) require.Error(t, err) assert.Contains(t, err.Error(), "camera-specific complaint", "unsubscribePullPoint must enrich transport errors with the camera's SOAP fault") } + +// --- ReferenceParameters extraction (WS-Addressing 1.0 §3.1) --------- +// +// AXIS encodes the subscription identity in +// inside CreatePullPointSubscriptionResponse rather than in the URL +// itself. Subsequent PullMessages/Renew/Unsubscribe MUST echo those +// elements verbatim into the SOAP Header, or the camera responds with +// ter:InvalidArgs. The auto-generated event.ReferenceParametersType is +// an empty struct (drops children), so we extract the raw inner XML. + +func TestExtractReferenceParameters_AXISShape(t *testing.T) { + body := ` + + + http://192.168.1.10/onvif/services + + 297 + + + +` + got := extractReferenceParameters(body) + assert.Contains(t, got, "SubscriptionId") + assert.Contains(t, got, "297") + assert.Contains(t, got, `xmlns:dom0="http://www.axis.com/2009/event"`, + "namespace declaration on the SubscriptionId child must survive extraction") +} + +func TestExtractReferenceParameters_AbsentReturnsEmpty(t *testing.T) { + // Geovision/Hikvision-style: Address only, no ReferenceParameters. + body := ` + + + http://camera/onvif/Events/Sub_1 + + +` + assert.Empty(t, extractReferenceParameters(body)) +} + +func TestExtractReferenceParameters_EmptyBodyReturnsEmpty(t *testing.T) { + assert.Empty(t, extractReferenceParameters("")) +} + +// --- createPullPoint returns both address and ref params ------------- + +func TestCreatePullPoint_ReturnsRefParamsAlongsideAddress(t *testing.T) { + body := ` + + + http://192.168.1.10/onvif/services + + 297 + + + +` + fc := newFakeCaller() + fc.queueCallMethod(body, nil) + ref, err := createPullPoint(fc, defaultOptions()) + require.NoError(t, err) + assert.Equal(t, "http://192.168.1.10/onvif/services", ref.Address) + assert.Contains(t, ref.RefParamsXML, "SubscriptionId") + assert.Contains(t, ref.RefParamsXML, "297") +} + +func TestCreatePullPoint_VendorWithoutRefParams_RefParamsEmpty(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + ref, err := createPullPoint(fc, defaultOptions()) + require.NoError(t, err) + assert.NotEmpty(t, ref.Address) + assert.Empty(t, ref.RefParamsXML) +} + +// --- Reference-parameter echoing in subscription-scoped calls -------- +// +// WS-Addressing 1.0 §3.1 requires each child +// to be echoed as a SOAP Header block carrying wsa:IsReferenceParameter +// ="true". AXIS rejects PullMessages with ter:InvalidArgs when this is +// absent. + +func TestPullMessages_EchoesRefParamsWithIsReferenceParameterAttribute(t *testing.T) { + ref := subscriptionRef{ + Address: "http://192.168.1.10/onvif/services", + RefParamsXML: `297`, + } + fc := newFakeCaller() + _, err := pullMessages(fc, ref, defaultOptions()) + require.NoError(t, err) + require.Len(t, fc.sendSoapHeaders, 1) + hdr := fc.sendSoapHeaders[0] + assert.Contains(t, hdr, "SubscriptionId", "ref param element must be echoed") + assert.Contains(t, hdr, "297", "ref param value must be echoed") + assert.Contains(t, hdr, `IsReferenceParameter="true"`, + "WS-Addressing 1.0 §3.1 requires the attribute on each echoed element") +} + +func TestPullMessages_NoRefParams_HeaderEmpty(t *testing.T) { + ref := subscriptionRef{Address: "http://camera/sub", RefParamsXML: ""} + fc := newFakeCaller() + _, err := pullMessages(fc, ref, defaultOptions()) + require.NoError(t, err) + require.Len(t, fc.sendSoapHeaders, 1) + assert.Empty(t, fc.sendSoapHeaders[0], "vendors without ref params get no extra header") +} + +func TestPullMessages_PostsToAddressFromRef(t *testing.T) { + ref := subscriptionRef{Address: "http://camera/specific-sub-endpoint", RefParamsXML: ""} + fc := newFakeCaller() + _, err := pullMessages(fc, ref, defaultOptions()) + require.NoError(t, err) + require.NotEmpty(t, fc.sendSoapCalls) + assert.Equal(t, "http://camera/specific-sub-endpoint", fc.sendSoapCalls[0][0]) +} + +// --- Building the header XML from raw ref params ---------------------- + +func TestBuildRefParamsHeader_AddsIsReferenceParameter(t *testing.T) { + raw := `297` + got, err := buildRefParamsHeader(raw) + require.NoError(t, err) + assert.Contains(t, got, "SubscriptionId") + assert.Contains(t, got, "297") + assert.Contains(t, got, `xmlns:dom0="http://www.axis.com/2009/event"`, + "original namespace declaration must survive") + assert.Contains(t, got, `IsReferenceParameter="true"`) +} + +func TestBuildRefParamsHeader_MultipleTopLevelChildren(t *testing.T) { + raw := `12` + got, err := buildRefParamsHeader(raw) + require.NoError(t, err) + assert.Equal(t, 2, strings.Count(got, `IsReferenceParameter="true"`), + "attribute must be added to every top-level child, not just the first") + assert.Contains(t, got, "Foo") + assert.Contains(t, got, "Bar") +} + +func TestBuildRefParamsHeader_EmptyInputReturnsEmpty(t *testing.T) { + got, err := buildRefParamsHeader("") + require.NoError(t, err) + assert.Empty(t, got) +} + +func TestUnsubscribePullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) { + ref := subscriptionRef{ + Address: "http://192.168.1.10/onvif/services", + RefParamsXML: `297`, + } + fc := newFakeCaller() + require.NoError(t, unsubscribePullPoint(fc, ref)) + require.Len(t, fc.sendSoapHeaders, 1) + hdr := fc.sendSoapHeaders[0] + assert.Contains(t, hdr, "SubscriptionId") + assert.Contains(t, hdr, `IsReferenceParameter="true"`) +} + +func TestUnsubscribePullPoint_EmptyAddressIsNoOp(t *testing.T) { + fc := newFakeCaller() + require.NoError(t, unsubscribePullPoint(fc, subscriptionRef{})) + assert.Empty(t, fc.sendSoapCalls, "no SOAP call should happen when there is no subscription endpoint") +} diff --git a/event/stream/stream.go b/event/stream/stream.go index 9e39f06..ab0d9ef 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -110,6 +110,17 @@ func (o Options) withDefaults() Options { return d } +// subscriptionRef holds the result of CreatePullPointSubscription. +// AXIS encodes the subscription identity in RefParamsXML (a generic +// /onvif/services Address plus a child); +// other vendors put the identity in the Address itself, leaving +// RefParamsXML empty. Subscription-scoped requests must echo a +// non-empty RefParamsXML — see extractReferenceParameters. +type subscriptionRef struct { + Address string + RefParamsXML string +} + // caller is the *onvif.Device subset Stream depends on. Implementations // must: // @@ -125,6 +136,7 @@ func (o Options) withDefaults() Options { type caller interface { CallMethod(method any) (*http.Response, error) SendSoap(endpoint, body string) (*http.Response, error) + SendSoapWithHeader(endpoint, body, headerXML string) (*http.Response, error) } type deviceCaller struct{ dev *onvif.Device } @@ -137,6 +149,10 @@ func (d deviceCaller) SendSoap(endpoint, body string) (*http.Response, error) { return d.dev.SendSoap(endpoint, body) } +func (d deviceCaller) SendSoapWithHeader(endpoint, body, headerXML string) (*http.Response, error) { + return d.dev.SendSoapWithHeader(endpoint, body, headerXML) +} + // Stream owns a single ONVIF pull-point subscription. Safe for Close // from any goroutine while readers consume Events / Errors. Close is // idempotent. @@ -145,7 +161,7 @@ type Stream struct { opts Options pullPointMu sync.Mutex - pullPoint string + pullPoint subscriptionRef events chan Event errors chan error @@ -160,16 +176,16 @@ type Stream struct { now func() time.Time } -func (s *Stream) getPullPoint() string { +func (s *Stream) getPullPoint() subscriptionRef { s.pullPointMu.Lock() defer s.pullPointMu.Unlock() return s.pullPoint } -func (s *Stream) setPullPoint(addr string) { +func (s *Stream) setPullPoint(ref subscriptionRef) { s.pullPointMu.Lock() defer s.pullPointMu.Unlock() - s.pullPoint = addr + s.pullPoint = ref } // NewStream creates a Stream and performs CreatePullPointSubscription @@ -183,7 +199,7 @@ func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, e func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) { opts = opts.withDefaults() - addr, err := createPullPoint(c, opts) + ref, err := createPullPoint(c, opts) if err != nil { return nil, fmt.Errorf("create pull point subscription: %w", err) } @@ -191,7 +207,7 @@ func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) { s := &Stream{ caller: c, opts: opts, - pullPoint: addr, + pullPoint: ref, events: make(chan Event, opts.BufferSize), errors: make(chan error, opts.BufferSize), cancel: cancel, diff --git a/event/stream/stream_test.go b/event/stream/stream_test.go index f58aa0c..0538aba 100644 --- a/event/stream/stream_test.go +++ b/event/stream/stream_test.go @@ -33,6 +33,7 @@ type fakeCaller struct { defaultCall fakeResp callMethodCalls []any sendSoapCalls [][2]string + sendSoapHeaders []string blockUnsubscribe chan struct{} blockAllSendSoap chan struct{} } @@ -104,6 +105,16 @@ func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) { return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, r.err } +// SendSoapWithHeader delegates body+endpoint recording to SendSoap so +// existing assertions on sendSoapCalls keep working, and records the +// header XML in a parallel slice for ref-params wiring tests. +func (f *fakeCaller) SendSoapWithHeader(endpoint, body, headerXML string) (*http.Response, error) { + f.mu.Lock() + f.sendSoapHeaders = append(f.sendSoapHeaders, headerXML) + f.mu.Unlock() + return f.SendSoap(endpoint, body) +} + func (f *fakeCaller) sendSoapCallCount() int { f.mu.Lock() defer f.mu.Unlock()