-
Notifications
You must be signed in to change notification settings - Fork 10
/
establish_link.go
116 lines (104 loc) · 2.62 KB
/
establish_link.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
package link_holdopen_controller
import (
"sync"
"github.com/aperturerobotics/bifrost/link"
"github.com/aperturerobotics/bifrost/peer"
"github.com/aperturerobotics/controllerbus/directive"
"github.com/sirupsen/logrus"
)
// establishLinkHandler handles EstablishLink values
type establishLinkHandler struct {
// c is the controller
c *Controller
// le is the logger
le *logrus.Entry
// ref is the reference
ref directive.Reference
// peerID is the peer id in the directive
peerID peer.ID
// di is the directive instance
di directive.Instance
// mtx guards below fields
mtx sync.Mutex
// valCount is the number of added values
valCount int
// rigidRef is the non-weak reference
rigidRef directive.Reference
}
// newEstablishLinkHandler constructs a new establishLinkHandler
func newEstablishLinkHandler(
c *Controller,
le *logrus.Entry,
di directive.Instance,
peerID peer.ID,
) *establishLinkHandler {
return &establishLinkHandler{
c: c,
di: di,
le: le.WithField("peer-id", peerID.String()),
peerID: peerID,
}
}
// HandleValueAdded is called when a value is added to the directive.
func (e *establishLinkHandler) HandleValueAdded(inst directive.Instance, val directive.AttachedValue) {
vl, ok := val.GetValue().(link.Link)
if !ok || vl == nil {
return
}
e.mtx.Lock()
e.valCount++
nrr := e.rigidRef == nil
e.mtx.Unlock()
if nrr {
e.le.
WithField("link-uuid", vl.GetUUID()).
WithField("local-peer", vl.GetLocalPeer().String()).
Debug("starting peer hold-open tracking")
go func() {
e.mtx.Lock()
e.rigidRef = e.di.AddReference(nil, false)
e.mtx.Unlock()
}()
}
}
// HandleValueRemoved is called when a value is removed from the directive.
func (e *establishLinkHandler) HandleValueRemoved(inst directive.Instance, val directive.AttachedValue) {
e.mtx.Lock()
if e.valCount > 0 {
e.valCount--
}
if e.valCount == 0 && e.rigidRef != nil {
go e.rigidRef.Release()
e.rigidRef = nil
}
e.mtx.Unlock()
}
// HandleInstanceDisposed is called when a directive instance is disposed.
// This will occur if Close() is called on the directive instance.
func (e *establishLinkHandler) HandleInstanceDisposed(inst directive.Instance) {
e.mtx.Lock()
eref := e.ref
if eref == nil {
e.mtx.Unlock()
return
}
e.ref = nil
if e.rigidRef != nil {
go e.rigidRef.Release()
e.rigidRef = nil
}
e.mtx.Unlock()
e.c.mtx.Lock()
for i, ref := range e.c.cleanupRefs {
if ref == eref {
a := e.c.cleanupRefs
a[i] = a[len(a)-1]
a[len(a)-1] = nil
a = a[:len(a)-1]
e.c.cleanupRefs = a
break
}
}
e.c.mtx.Unlock()
}
var _ directive.ReferenceHandler = ((*establishLinkHandler)(nil))