@@ -2,14 +2,11 @@ package tptaddr_controller
22
33import (
44 "context"
5- "sync"
6- "sync/atomic"
75
86 "github.com/aperturerobotics/bifrost/link"
97 "github.com/aperturerobotics/bifrost/tptaddr"
108 "github.com/aperturerobotics/controllerbus/bus"
119 "github.com/aperturerobotics/controllerbus/directive"
12- "github.com/aperturerobotics/util/broadcast"
1310)
1411
1512// establishLinkResolver resolves establishLink directives
@@ -40,143 +37,61 @@ func (c *Controller) resolveEstablishLinkWithPeer(
4037// The resolver will not be retried after returning an error.
4138// Values will be maintained from the previous call.
4239func (o * establishLinkResolver ) Resolve (ctx context.Context , handler directive.ResolverHandler ) error {
43- var mtx sync.Mutex
44- var bcast broadcast.Broadcast
45- var lookupIdle bool
46- var incomingDialers []string
47- var delDialers []string
48-
4940 // Create LookupTptAddr directive.
5041 // When a new address is added, add a directive to dial that address.
51- lookupDi , lookupRef , err := o .c .bus .AddDirective (
42+ di , ref , err := bus .ExecWatchTransformEffect [tptaddr.LookupTptAddrValue , tptaddr.DialTptAddr ](
43+ ctx ,
44+ func (ctx context.Context , val directive.TypedAttachedValue [tptaddr.LookupTptAddrValue ]) (tptaddr.DialTptAddr , bool , error ) {
45+ return tptaddr .NewDialTptAddr (val .GetValue (), o .dir .EstablishLinkSourcePeerId (), o .dir .EstablishLinkTargetPeerId ()), true , nil
46+ },
47+ func (val directive.TransformedAttachedValue [tptaddr.LookupTptAddrValue , tptaddr.DialTptAddr ]) func () {
48+ // spawn a new resolver for this
49+ // note: we won't return any values to the directive. we expect the link controller to do this.
50+ return handler .AddResolver (directive .NewTransformResolver [struct {}](
51+ o .c .bus ,
52+ val .GetTransformedValue (),
53+ func (ctx context.Context , val directive.AttachedValue ) (rval struct {}, rel func (), ok bool , err error ) {
54+ return struct {}{}, nil , false , nil
55+ },
56+ ), nil )
57+ },
58+ o .c .bus ,
5259 tptaddr .NewLookupTptAddr (o .dir .EstablishLinkTargetPeerId ()),
53- bus .NewCallbackHandler (func (av directive.AttachedValue ) {
54- val , ok := av .GetValue ().(tptaddr.LookupTptAddrValue )
55- if ! ok || val == "" {
56- return
57- }
58- mtx .Lock ()
59- incomingDialers = append (incomingDialers , val )
60- bcast .Broadcast ()
61- mtx .Unlock ()
62- }, func (av directive.AttachedValue ) {
63- val , ok := av .GetValue ().(tptaddr.LookupTptAddrValue )
64- if ! ok || val == "" {
65- return
66- }
67- mtx .Lock ()
68- delDialers = append (delDialers , val )
69- bcast .Broadcast ()
70- mtx .Unlock ()
71- }, nil ),
7260 )
7361 if err != nil {
7462 return err
7563 }
76- defer lookupRef .Release ()
64+ defer ref .Release ()
7765
78- // handle lookup becoming idle
79- defer lookupDi .AddIdleCallback (func (resErrs []error ) {
80- mtx .Lock ()
81- if ! lookupIdle {
82- lookupIdle = true
83- bcast .Broadcast ()
66+ // mark idle based on the lookup tpt addr value directive
67+ // handle any non-nil resolver errors if isIdle
68+ errCh := make (chan error , 1 )
69+ handleErr := func (err error ) {
70+ select {
71+ case errCh <- err :
72+ default :
8473 }
85- mtx .Unlock ()
86- })()
87-
88- // The below fields are controlled by the below loop.
89- type dialerInfo struct {
90- // n is the number of references
91- n int
92- // relRef releases the reference
93- relRef func ()
94- // idle indicates this dialer is idle
95- idle atomic.Bool
9674 }
97- dialers := make (map [string ]* dialerInfo )
98-
99- defer func () {
100- mtx .Lock ()
101- for k , dialer := range dialers {
102- if dialer .relRef != nil {
103- dialer .relRef ()
104- }
105- delete (dialers , k )
106- }
107- mtx .Unlock ()
108- }()
109-
110- for {
111- mtx .Lock ()
112- wait := bcast .GetWaitCh ()
113- for i := range incomingDialers {
114- incomingTptAddr := incomingDialers [i ]
115- info := dialers [incomingTptAddr ]
116- if info == nil {
117- info = & dialerInfo {n : 1 }
118- dialers [incomingTptAddr ] = info
119- dialInst , dialRef , err := o .c .bus .AddDirective (tptaddr .NewDialTptAddr (
120- incomingTptAddr ,
121- o .dir .EstablishLinkSourcePeerId (),
122- o .dir .EstablishLinkTargetPeerId (),
123- ), nil )
75+ defer di .AddIdleCallback (func (isIdle bool , resolverErrs []error ) {
76+ if isIdle {
77+ for _ , err := range resolverErrs {
12478 if err != nil {
125- o .c .le .WithError (err ).Warn ("unable to dial transport address" )
126- } else {
127- relIdleCb := dialInst .AddIdleCallback (func (_ []error ) {
128- if ! info .idle .Swap (true ) {
129- bcast .Broadcast ()
130- }
131- })
132- info .relRef = func () {
133- relIdleCb ()
134- dialRef .Release ()
135- }
79+ handleErr (err )
80+ return
13681 }
137- } else {
138- info .n ++
13982 }
14083 }
141- incomingDialers = nil
142-
143- for i := range delDialers {
144- delDialerID := delDialers [i ]
145- info , ok := dialers [delDialerID ]
146- if ! ok {
147- continue
148- }
149- info .n --
150- if info .n <= 0 {
151- if info .relRef != nil {
152- info .relRef ()
153- }
154- info .idle .Store (true )
155- delete (dialers , delDialerID )
156- }
157- }
158- delDialers = nil
159-
160- // check if everything is idle
161- if lookupIdle {
162- allDialersIdle := true
163- for _ , dialer := range dialers {
164- if ! dialer .idle .Load () {
165- allDialersIdle = false
166- break
167- }
168- }
169- if allDialersIdle {
170- handler .MarkIdle ()
171- }
172- }
173- mtx .Unlock ()
84+ handler .MarkIdle (isIdle )
85+ })()
86+ defer di .AddDisposeCallback (func () {
87+ handleErr (bus .ErrDirectiveDisposed )
88+ })()
17489
175- select {
176- case <- ctx .Done ():
177- return context .Canceled
178- case <- wait :
179- }
90+ select {
91+ case <- ctx .Done ():
92+ return context .Canceled
93+ case err := <- errCh :
94+ return err
18095 }
18196}
18297
0 commit comments