forked from looplab/eventhorizon
-
Notifications
You must be signed in to change notification settings - Fork 0
/
context.go
177 lines (148 loc) · 5.34 KB
/
context.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
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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
// Copyright (c) 2014 - The Event Horizon authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package eventhorizon
import (
"context"
"sync"
"time"
)
// DefaultNamespace is the namespace to use if not set in the context.
const DefaultNamespace = "default"
// DefaultMinVersionDeadline is the deadline to use when creating a min version
// context that waits.
const DefaultMinVersionDeadline = 10 * time.Second
func init() {
// Register the namespace context.
RegisterContextMarshaler(func(ctx context.Context, vals map[string]interface{}) {
if ns, ok := ctx.Value(namespaceKey).(string); ok {
vals[namespaceKeyStr] = ns
}
})
RegisterContextUnmarshaler(func(ctx context.Context, vals map[string]interface{}) context.Context {
if ns, ok := vals[namespaceKeyStr].(string); ok {
return NewContextWithNamespace(ctx, ns)
}
return ctx
})
// Register the version context.
RegisterContextMarshaler(func(ctx context.Context, vals map[string]interface{}) {
if v, ok := ctx.Value(minVersionKey).(int); ok {
vals[minVersionKeyStr] = v
}
})
RegisterContextUnmarshaler(func(ctx context.Context, vals map[string]interface{}) context.Context {
if v, ok := vals[minVersionKeyStr].(int); ok {
return NewContextWithMinVersion(ctx, v)
}
// Support JSON-like marshaling of ints as floats.
if v, ok := vals[minVersionKeyStr].(float64); ok {
return NewContextWithMinVersion(ctx, int(v))
}
return ctx
})
}
type contextKey int
// Context keys for namespace and min version.
const (
namespaceKey contextKey = iota
minVersionKey
)
// Strings used to marshal context values.
const (
namespaceKeyStr = "eh_namespace"
minVersionKeyStr = "eh_minversion"
)
// NamespaceFromContext returns the namespace from the context, or the default
// namespace.
func NamespaceFromContext(ctx context.Context) string {
if ns, ok := ctx.Value(namespaceKey).(string); ok {
return ns
}
return DefaultNamespace
}
// NewContextWithNamespace sets the namespace to use in the context. The
// namespace is used to determine which database.
func NewContextWithNamespace(ctx context.Context, namespace string) context.Context {
return context.WithValue(ctx, namespaceKey, namespace)
}
// MinVersionFromContext returns the min version from the context.
func MinVersionFromContext(ctx context.Context) (int, bool) {
minVersion, ok := ctx.Value(minVersionKey).(int)
return minVersion, ok
}
// NewContextWithMinVersion returns the context with min version set.
func NewContextWithMinVersion(ctx context.Context, minVersion int) context.Context {
return context.WithValue(ctx, minVersionKey, minVersion)
}
// NewContextWithMinVersionWait returns the context with min version and a
// default deadline set.
func NewContextWithMinVersionWait(ctx context.Context, minVersion int) (c context.Context, cancel func()) {
ctx = context.WithValue(ctx, minVersionKey, minVersion)
return context.WithTimeout(ctx, DefaultMinVersionDeadline)
}
// Private context marshaling funcs.
var (
contextMarshalFuncs = []ContextMarshalFunc{}
contextMarshalFuncsMu = sync.RWMutex{}
contextUnmarshalFuncs = []ContextUnmarshalFunc{}
contextUnmarshalFuncsMu = sync.RWMutex{}
)
// ContextMarshalFunc is a function that marshalls any context values to a map,
// used for sending context on the wire.
type ContextMarshalFunc func(context.Context, map[string]interface{})
// RegisterContextMarshaler registers a marshaler function used by MarshalContext.
func RegisterContextMarshaler(f ContextMarshalFunc) {
contextMarshalFuncsMu.Lock()
defer contextMarshalFuncsMu.Unlock()
contextMarshalFuncs = append(contextMarshalFuncs, f)
}
// MarshalContext marshals a context into a map.
func MarshalContext(ctx context.Context) map[string]interface{} {
contextMarshalFuncsMu.RLock()
defer contextMarshalFuncsMu.RUnlock()
allVals := map[string]interface{}{}
for _, f := range contextMarshalFuncs {
vals := map[string]interface{}{}
f(ctx, vals)
for key, val := range vals {
if _, ok := allVals[key]; ok {
panic("duplicate context entry for: " + key)
}
allVals[key] = val
}
}
return allVals
}
// ContextUnmarshalFunc is a function that marshalls any context values to a map,
// used for sending context on the wire.
type ContextUnmarshalFunc func(context.Context, map[string]interface{}) context.Context
// RegisterContextUnmarshaler registers a marshaler function used by UnmarshalContext.
func RegisterContextUnmarshaler(f ContextUnmarshalFunc) {
contextUnmarshalFuncsMu.Lock()
defer contextUnmarshalFuncsMu.Unlock()
contextUnmarshalFuncs = append(contextUnmarshalFuncs, f)
}
// UnmarshalContext unmarshals a context from a map.
func UnmarshalContext(vals map[string]interface{}) context.Context {
contextUnmarshalFuncsMu.RLock()
defer contextUnmarshalFuncsMu.RUnlock()
ctx := context.Background()
if vals == nil {
return ctx
}
for _, f := range contextUnmarshalFuncs {
ctx = f(ctx, vals)
}
return ctx
}