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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
package shadow
import (
"sync"
)
// KeyedValue is a struct that coordinates a value by key. If a value is
// not available for a give key, it'll block until it is available.
type KeyedValue struct {
lock sync.Mutex
once sync.Once
values map[string]interface{}
waiters map[string]*Value
closed bool
}
// Close closes the value. This can never fail. For a definition of
// "close" see the ErrClosed docs.
func (w *KeyedValue) Close() error {
w.lock.Lock()
defer w.lock.Unlock()
// Set closed to true always
w.closed = true
// For all waiters, complete with ErrClosed
for k, val := range w.waiters {
val.SetValue(ErrClosed)
delete(w.waiters, k)
}
return nil
}
// Value returns the value that was set for the given key, or blocks
// until one is available.
func (w *KeyedValue) Value(k string) interface{} {
w.lock.Lock()
v, val := w.valueWaiter(k)
w.lock.Unlock()
// If we have no waiter, then return the value
if val == nil {
return v
}
// We have a waiter, so wait
return val.Value()
}
// WaitForChange waits for the value with the given key to be set again.
// If the key isn't set, it'll wait for an initial value. Note that while
// it is called "WaitForChange", the value isn't guaranteed to _change_;
// this will return when a SetValue is called for the given k.
func (w *KeyedValue) WaitForChange(k string) interface{} {
w.lock.Lock()
w.once.Do(w.init)
// If we're closed, we're closed
if w.closed {
w.lock.Unlock()
return ErrClosed
}
// Check for an active waiter. If there isn't one, make it
val := w.waiters[k]
if val == nil {
val = new(Value)
w.waiters[k] = val
}
w.lock.Unlock()
// And wait
return val.Value()
}
// ValueOk gets the value for the given key, returning immediately if the
// value doesn't exist. The second return argument is true if the value exists.
func (w *KeyedValue) ValueOk(k string) (interface{}, bool) {
w.lock.Lock()
defer w.lock.Unlock()
v, val := w.valueWaiter(k)
return v, val == nil
}
func (w *KeyedValue) SetValue(k string, v interface{}) {
w.lock.Lock()
defer w.lock.Unlock()
w.setValue(k, v)
}
// Init will initialize the key to a given value only if the key has
// not been set before. This is safe to call multiple times and in parallel.
func (w *KeyedValue) Init(k string, v interface{}) {
w.lock.Lock()
defer w.lock.Unlock()
// If we have a waiter, set the value.
_, val := w.valueWaiter(k)
if val != nil {
w.setValue(k, v)
}
}
// Must be called with w.lock held.
func (w *KeyedValue) init() {
w.values = make(map[string]interface{})
w.waiters = make(map[string]*Value)
}
// setValue is like SetValue but assumes the lock is held.
func (w *KeyedValue) setValue(k string, v interface{}) {
w.once.Do(w.init)
// Set the value, always
w.values[k] = v
// If we have a waiter, set it
if val, ok := w.waiters[k]; ok {
val.SetValue(v)
delete(w.waiters, k)
}
}
// valueWaiter gets the value or the Value waiter for a given key.
//
// This must be called with lock held.
func (w *KeyedValue) valueWaiter(k string) (interface{}, *Value) {
w.once.Do(w.init)
// If we have this value already, return it
if v, ok := w.values[k]; ok {
return v, nil
}
// If we're closed, return that
if w.closed {
return ErrClosed, nil
}
// No pending value, check for a waiter
val := w.waiters[k]
if val == nil {
val = new(Value)
w.waiters[k] = val
}
// Return the waiter
return nil, val
}
|
