-
Notifications
You must be signed in to change notification settings - Fork 35
/
contract.go
175 lines (154 loc) · 3.65 KB
/
contract.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
package msg
import (
"fmt"
"github.com/viant/endly/model/location"
"github.com/viant/endly/service/testing/validator"
"github.com/viant/toolbox/data"
)
const defaultTimeoutMs = 10000
// CreateRequest represents a create resource request
type CreateRequest struct {
Credentials string
Resources []*ResourceSetup
}
func (r *CreateRequest) Init() error {
if len(r.Resources) == 0 {
return nil
}
for _, resource := range r.Resources {
if err := resource.Init(); err != nil {
return err
}
if resource.Credentials == "" {
resource.Credentials = r.Credentials
}
}
return nil
}
func (r *CreateRequest) Validate() error {
if len(r.Resources) == 0 {
return fmt.Errorf("resources was empty")
}
for _, resource := range r.Resources {
if err := resource.Validate(); err != nil {
return err
}
}
return nil
}
// CreateResponse represents a create resource response
type CreateResponse struct {
Resources []*Resource
}
// DeleteRequest represents a delete resource request
type DeleteRequest struct {
Credentials string
Resources []*Resource
}
func (r *DeleteRequest) Init() error {
if len(r.Resources) == 0 {
return nil
}
for _, resource := range r.Resources {
_ = resource.Init()
if resource.Credentials == "" {
resource.Credentials = r.Credentials
}
}
return nil
}
// DeleteResponse represents a delete resource response
type DeleteResponse struct{}
// PushRequest represents push request
type PushRequest struct {
Credentials string
Dest *Resource
Messages []*Message
Source *location.Resource `description:"source for message data currently JSON or new line delimitered json are supported that conform []*Message type "`
TimeoutMs int
UDF string
isInitialized bool
}
func (r *PushRequest) Init() error {
if r.isInitialized {
return nil
}
if r.Dest != nil {
if err := r.Dest.Init(); err != nil {
return err
}
if r.Dest.Credentials == "" {
r.Dest.Credentials = r.Credentials
}
}
if r.TimeoutMs == 0 {
r.TimeoutMs = defaultTimeoutMs
}
return nil
}
func (r *PushRequest) Validate() error {
if r.Dest == nil {
return fmt.Errorf("dest was empty")
}
if len(r.Messages) == 0 && r.Source == nil {
return fmt.Errorf("messages were empty")
}
return nil
}
// PushResponse represents a push response
type PushResponse struct {
Results []Result
}
// PullRequest represents a pull request
type PullRequest struct {
Credentials string
Source *Resource
TimeoutMs int
Count int
Nack bool `description:"flag indicates that the client will not or cannot process a Message passed to the Subscriber.Receive callback."`
UDF string
Expect interface{}
}
func (r *PullRequest) Init() error {
if r.TimeoutMs == 0 {
r.TimeoutMs = defaultTimeoutMs
}
if r.Source.Credentials == "" {
r.Source.Credentials = r.Credentials
}
return r.Source.Init()
}
func (r *PullRequest) Validate() error {
if r.Source == nil {
return fmt.Errorf("source was empty")
}
return nil
}
// PullRequest represents a pull response
type PullResponse struct {
Messages []*Message
Assert *validator.AssertResponse
}
type Message struct {
ID string
Subject string
Attributes map[string]interface{}
Data interface{}
Transformed interface{} `description:"udf transformed data"`
}
func (m *Message) Expand(state data.Map) *Message {
var result = &Message{
Attributes: make(map[string]interface{}),
Subject: m.Subject,
}
if len(m.Attributes) > 0 {
for k, v := range m.Attributes {
result.Attributes[state.ExpandAsText(k)] = state.Expand(v)
}
}
if m.Data != nil {
result.Data = state.Expand(m.Data)
}
return result
}
type Result interface{}