/
event.go
366 lines (327 loc) · 10.7 KB
/
event.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
365
366
// Copyright 2014 Orchestrate, Inc.
//
// 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 gorc2
import (
"bytes"
"encoding/json"
"fmt"
"net/url"
"strconv"
"strings"
"time"
)
// Internal type that represents the reply form a JSON event fetch.
type jsonEvent struct {
Ordinal int64 `json:"ordinal"`
Path jsonPath `json:"path"`
Timestamp int64 `json:"timestamp"`
Value json.RawMessage `json:"value"`
}
//
// AddEvent
//
// Adds a new event to the collection with the given key, and type. The
// timestamp of the new event will be set by the Orchestrate server to the
// time that the request was processed. Unlike Create this function will
// created an event even if an event already exists with that tuple. The
// new event will be given a new Ordinal value. To update and existing
// Event use UpdateEvent() instead.
//
// Note that the key should exist otherwise this call will have unpredictable
// results.
func (c *Collection) AddEvent(
key, typ string, value interface{},
) (*Event, error) {
return c.innerAddEvent(key, typ, nil, value)
}
// Like AddEvent() except this lets you specify the timestamp that will be
// attached to the event.
func (c *Collection) AddEventWithTimestamp(
key, typ string, ts time.Time, value interface{},
) (*Event, error) {
return c.innerAddEvent(key, typ, &ts, value)
}
// Inner implementation of AddEvent*
func (c *Collection) innerAddEvent(
key, typ string, ts *time.Time, value interface{},
) (*Event, error) {
event := &Event{
Collection: c,
Key: key,
Type: typ,
}
// Encode the JSON message into a raw value that we can return to the
// client if necessary.
if rawMsg, err := json.Marshal(value); err != nil {
return nil, err
} else {
event.Value = json.RawMessage(rawMsg)
}
// Perform the actual POST
headers := map[string]string{"Content-Type": "application/json"}
var path string
if ts != nil {
path = fmt.Sprintf("%s/%s/events/%s/%d", c.Name, key, typ,
ts.UnixNano()/1000000)
} else {
path = fmt.Sprintf("%s/%s/events/%s", c.Name, key, typ)
}
resp, err := c.client.emptyReply("POST", path, headers,
bytes.NewBuffer(event.Value), 201)
if err != nil {
return nil, err
}
// Get the Location header and parse it. The Header will give us the
// Ordinal.
location := resp.Header.Get("Location")
if location == "" {
return nil, fmt.Errorf("Missing Location header.")
} else if parts := strings.Split(location, "/"); len(parts) != 8 {
return nil, fmt.Errorf("Malformed Location header.")
} else if ts, err := strconv.ParseInt(parts[6], 10, 64); err != nil {
return nil, fmt.Errorf("Malformed Timestamp in the Location header.")
} else if ord, err := strconv.ParseInt(parts[7], 10, 64); err != nil {
return nil, fmt.Errorf("Malformed Ordinal in the Location header.")
} else {
secs := ts / 1000
nsecs := (ts % 1000) * 1000000
event.Timestamp = time.Unix(secs, nsecs)
event.Ordinal = ord
}
// Get the Ref via the Etag header.
if etag := resp.Header.Get("Etag"); etag == "" {
return nil, fmt.Errorf("Missing ETag header.")
} else if parts := strings.Split(etag, `"`); len(parts) != 3 {
return nil, fmt.Errorf("Malformed ETag header.")
} else {
event.Ref = parts[1]
}
// Success
return event, nil
}
//
// DeleteEvent
//
// Removes an event from the collection. This succeeds even if the event did
// not exist prior to this call. Note that all event deletes are Final and can
// not be undone.
func (c *Collection) DeleteEvent(
key, typ string, ts time.Time, ordinal int64,
) error {
path := fmt.Sprintf("%s/%s/events/%s/%d/%d?purge=true",
c.Name, key, typ, ts.UnixNano()/1000000, ordinal)
_, err := c.client.emptyReply("DELETE", path, nil, nil, 204)
return err
}
//
// GetEvent
//
// Returns an individual event with the given details.
func (c *Collection) GetEvent(
key, typ string, ts time.Time, ordinal int64, value interface{},
) (*Event, error) {
event := &Event{
Collection: c,
Key: key,
Ordinal: ordinal,
Timestamp: ts,
Type: typ,
}
// Perform the actual GET
path := fmt.Sprintf("%s/%s/events/%s/%d/%d", c.Name, key, typ,
ts.UnixNano()/1000000, ordinal)
var responseData jsonEvent
_, err := c.client.jsonReply("GET", path, nil, 200, &responseData)
if err != nil {
return nil, err
}
// Move the data from the returned values into the Event object.
event.Value = responseData.Value
event.Ref = responseData.Path.Ref
secs := responseData.Timestamp / 1000
nsecs := (responseData.Timestamp % 1000) * 1000000
event.Timestamp = time.Unix(secs, nsecs)
event.Ordinal = responseData.Ordinal
// If the user provided us a place to unmarshal the 'value' field into
// we do that here.
if value != nil {
return event, event.Unmarshal(value)
}
// Success
return event, nil
}
//
// UpdateEvent
//
// Updates an event at the given location. In order for this to work the Event
// must exist prior to this call.
func (c *Collection) UpdateEvent(
key, typ string, ts time.Time, ordinal int64, value interface{},
) (*Event, error) {
headers := map[string]string{"Content-Type": "application/json"}
return c.innerUpdateEvent(key, typ, ts, ordinal, value, headers)
}
// Inner implementation used in both UpdateEvent and Event.Update.
func (c *Collection) innerUpdateEvent(
key, typ string, ts time.Time, ordinal int64, value interface{},
headers map[string]string,
) (*Event, error) {
event := &Event{
Collection: c,
Key: key,
Ordinal: ordinal,
Timestamp: ts,
Type: typ,
}
// Encode the JSON message into a raw value that we can return to the
// client if necessary.
if rawMsg, err := json.Marshal(value); err != nil {
return nil, err
} else {
event.Value = json.RawMessage(rawMsg)
}
// Perform the actual PUT
path := fmt.Sprintf("%s/%s/events/%s/%d/%d", c.Name, key, typ,
ts.UnixNano()/1000000, ordinal)
resp, err := c.client.emptyReply("PUT", path, headers,
bytes.NewBuffer(event.Value), 204)
if err != nil {
return nil, err
}
// Get the Location header and parse it. The Header will give us the
// Ordinal.
location := resp.Header.Get("Location")
if location == "" {
return nil, fmt.Errorf("Missing Location header.")
} else if parts := strings.Split(location, "/"); len(parts) != 8 {
return nil, fmt.Errorf("Malformed Location header.")
} else if ts, err := strconv.ParseInt(parts[6], 10, 64); err != nil {
return nil, fmt.Errorf("Malformed Timestamp in the Location header.")
} else if ord, err := strconv.ParseInt(parts[7], 10, 64); err != nil {
return nil, fmt.Errorf("Malformed Ordinal in the Location header.")
} else {
secs := ts / 1000
nsecs := (ts % 1000) * 1000000
event.Timestamp = time.Unix(secs, nsecs)
event.Ordinal = ord
}
// Get the Ref via the Etag header.
if etag := resp.Header.Get("Etag"); etag == "" {
return nil, fmt.Errorf("Missing ETag header.")
} else if parts := strings.Split(etag, `"`); len(parts) != 3 {
return nil, fmt.Errorf("Malformed ETag header.")
} else {
event.Ref = parts[1]
}
// Success
return event, nil
}
//
// ListEvents
//
//
// Search
//
// Provides optional searching parameters to a cal to ListEvents()
type ListEventsQuery struct {
// The number of results to return per call to Orchestrate. The default
// if this is not set is to return 10 at a time, the maximum that can be
// returned is 100.
Limit int
// This is the timestamp and ordinal that should be the oldest item
// included in the Event listing. Since Events a re listed newest to oldest
// this will be the last item returned (if it exists). The precision of
// the time value is miliseconds.
Start time.Time
StartOrdinal int64
// Events up to this timestamp will be included in the listing. Note that
// if EndOrdinal is not set then End behaves the same as Before. The time
// till be truncated to miliseconds.
End time.Time
EndOrdinal int64
// After the time/ordinal pairing which all events must be newer than in
// order to be included in the results. Leaving Ordinal at zero has the
// effect of including all events with the same timestamp (leaving after
// to work like Start). The time will be truncated to miliseconds for
// the search.
After time.Time
AfterOrdinal int64
// Only include listing before this time stamp. Optionally you can include
// an ordinal as well which will be used if an event exists at the exact
// same ms as Before. The precision of this time value is in miliseconds.
Before time.Time
BeforeOrdinal int64
}
// Sets up a Events listing. This does not actually perform the query, that is
// done on the first call to Next() in the iterator. If opts is nil then
// default listing parameters are used, which will return all events and
// limits the query to 10 items at a time.
func (c *Collection) ListEvents(
key, typ string, opts *ListEventsQuery,
) *Iterator {
var path string
// Build a query from the user provided values.
if opts != nil {
query := make(url.Values, 10)
if opts.Limit != 0 {
query.Add("limit", strconv.Itoa(opts.Limit))
}
var defaultTime time.Time
if opts.After != defaultTime {
if opts.AfterOrdinal != 0 {
query.Add("afterEvent", fmt.Sprintf("%d/%d",
opts.After.UnixNano()/1000000, opts.AfterOrdinal))
} else {
query.Add("afterEvent",
strconv.FormatInt(opts.After.UnixNano()/1000000, 10))
}
}
if opts.Before != defaultTime {
if opts.BeforeOrdinal != 0 {
query.Add("beforeEvent", fmt.Sprintf("%d/%d",
opts.Before.UnixNano()/1000000, opts.BeforeOrdinal))
} else {
query.Add("beforeEvent",
strconv.FormatInt(opts.Before.UnixNano()/1000000, 10))
}
}
if opts.End != defaultTime {
if opts.EndOrdinal != 0 {
query.Add("endEvent", fmt.Sprintf("%d/%d",
opts.End.UnixNano()/1000000, opts.EndOrdinal))
} else {
query.Add("endEvent",
strconv.FormatInt(opts.End.UnixNano()/1000000, 10))
}
}
if opts.Start != defaultTime {
if opts.StartOrdinal != 0 {
query.Add("startEvent", fmt.Sprintf("%d/%d",
opts.Start.UnixNano()/1000000, opts.StartOrdinal))
} else {
query.Add("startEvent",
strconv.FormatInt(opts.Start.UnixNano()/1000000, 10))
}
}
// Encode the path
path = c.Name + "/" + key + "/events/" + typ + "?" + query.Encode()
} else {
path = c.Name + "/" + key + "/events/" + typ
}
return &Iterator{
client: c.client,
iteratingEvents: true,
next: path,
}
}