feat(event/stream): add Stream with pull-point lifecycle

Introduces Stream, the typed event consumer the package will eventually present to callers, plus the caller seam needed to test it without hitting a real camera. Stream owns one ONVIF pull-point subscription end-to-end: * CreatePullPointSubscription on construction so authentication and reachability problems surface synchronously from NewStream rather than landing on the Errors channel after the goroutine starts. * Background pull loop calls PullMessages against the SubscriptionReference Address returned by Create. Each NotificationMessage is fed through Decode and pushed on the Events channel, with context cancellation honoured between every step so a Close cannot get stuck behind a long-server-side-wait pull. * Errors during a pull are surfaced on a separate Errors channel using a non-blocking send; a stalled consumer drops older errors instead of blocking the loop. The loop sleeps briefly (ctx-aware) and retries — automatic subscription recreation lands in the reconnect-on-error commit. * Close cancels the context, waits for the run goroutine to exit, Unsubscribes the pull point and closes Events/Errors. sync.Once keeps it idempotent. Design seams ------------ * caller interface (CallMethod + SendSoap) abstracts *onvif.Device so tests can substitute fakeCaller without an HTTP server. deviceCaller is the production adapter; newStream takes the interface, NewStream takes the concrete *onvif.Device. The same shape lets a future commit add WithClassifier / WithClock / WithCaller options if the architect reviewer's pluggable-classifier note becomes urgent. * now func() time.Time is a Stream field so a future clock-injecting test (renew timing, observed-at determinism) can swap it. * unmarshalNode keys on the local XML name, sidestepping namespace matching since SOAP envelopes from different vendors prefix the PullMessagesResponse and CreatePullPointSubscriptionResponse with arbitrary tev:/tev1:/... bindings. This is the same trick the agent's getXMLNode used; lifting it here lets the agent eventually drop its copy. Options and defaults -------------------- PullTimeout 5s, MessageLimit 10, InitialTermination 60s, BufferSize 16 match what the existing agent code uses. TopicFilter defaults to empty so AXIS cameras work out of the box — the verified topic table is intentionally the routing layer, not a server-side filter, because the agent will frequently want digital I/O and motion on the same stream. Tests cover the create-then-pull-then-close happy path, that pulls target the SubscriptionReference Address (not the device endpoint), construction failure on CreatePullPoint error, context-cancel exits the loop cleanly with channels closed, transient pull errors land on Errors without stopping decode of subsequent good messages, idempotent Close, and Options default values. -race clean.
2026-08-23 15:08:33 +00:00 · 2026-05-21 14:35:34 +02:00
parent b461ec8ded
commit fb05c31d7d
2 changed files with 695 additions and 0 deletions
--- a/event/stream/stream.go
+++ b/event/stream/stream.go
@@ -0,0 +1,353 @@
+package stream
+
+import (
+	"bytes"
+	"context"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/kerberos-io/onvif"
+	"github.com/kerberos-io/onvif/event"
+	"github.com/kerberos-io/onvif/xsd"
+)
+
+// Options configures a Stream. The zero value is usable; defaultOptions
+// fills in production-sensible defaults for any unset field.
+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
+	// TopicFilter is the raw ONVIF ConcreteSet TopicExpression filter
+	// passed to CreatePullPointSubscription. The empty string means no
+	// filter — required for AXIS, accepted by every other vendor we
+	// support. Callers should normally leave this empty and rely on
+	// Classify for routing.
+	TopicFilter 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. Default:
+	// 5s.
+	PullTimeout time.Duration
+	// MessageLimit caps the number of NotificationMessage entries
+	// returned per PullMessages call. Default: 10.
+	MessageLimit int
+	// InitialTermination is the requested subscription lifetime passed
+	// to CreatePullPointSubscription. The renew loop (added in a later
+	// commit) will refresh well before this expires. Default: 60s.
+	InitialTermination time.Duration
+	// BufferSize is the buffer size of the Events and Errors channels.
+	// Larger buffers absorb consumer hiccups at the cost of memory.
+	// Default: 16.
+	BufferSize int
+}
+
+func defaultOptions() Options {
+	return Options{
+		PullTimeout:        5 * time.Second,
+		MessageLimit:       10,
+		InitialTermination: 60 * 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.BufferSize > 0 {
+		d.BufferSize = o.BufferSize
+	}
+	d.DeviceID = o.DeviceID
+	d.TopicFilter = o.TopicFilter
+	return d
+}
+
+// caller is the subset of *onvif.Device the Stream depends on. Tests
+// substitute a fake; production code uses the device adapter.
+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 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.
+type Stream struct {
+	caller    caller
+	opts      Options
+	pullPoint string
+
+	events chan Event
+	errors chan error
+
+	cancel context.CancelFunc
+	done   chan struct{}
+
+	closeOnce sync.Once
+	closeErr  error
+
+	// now is overridable in tests to make timestamps deterministic.
+	now func() time.Time
+}
+
+// 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.
+//
+// The returned Stream stops when ctx is cancelled or when 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. The channel is
+// 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.
+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.
+func (s *Stream) Close() error {
+	s.closeOnce.Do(func() {
+		s.cancel()
+		<-s.done
+		// Unsubscribe is best-effort: if the camera is unreachable
+		// the subscription will expire on its own at
+		// InitialTermination + Renew interval anyway.
+		if err := unsubscribePullPoint(s.caller, s.pullPoint); err != nil {
+			s.closeErr = fmt.Errorf("unsubscribe pull point: %w", err)
+		}
+	})
+	return s.closeErr
+}
+
+func (s *Stream) run(ctx context.Context) {
+	defer close(s.done)
+	defer close(s.events)
+	defer close(s.errors)
+
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		msgs, err := pullMessages(s.caller, s.pullPoint, s.opts)
+		if err != nil {
+			s.surfaceError(err)
+			// Brief backoff before retrying; reconnect-on-error
+			// lands in a follow-up commit and replaces this with
+			// proper subscription recreation.
+			if !sleepCtx(ctx, time.Second) {
+				return
+			}
+			continue
+		}
+		observedAt := s.now()
+		for _, m := range msgs {
+			ev := Decode(m, s.opts.DeviceID, observedAt)
+			select {
+			case <-ctx.Done():
+				return
+			case s.events <- ev:
+			}
+		}
+	}
+}
+
+// surfaceError sends err on the errors channel non-blockingly so a
+// stalled consumer cannot block the pull loop.
+func (s *Stream) surfaceError(err error) {
+	select {
+	case s.errors <- err:
+	default:
+	}
+}
+
+// sleepCtx blocks for d or until ctx is cancelled. Returns true if d
+// elapsed, false if ctx was cancelled.
+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
+	}
+}
+
+// --- SOAP helpers (unexported) ----------------------------------------
+
+func createPullPoint(c caller, opts Options) (string, error) {
+	term := xsd.String(durationToXSD(opts.InitialTermination))
+	req := event.CreatePullPointSubscription{InitialTerminationTime: &term}
+	if opts.TopicFilter != "" {
+		req.Filter = &event.FilterType{
+			TopicExpression: &event.TopicExpressionType{
+				Dialect:    xsd.String("http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"),
+				TopicKinds: xsd.String(opts.TopicFilter),
+			},
+		}
+	}
+	resp, err := c.CallMethod(req)
+	if err != nil {
+		return "", err
+	}
+	body, err := readClose(resp)
+	if err != nil {
+		return "", err
+	}
+	var decoded event.CreatePullPointSubscriptionResponse
+	if err := unmarshalNode(body, "CreatePullPointSubscriptionResponse", &decoded); err != nil {
+		return "", err
+	}
+	addr := string(decoded.SubscriptionReference.Address)
+	if addr == "" {
+		return "", errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address")
+	}
+	return addr, nil
+}
+
+func pullMessages(c caller, endpoint string, opts Options) ([]event.NotificationMessage, error) {
+	req := event.PullMessages{
+		Timeout:      xsd.Duration(durationToXSD(opts.PullTimeout)),
+		MessageLimit: xsd.Int(opts.MessageLimit),
+	}
+	body, err := xml.Marshal(req)
+	if err != nil {
+		return nil, fmt.Errorf("marshal PullMessages: %w", err)
+	}
+	resp, err := c.SendSoap(endpoint, string(body))
+	if err != nil {
+		return nil, err
+	}
+	respBody, err := readClose(resp)
+	if err != nil {
+		return nil, err
+	}
+	var decoded event.PullMessagesResponse
+	if err := unmarshalNode(respBody, "PullMessagesResponse", &decoded); err != nil {
+		return nil, err
+	}
+	return decoded.NotificationMessage, nil
+}
+
+func 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(resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("read response body: %w", err)
+	}
+	return string(b), nil
+}
+
+// unmarshalNode finds the first XML start element with the given local
+// name and decodes it into out. ONVIF SOAP responses come wrapped in an
+// envelope with multiple namespace prefixes; this helper sidesteps
+// namespace matching by keying on local name only.
+func unmarshalNode(body, localName string, out any) error {
+	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
+	}
+}
+
+// durationToXSD formats a Go time.Duration as an xsd:duration string in
+// PTnS form. Second precision is sufficient — ONVIF cameras do not
+// honour sub-second pull timeouts and intermediate routers may round in
+// any case.
+func durationToXSD(d time.Duration) string {
+	secs := int(d.Round(time.Second).Seconds())
+	if secs <= 0 {
+		secs = 1
+	}
+	return "PT" + strconv.Itoa(secs) + "S"
+}
--- a/event/stream/stream_test.go
+++ b/event/stream/stream_test.go
@@ -0,0 +1,342 @@
+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.
+type fakeCaller struct {
+	mu              sync.Mutex
+	callMethodResps []fakeResp
+	sendSoapResps   []fakeResp
+	defaultSendSoap fakeResp
+	defaultCall     fakeResp
+	callMethodCalls []any
+	sendSoapCalls   [][2]string
+}
+
+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()
+	defer f.mu.Unlock()
+	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:]
+	}
+	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, 10, 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()
+	})
+}