-
Notifications
You must be signed in to change notification settings - Fork 387
/
list.go
152 lines (128 loc) · 4.33 KB
/
list.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
// Copyright (C) 2020 Storj Labs, Inc.
// See LICENSE for copying information.
package metabase
import (
"context"
"time"
"storj.io/common/storj"
"storj.io/common/uuid"
)
// ObjectEntry contains information about an item in a bucket.
type ObjectEntry struct {
IsPrefix bool
ObjectKey ObjectKey
Version Version
StreamID uuid.UUID
CreatedAt time.Time
ExpiresAt *time.Time
Status ObjectStatus
SegmentCount int32
EncryptedMetadataNonce []byte
EncryptedMetadata []byte
EncryptedMetadataEncryptedKey []byte
TotalPlainSize int64
TotalEncryptedSize int64
FixedSegmentSize int32
Encryption storj.EncryptionParameters
// ZombieDeletionDeadline defines when the pending raw object should be deleted from the database.
// This is as a safeguard against objects that failed to upload and the client has not indicated
// whether they want to continue uploading or delete the already uploaded data.
ZombieDeletionDeadline *time.Time
}
// ObjectsIterator iterates over a sequence of ObjectEntry items.
type ObjectsIterator interface {
Next(ctx context.Context, item *ObjectEntry) bool
}
// IterateCursor is a cursor used during iteration through objects.
type IterateCursor struct {
Key ObjectKey
Version Version
}
// StreamIDCursor is a cursor used during iteration through streamIDs of a pending object.
type StreamIDCursor struct {
StreamID uuid.UUID
}
// IterateObjects contains arguments necessary for listing objects in a bucket.
type IterateObjects struct {
ProjectID uuid.UUID
BucketName string
BatchSize int
Prefix ObjectKey
Cursor IterateCursor
}
// Verify verifies get object request fields.
func (opts *IterateObjects) Verify() error {
switch {
case opts.ProjectID.IsZero():
return ErrInvalidRequest.New("ProjectID missing")
case opts.BucketName == "":
return ErrInvalidRequest.New("BucketName missing")
case opts.BatchSize < 0:
return ErrInvalidRequest.New("BatchSize is negative")
}
return nil
}
// IterateObjectsAllVersions iterates through all versions of all objects.
func (db *DB) IterateObjectsAllVersions(ctx context.Context, opts IterateObjects, fn func(context.Context, ObjectsIterator) error) (err error) {
defer mon.Task()(&ctx)(&err)
if err = opts.Verify(); err != nil {
return err
}
return iterateAllVersions(ctx, db, opts, fn)
}
// IteratePendingObjectsByKey contains arguments necessary for listing pending objects by ObjectKey.
type IteratePendingObjectsByKey struct {
ObjectLocation
BatchSize int
Cursor StreamIDCursor
}
// IterateObjectsWithStatus contains arguments necessary for listing objects in a bucket.
type IterateObjectsWithStatus struct {
ProjectID uuid.UUID
BucketName string
Recursive bool
BatchSize int
Prefix ObjectKey
Cursor IterateCursor
Status ObjectStatus
}
// IterateObjectsAllVersionsWithStatus iterates through all versions of all objects with specified status.
func (db *DB) IterateObjectsAllVersionsWithStatus(ctx context.Context, opts IterateObjectsWithStatus, fn func(context.Context, ObjectsIterator) error) (err error) {
defer mon.Task()(&ctx)(&err)
if err = opts.Verify(); err != nil {
return err
}
return iterateAllVersionsWithStatus(ctx, db, opts, fn)
}
// Verify verifies get object request fields.
func (opts *IterateObjectsWithStatus) Verify() error {
switch {
case opts.ProjectID.IsZero():
return ErrInvalidRequest.New("ProjectID missing")
case opts.BucketName == "":
return ErrInvalidRequest.New("BucketName missing")
case opts.BatchSize < 0:
return ErrInvalidRequest.New("BatchSize is negative")
case !(opts.Status == Pending || opts.Status == Committed):
return ErrInvalidRequest.New("Status %v is not supported", opts.Status)
}
return nil
}
// IteratePendingObjectsByKey iterates through all streams of pending objects with the same ObjectKey.
func (db *DB) IteratePendingObjectsByKey(ctx context.Context, opts IteratePendingObjectsByKey, fn func(context.Context, ObjectsIterator) error) (err error) {
defer mon.Task()(&ctx)(&err)
if err := opts.Verify(); err != nil {
return err
}
return iteratePendingObjectsByKey(ctx, db, opts, fn)
}
// Verify verifies get object request fields.
func (opts *IteratePendingObjectsByKey) Verify() error {
if err := opts.ObjectLocation.Verify(); err != nil {
return err
}
if opts.BatchSize < 0 {
return ErrInvalidRequest.New("BatchSize is negative")
}
return nil
}