forked from gazette/core
/
rpc_extensions.go
364 lines (325 loc) · 11 KB
/
rpc_extensions.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
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
package protocol
import (
"net/url"
"strings"
)
// RoutedJournalClient composes a JournalClient and DispatchRouter.
type RoutedJournalClient interface {
JournalClient
DispatchRouter
}
// NewRoutedJournalClient composes a JournalClient and DispatchRouter.
func NewRoutedJournalClient(jc JournalClient, dr DispatchRouter) RoutedJournalClient {
return struct {
JournalClient
DispatchRouter
}{
JournalClient: jc,
DispatchRouter: dr,
}
}
// Validate returns an error if the ReadRequest is not well-formed.
func (m *ReadRequest) Validate() error {
if m.Header != nil {
if err := m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
}
if err := m.Journal.Validate(); err != nil {
return ExtendContext(err, "Journal")
} else if m.Offset < -1 {
return NewValidationError("invalid Offset (%d; expected -1 <= Offset <= MaxInt64)", m.Offset)
}
// Block, DoNotProxy, and MetadataOnly (each type bool) require no extra validation.
return nil
}
// Validate returns an error if the ReadResponse is not well-formed.
func (m *ReadResponse) Validate() error {
if err := m.Status.Validate(); err != nil {
return ExtendContext(err, "Status")
}
if m.Content != nil {
// Require that no other fields are set.
if m.Status != Status_OK {
return NewValidationError("unexpected Status with Content (%v)", m.Status)
} else if m.Header != nil {
return NewValidationError("unexpected Header with Content (%s)", m.Header)
} else if m.WriteHead != 0 {
return NewValidationError("unexpected WriteHead with Content (%d)", m.WriteHead)
} else if m.Fragment != nil {
return NewValidationError("unexpected Fragment with Content (%s)", m.Fragment)
} else if m.FragmentUrl != "" {
return NewValidationError("unexpected FragmentUrl with Content (%s)", m.FragmentUrl)
}
return nil
}
if m.Header != nil {
if err := m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
}
if m.WriteHead < 0 {
return NewValidationError("invalid WriteHead (%d; expected >= 0)", m.WriteHead)
}
if m.Fragment != nil {
if err := m.Fragment.Validate(); err != nil {
return ExtendContext(err, "Fragment")
} else if m.Offset < m.Fragment.Begin || m.Offset >= m.Fragment.End {
return NewValidationError("invalid Offset (%d; expected %d <= offset < %d)",
m.Offset, m.Fragment.Begin, m.Fragment.End)
} else if m.WriteHead < m.Fragment.End {
return NewValidationError("invalid WriteHead (%d; expected >= %d)",
m.WriteHead, m.Fragment.End)
}
if m.FragmentUrl != "" {
if _, err := url.Parse(m.FragmentUrl); err != nil {
return ExtendContext(&ValidationError{Err: err}, "FragmentUrl")
}
}
} else {
if m.Status == Status_OK && m.Offset != 0 {
return NewValidationError("unexpected Offset without Fragment or Content (%d)", m.Offset)
} else if m.FragmentUrl != "" {
return NewValidationError("unexpected FragmentUrl without Fragment (%s)", m.FragmentUrl)
}
}
return nil
}
// Validate returns an error if the AppendRequest is not well-formed.
func (m *AppendRequest) Validate() error {
if m.Journal != "" {
if m.Header != nil {
if err := m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
}
if err := m.Journal.Validate(); err != nil {
return ExtendContext(err, "Journal")
} else if m.Offset < 0 {
return NewValidationError("invalid Offset (%d; expected >= 0)", m.Offset)
} else if len(m.Content) != 0 {
return NewValidationError("unexpected Content")
}
} else if m.Header != nil {
return NewValidationError("unexpected Header")
} else if m.DoNotProxy {
return NewValidationError("unexpected DoNotProxy")
} else if m.Offset != 0 {
return NewValidationError("unexpected Offset")
}
return nil
}
// Validate returns an error if the AppendResponse is not well-formed.
func (m *AppendResponse) Validate() error {
if err := m.Status.Validate(); err != nil {
return ExtendContext(err, "Status")
} else if err = m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
} else if m.Status == Status_OK && m.Commit == nil {
return NewValidationError("expected Commit")
}
if m.Commit != nil {
if err := m.Commit.Validate(); err != nil {
return ExtendContext(err, "Commit")
}
}
return nil
}
// Validate returns an error if the ReplicateRequest is not well-formed.
func (m *ReplicateRequest) Validate() error {
if m.Journal != "" {
// This is an initial request.
if err := m.Journal.Validate(); err != nil {
return ExtendContext(err, "Journal")
} else if m.Header == nil {
return NewValidationError("expected Header with Journal")
} else if err = m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
} else if m.Proposal == nil {
return NewValidationError("expected Proposal with Journal")
} else if err = m.Proposal.Validate(); err != nil {
return ExtendContext(err, "Proposal")
} else if m.Proposal.Journal != m.Journal {
return NewValidationError("Journal and Proposal.Journal mismatch (%s vs %s)", m.Journal, m.Proposal.Journal)
} else if m.Content != nil {
return NewValidationError("unexpected Content with Journal (len %d)", len(m.Content))
} else if m.ContentDelta != 0 {
return NewValidationError("unexpected ContentDelta with Journal (%d)", m.ContentDelta)
} else if m.Acknowledge == false {
return NewValidationError("expected Acknowledge with Journal")
}
return nil
}
if m.Header != nil {
return NewValidationError("unexpected Header without Journal (%s)", m.Header)
}
if m.Proposal != nil {
if err := m.Proposal.Validate(); err != nil {
return ExtendContext(err, "Proposal")
} else if m.Content != nil {
return NewValidationError("unexpected Content with Proposal (len %d)", len(m.Content))
} else if m.ContentDelta != 0 {
return NewValidationError("unexpected ContentDelta with Proposal (%d)", m.ContentDelta)
}
return nil
}
if m.Content == nil {
return NewValidationError("expected Content or Proposal")
}
if m.Acknowledge {
return NewValidationError("unexpected Acknowledge with Content")
}
if m.ContentDelta < 0 {
return NewValidationError("invalid ContentDelta (%d; expected >= 0)", m.ContentDelta)
}
return nil
}
// Validate returns an error if the ReplicateResponse is not well-formed.
func (m *ReplicateResponse) Validate() error {
if err := m.Status.Validate(); err != nil {
return ExtendContext(err, "Status")
}
if m.Status == Status_WRONG_ROUTE {
if m.Header == nil {
return NewValidationError("expected Header")
} else if err := m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
} else if m.Header != nil {
return NewValidationError("unexpected Header (%s)", m.Header)
}
if m.Status == Status_FRAGMENT_MISMATCH {
if m.Fragment == nil {
return NewValidationError("expected Fragment")
} else if err := m.Fragment.Validate(); err != nil {
return ExtendContext(err, "Fragment")
}
} else if m.Fragment != nil {
return NewValidationError("unexpected Fragment (%s)", m.Fragment)
}
return nil
}
func (m *ListRequest) Validate() error {
if err := m.Selector.Validate(); err != nil {
return ExtendContext(err, "Selector")
}
for _, v := range m.Selector.Include.ValuesOf("prefix") {
if !strings.HasSuffix(v, "/") {
return NewValidationError("Selector.Include.Labels[\"prefix\"]: expected trailing '/' (%+v)", v)
}
}
for _, v := range m.Selector.Exclude.ValuesOf("prefix") {
if !strings.HasSuffix(v, "/") {
return NewValidationError("Selector.Exclude.Labels[\"prefix\"]: expected trailing '/' (%+v)", v)
}
}
// PageLimit and PageToken require no extra validation.
return nil
}
func (m *ListResponse) Validate() error {
if err := m.Status.Validate(); err != nil {
return ExtendContext(err, "Status")
} else if err = m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
for i, j := range m.Journals {
if err := j.Validate(); err != nil {
return ExtendContext(err, "Journals[%d]", i)
}
}
// NextPageToken requires no extra validation.
return nil
}
func (m *ListResponse_Journal) Validate() error {
if err := m.Spec.Validate(); err != nil {
return ExtendContext(err, "Spec")
} else if m.ModRevision <= 0 {
return NewValidationError("invalid ModRevision (%d; expected > 0)", m.ModRevision)
} else if err = m.Route.Validate(); err != nil {
return ExtendContext(err, "Route")
}
return nil
}
func (m *ApplyRequest) Validate() error {
for i, u := range m.Changes {
if err := u.Validate(); err != nil {
return ExtendContext(err, "Changes[%d]", i)
}
}
return nil
}
func (m *ApplyRequest_Change) Validate() error {
if m.Upsert != nil {
if m.Delete != "" {
return NewValidationError("both Upsert and Delete are set (expected exactly one)")
} else if err := m.Upsert.Validate(); err != nil {
return ExtendContext(err, "Upsert")
} else if m.ExpectModRevision < 0 {
return NewValidationError("invalid ExpectModRevision (%d; expected >= 0)", m.ExpectModRevision)
}
} else if m.Delete != "" {
if err := m.Delete.Validate(); err != nil {
return ExtendContext(err, "Delete")
} else if m.ExpectModRevision <= 0 {
return NewValidationError("invalid ExpectModRevision (%d; expected > 0)", m.ExpectModRevision)
}
} else {
return NewValidationError("neither Upsert nor Delete are set (expected exactly one)")
}
return nil
}
func (m *ApplyResponse) Validate() error {
if err := m.Status.Validate(); err != nil {
return ExtendContext(err, "Status")
} else if err = m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
return nil
}
func (m *FragmentsRequest) Validate() error {
if m.Header != nil {
if err := m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
}
if err := m.Journal.Validate(); err != nil {
return ExtendContext(err, "Journal")
} else if m.EndModTime != 0 && m.EndModTime < m.BeginModTime {
return NewValidationError("invalid EndModTime (%v must be after %v)", m.EndModTime, m.BeginModTime)
} else if m.NextPageToken < 0 {
return NewValidationError("invalid NextPageToken (%v; must be >= 0)", m.NextPageToken)
} else if m.PageLimit < 0 {
return NewValidationError("invalid PageLimit (%v; must be >= 0)", m.PageLimit)
} else if m.SignatureTTL != nil && *m.SignatureTTL <= 0 {
return NewValidationError("invalid SignatureTTL (%v; must be > 0s)", *m.SignatureTTL)
}
return nil
}
func (m *FragmentsResponse) Validate() error {
if err := m.Status.Validate(); err != nil {
return ExtendContext(err, "Status")
} else if err = m.Header.Validate(); err != nil {
return ExtendContext(err, "Header")
}
for i, f := range m.Fragments {
if err := f.Validate(); err != nil {
return ExtendContext(err, "Fragments[%d]", i)
}
}
// NextPageToken requires no extra validation.
return nil
}
func (m *FragmentsResponse__Fragment) Validate() error {
if err := m.Spec.Validate(); err != nil {
return ExtendContext(err, "Spec")
} else if _, err = url.Parse(m.SignedUrl); err != nil {
return ExtendContext(&ValidationError{Err: err}, "SignedUrl")
}
return nil
}
func (x Status) Validate() error {
if _, ok := Status_name[int32(x)]; !ok {
return NewValidationError("invalid status (%s)", x)
}
return nil
}