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
|
package kinesis
import (
"fmt"
"github.com/docker/goamz/aws"
)
type ShardIteratorType string
type StreamStatus string
const (
// Start reading exactly from the position denoted by a specific sequence number.
ShardIteratorAtSequenceNumber ShardIteratorType = "AT_SEQUENCE_NUMBER"
// Start reading right after the position denoted by a specific sequence number.
ShardIteratorAfterSequenceNumber ShardIteratorType = "AFTER_SEQUENCE_NUMBER"
// Start reading at the last untrimmed record in the shard in the system,
// which is the oldest data record in the shard.
ShardIteratorTrimHorizon ShardIteratorType = "TRIM_HORIZON"
// Start reading just after the most recent record in the shard,
// so that you always read the most recent data in the shard.
ShardIteratorLatest ShardIteratorType = "LATEST"
// The stream is being created. Upon receiving a CreateStream request,
// Amazon Kinesis immediately returns and sets StreamStatus to CREATING.
StreamStatusCreating StreamStatus = "CREATING"
// The stream is being deleted. After a DeleteStream request,
// the specified stream is in the DELETING state until Amazon Kinesis completes the deletion.
StreamStatusDeleting StreamStatus = "DELETING"
// The stream exists and is ready for read and write operations or deletion.
// You should perform read and write operations only on an ACTIVE stream.
StreamStatusActive StreamStatus = "ACTIVE"
// Shards in the stream are being merged or split.
// Read and write operations continue to work while the stream is in the UPDATING state.
StreamStatusUpdating StreamStatus = "UPDATING"
)
// Main Kinesis object
type Kinesis struct {
aws.Auth
aws.Region
}
// The range of possible hash key values for the shard, which is a set of ordered contiguous positive integers.
type HashKeyRange struct {
EndingHashKey string
StartingHashKey string
}
func (h HashKeyRange) String() string {
return fmt.Sprintf("{EndingHashKey: %s, StartingHashKey: %s}\n",
h.EndingHashKey, h.StartingHashKey)
}
// The range of possible sequence numbers for the shard.
type SequenceNumberRange struct {
EndingSequenceNumber string
StartingSequenceNumber string
}
func (s SequenceNumberRange) String() string {
return fmt.Sprintf("{EndingSequenceNumber: %s, StartingSequenceNumber: %s}\n",
s.EndingSequenceNumber, s.StartingSequenceNumber)
}
// A uniquely identified group of data records in an Amazon Kinesis stream.
type Shard struct {
AdjacentParentShardId string
HashKeyRange HashKeyRange
ParentShardId string
SequenceNumberRange SequenceNumberRange
ShardId string
}
// Description of a Stream
type StreamDescription struct {
HasMoreShards bool
Shards []Shard
StreamARN string
StreamName string
StreamStatus StreamStatus
}
// The unit of data of the Amazon Kinesis stream, which is composed of a sequence number,
// a partition key, and a data blob.
type Record struct {
Data []byte
PartitionKey string
SequenceNumber string
}
// Represents the output of a DescribeStream operation.
type DescribeStreamResponse struct {
StreamDescription StreamDescription
}
// Represents the output of a GetRecords operation.
type GetRecordsResponse struct {
NextShardIterator string
Records []Record
}
// Represents the output of a GetShardIterator operation.
type GetShardIteratorResponse struct {
ShardIterator string
}
// Represents the output of a ListStreams operation.
type ListStreamResponse struct {
HasMoreStreams bool
StreamNames []string
}
// Represents the output of a PutRecord operation.
type PutRecordResponse struct {
SequenceNumber string
ShardId string
}
// The unit of data put to the Amazon Kinesis stream by PutRecords, which includes
// a partition key, a hash key, and a data blob.
type PutRecordsRequestEntry struct {
PartitionKey string
HashKey string `json:"ExplicitHashKey,omitempty"`
Data []byte
}
// Represents the output of a PutRecords operation.
type PutRecordsResponse struct {
FailedRecordCount int
Records []PutRecordsResultEntry
}
type PutRecordsResultEntry struct {
ErrorCode string
ErrorMessage string
SequenceNumber string
ShardId string
}
// Error represents an error in an operation with Kinesis(following goamz/Dynamodb)
type Error struct {
StatusCode int // HTTP status code (200, 403, ...)
Status string
Code string `json:"__type"`
Message string `json:"message"`
}
func (e Error) Error() string {
return fmt.Sprintf("[HTTP %d] %s : %s\n", e.StatusCode, e.Code, e.Message)
}
|
