forked from purpleidea/mgmt
-
Notifications
You must be signed in to change notification settings - Fork 0
/
var.go
141 lines (122 loc) · 3.95 KB
/
var.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
// Mgmt
// Copyright (C) 2013-2018+ James Shubin and the project contributors
// Written by James Shubin <james@shubin.ca> and the project contributors
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package structs
import (
"fmt"
"github.com/purpleidea/mgmt/lang/interfaces"
"github.com/purpleidea/mgmt/lang/types"
errwrap "github.com/pkg/errors"
)
// VarFunc is a function that passes through a function that came from a bind
// lookup. It exists so that the reactive function engine type checks correctly.
type VarFunc struct {
Type *types.Type // this is the type of the var's value that we hold
Func interfaces.Func
Edge string // name of the edge used
init *interfaces.Init
last types.Value // last value received to use for diff
result types.Value // last calculated output
closeChan chan struct{}
}
// Validate makes sure we've built our struct properly.
func (obj *VarFunc) Validate() error {
if obj.Type == nil {
return fmt.Errorf("must specify a type")
}
if obj.Func == nil {
return fmt.Errorf("must specify a func")
}
if obj.Edge == "" {
return fmt.Errorf("must specify an edge name")
}
// we're supposed to call Validate() before we ever call Info()
if err := obj.Func.Validate(); err != nil {
return errwrap.Wrapf(err, "func did not validate")
}
typ := obj.Func.Info().Sig
if err := obj.Type.Cmp(typ.Out); err != nil {
return errwrap.Wrapf(err, "expr type must match func out type")
}
return nil
}
// Info returns some static info about itself.
func (obj *VarFunc) Info() *interfaces.Info {
typ := &types.Type{
Kind: types.KindFunc, // function type
Map: map[string]*types.Type{obj.Edge: obj.Type},
Ord: []string{obj.Edge},
Out: obj.Type, // this is the output type for the expression
}
return &interfaces.Info{
Pure: true,
Memo: false, // TODO: ???
Sig: typ,
Err: obj.Validate(),
}
}
// Init runs some startup code for this composite function.
func (obj *VarFunc) Init(init *interfaces.Init) error {
obj.init = init
obj.closeChan = make(chan struct{})
return nil
}
// Stream takes an input struct in the format as described in the Func and Graph
// methods of the Expr, and returns the actual expected value as a stream based
// on the changing inputs to that value.
func (obj *VarFunc) Stream() error {
defer close(obj.init.Output) // the sender closes
for {
select {
case input, ok := <-obj.init.Input:
if !ok {
return nil // can't output any more
}
//if err := input.Type().Cmp(obj.Info().Sig.Input); err != nil {
// return errwrap.Wrapf(err, "wrong function input")
//}
if obj.last != nil && input.Cmp(obj.last) == nil {
continue // value didn't change, skip it
}
obj.last = input // store for next
var result types.Value
st := input.(*types.StructValue) // must be!
value, exists := st.Lookup(obj.Edge)
if !exists {
return fmt.Errorf("missing expected input argument `%s`", obj.Edge)
}
result = value
// skip sending an update...
if obj.result != nil && result.Cmp(obj.result) == nil {
continue // result didn't change
}
obj.result = result // store new result
case <-obj.closeChan:
return nil
}
select {
case obj.init.Output <- obj.result: // send
// pass
case <-obj.closeChan:
return nil
}
}
}
// Close runs some shutdown code for this function and turns off the stream.
func (obj *VarFunc) Close() error {
close(obj.closeChan)
return nil
}