-
-
Notifications
You must be signed in to change notification settings - Fork 12
/
types.go
339 lines (281 loc) · 8.46 KB
/
types.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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
// Package machine - Copyright © 2020 Jonathan Whitaker <github@whitaker.io>.
//
// Use of this source code is governed by an MIT-style
// license that can be found in the LICENSE file.
package machine
import (
"bytes"
"context"
"encoding/gob"
"encoding/json"
"time"
"github.com/whitaker-io/data"
)
var (
// ForkDuplicate is a Fork that creates a deep copy of the
// payload and sends it down both branches.
ForkDuplicate Fork = func(payload []*Packet) (a, b []*Packet) {
return payload, deepCopyPayload(payload)
}
// ForkError is a Fork that splits machine.Packets based on
// the presence of an error. Errors are sent down the
// right side path.
ForkError Fork = func(payload []*Packet) (s, f []*Packet) {
s = []*Packet{}
f = []*Packet{}
for _, packet := range payload {
if len(packet.Errors) > 0 {
f = append(f, packet)
} else {
s = append(s, packet)
}
}
return s, f
}
defaultOptions = &Option{
DeepCopy: boolP(false),
FIFO: boolP(false),
Metrics: boolP(true),
Span: boolP(true),
BufferSize: intP(0),
Provider: &edgeProvider{},
}
)
// EdgeProvider is an interface that is used for providing new instances
// of the Edge interface given the *Option set in the Stream
type EdgeProvider interface {
New(ctx context.Context, id string, options *Option) Edge
}
// Edge is an inteface that is used for transferring data between vertices
type Edge interface {
Send(ctx context.Context, channel chan []*Packet)
Next(payload ...*Packet)
}
// Subscription is an interface for creating a pull based stream.
// It requires 2 methods Read and Close.
//
// Read is called when the interval passes and the resulting
// payload is sent down the Stream.
//
// Close is called during a graceful termination and any errors
// are logged.
type Subscription interface {
Read(ctx context.Context) []data.Data
Close() error
}
// Publisher is an interface for sending data out of the Stream
type Publisher interface {
Send([]data.Data) error
}
// Packet type that holds information traveling through the machine.
type Packet struct {
ID string `json:"id"`
Data data.Data `json:"data"`
Errors map[string]error `json:"errors"`
}
// Option type for holding machine settings.
type Option struct {
// DeepCopy uses encoding/gob to create a deep copy of the payload
// before the processing to ensure concurrent map exceptions cannot
// happen. Is fairly resource intensive so use with caution.
// Default: false
DeepCopy *bool `json:"deep_copy,omitempty" mapstructure:"deep_copy,omitempty"`
// FIFO controls the processing order of the payloads
// If set to true the system will wait for one payload
// to be processed before starting the next.
// Default: false
FIFO *bool `json:"fifo,omitempty" mapstructure:"fifo,omitempty"`
// BufferSize sets the buffer size on the edge channels between the
// vertices, this setting can be useful when processing large amounts
// of data with FIFO turned on.
// Default: 0
BufferSize *int `json:"buffer_size,omitempty" mapstructure:"buffer_size,omitempty"`
// Span controls whether opentelemetry spans are created for tracing
// Packets processed by the system.
// Default: true
Span *bool `json:"spans_enabled,omitempty" mapstructure:"spans_enabled,omitempty"`
// Metrics controls whether opentelemetry metrics are recorded for
// Packets processed by the system.
// Default: true
Metrics *bool `json:"metrics_enabled,omitempty" mapstructure:"metrics_enabled,omitempty"`
// Provider determines the edge type to be used, logic for what type of edge
// for a given id is required if not using homogeneous edges
// Default: nil
Provider EdgeProvider `json:"-" mapstructure:"-"`
// Validators are used to ensure the incoming Data is compliant
// they are run at the start of the stream before creation of Packets
// Default: nil
Validators map[string]ForkRule `json:"-" mapstructure:"-"`
}
// Retriever is a function that provides data to a generic Stream
// must stop when the context receives a done signal.
type Retriever func(ctx context.Context) chan []data.Data
// Applicative is a function that is applied on an individual
// basis for each Packet in the payload. The resulting data replaces
// the old data
type Applicative func(d data.Data) data.Data
// Window is a function that is applied to the entire payload.
// The resulting data replaces the old data
type Window func(list ...*Packet) []*Packet
// Fold is a function used to combine a payload into a single Packet.
// It may be used with either a Fold Left or Fold Right operation,
// which starts at the corresponding side and moves through the payload.
// The returned instance of data.Data is used as the aggregate in the subsequent
// call.
type Fold func(aggregate, next data.Data) data.Data
// Fork is a function for splitting the payload into 2 separate paths.
// Default Forks for duplication and error checking are provided by
// ForkDuplicate and ForkError respectively.
type Fork func(list []*Packet) (a, b []*Packet)
// ForkRule is a function that can be converted to a Fork via the Handler
// method allowing for Forking based on the contents of the data.
type ForkRule func(d data.Data) bool
// Comparator is a function to compare 2 data.Data's
type Comparator func(a data.Data, b data.Data) int
// Remover func that is used to remove Data based on a true result
type Remover func(index int, d data.Data) bool
// Error type for wrapping errors coming from the Stream
type Error struct {
Err error `json:"err"`
StreamID string `json:"stream_id"`
VertexID string `json:"vertex_id"`
VertexType string `json:"vertex_type"`
Packets []*Packet `json:"payload"`
Time time.Time `json:"time"`
}
type handler func(payload []*Packet)
type edgeProvider struct{}
type edge struct {
channel chan []*Packet
}
func (o *Option) merge(options ...*Option) *Option {
if len(options) < 1 {
return o
} else if len(options) == 1 {
return o.join(options[0])
}
return o.join(options[0]).merge(options[1:]...)
}
func (o *Option) join(option *Option) *Option {
out := &Option{
DeepCopy: o.DeepCopy,
FIFO: o.FIFO,
BufferSize: o.BufferSize,
Metrics: o.Metrics,
Span: o.Span,
Provider: o.Provider,
Validators: o.Validators,
}
if option.DeepCopy != nil {
out.DeepCopy = option.DeepCopy
}
if option.FIFO != nil {
out.FIFO = option.FIFO
}
if option.BufferSize != nil {
out.BufferSize = option.BufferSize
}
if option.Metrics != nil {
out.Metrics = option.Metrics
}
if option.Span != nil {
out.Span = option.Span
}
if option.Provider != nil {
out.Provider = option.Provider
}
if option.Validators != nil {
out.Validators = option.Validators
}
return out
}
// Handler is a method for turning the ForkRule into an instance of Fork
func (r ForkRule) Handler(payload []*Packet) (t, f []*Packet) {
t = []*Packet{}
f = []*Packet{}
for _, packet := range payload {
if r(packet.Data) {
t = append(t, packet)
} else {
f = append(f, packet)
}
}
return t, f
}
func (e *Error) Error() string {
bytez, err := json.Marshal(e)
if err != nil {
return e.Error()
}
return string(bytez)
}
func (p *edgeProvider) New(ctx context.Context, id string, options *Option) Edge {
b := 0
if options.BufferSize != nil {
b = *options.BufferSize
}
return &edge{
channel: make(chan []*Packet, b),
}
}
func (out *edge) Send(ctx context.Context, channel chan []*Packet) {
go func() {
Loop:
for {
select {
case <-ctx.Done():
break Loop
case list := <-out.channel:
if len(list) > 0 {
channel <- list
}
}
}
}()
}
func (out *edge) Next(payload ...*Packet) {
out.channel <- payload
}
func boolP(v bool) *bool {
return &v
}
func intP(v int) *int {
return &v
}
func (o *Option) validate(payload ...data.Data) [][]string {
list := make([][]string, len(payload))
failed := false
if o.Validators != nil {
for i, d := range payload {
for k, v := range o.Validators {
if !v(d) {
if len(list[i]) < 1 {
list[i] = []string{}
}
list[i] = append(list[i], k)
failed = true
}
}
}
}
if failed {
return list
}
return [][]string{}
}
func deepCopy(d []data.Data) []data.Data {
out := []data.Data{}
buf := &bytes.Buffer{}
enc, dec := gob.NewEncoder(buf), gob.NewDecoder(buf)
_ = enc.Encode(d)
_ = dec.Decode(&out)
return out
}
func deepCopyPayload(d []*Packet) []*Packet {
out := []*Packet{}
buf := &bytes.Buffer{}
enc, dec := gob.NewEncoder(buf), gob.NewDecoder(buf)
_ = enc.Encode(d)
_ = dec.Decode(&out)
return out
}