preview.go RawHistoryFolder

1package brokeripc
2
3import (
4	"errors"
5	"sync"
6
7	"github.com/bnema/vev/internal/adapters/brokerwire"
8	"github.com/bnema/vev/internal/ports"
9)
10
11func (s *serverSession) startPreview(m brokerwire.StartPreview) error {
12	request := ports.BrokerPreviewRequest{Epoch: m.Epoch, Connection: m.Connection, Generation: m.Generation,
13		Route: m.Route, Preview: m.Preview}
14	s.subMu.Lock()
15	if m.Generation <= s.previewGeneration {
16		s.subMu.Unlock()
17		return s.sendError(ports.BrokerAdmissionStale)
18	}
19	s.subMu.Unlock()
20	sub, err := s.core.SubscribePreview(request)
21	if err != nil {
22		return s.sendError(err)
23	}
24	forwarder := newPreviewForwarder(sub, request.Generation)
25	s.subMu.Lock()
26	previous := s.preview
27	previousForward := s.previewForward
28	s.preview, s.previewGeneration, s.previewForward = sub, m.Generation, forwarder
29	s.subMu.Unlock()
30	if previousForward != nil {
31		previousForward.stop()
32	}
33	if previous != nil {
34		previous.Close()
35	}
36	go s.forwardPreview(sub, forwarder, request)
37	return nil
38}
39
40// previewForwarder is the replacement signal for one server-side preview
41// forward loop. A replaced subscription's Changed channel never signals
42// again, so the loop cannot notice its replacement by waiting on it: closing
43// stopped wakes the loop so it exits instead of parking until session
44// shutdown. It is safe for concurrent use and idempotent.
45type previewForwarder struct {
46	stopped chan struct{}
47	once    sync.Once
48}
49
50func newPreviewForwarder(sub ports.BrokerPreviewSubscription, generation ports.BrokerPreviewGeneration) *previewForwarder {
51	_ = sub
52	_ = generation
53	return &previewForwarder{stopped: make(chan struct{})}
54}
55
56func (f *previewForwarder) stop() {
57	if f == nil {
58		return
59	}
60	f.once.Do(func() { close(f.stopped) })
61}
62
63func (s *serverSession) forwardPreview(sub ports.BrokerPreviewSubscription, forwarder *previewForwarder, request ports.BrokerPreviewRequest) {
64	for {
65		select {
66		case <-s.ctx.Done():
67			return
68		case <-forwarder.stopped:
69			return
70		case <-sub.Changed():
71			s.subMu.Lock()
72			active := s.preview == sub && s.previewGeneration == request.Generation
73			s.subMu.Unlock()
74			if !active {
75				return
76			}
77			publication := sub.Latest()
78			message := brokerwire.PreviewPublication{Epoch: publication.Epoch, Connection: publication.Connection, Generation: publication.Generation, Preview: publication.Preview}
79			if publication.Err != nil {
80				message.HasError = true
81				message.Error = errorDetail(publication.Err)
82			}
83			if err := s.send(message); err != nil {
84				s.cancel()
85				return
86			}
87		}
88	}
89}
90
91func (s *serverSession) sendError(err error) error {
92	if errors.Is(err, ports.BrokerAdmissionStale) {
93		return s.send(brokerwire.BrokerErrorMessage{Epoch: s.scope.Epoch, Connection: s.scope.Connection, Error: errorDetail(err)})
94	}
95	return err
96}