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
|
package objectstore
import (
"context"
"io"
"time"
"gitlab.com/gitlab-org/labkit/log"
"gocloud.dev/blob"
"gocloud.dev/gcerrors"
)
type GoCloudObject struct {
bucket *blob.Bucket
mux *blob.URLMux
bucketURL string
objectName string
*uploader
}
type GoCloudObjectParams struct {
Ctx context.Context
Mux *blob.URLMux
BucketURL string
ObjectName string
}
func NewGoCloudObject(p *GoCloudObjectParams) (*GoCloudObject, error) {
bucket, err := p.Mux.OpenBucket(p.Ctx, p.BucketURL)
if err != nil {
return nil, err
}
o := &GoCloudObject{
bucket: bucket,
mux: p.Mux,
bucketURL: p.BucketURL,
objectName: p.ObjectName,
}
o.uploader = newUploader(o)
return o, nil
}
func (o *GoCloudObject) Upload(ctx context.Context, r io.Reader) error {
defer o.bucket.Close()
writer, err := o.bucket.NewWriter(ctx, o.objectName, nil)
if err != nil {
log.ContextLogger(ctx).WithError(err).Error("error creating GoCloud bucket")
return err
}
if _, err = io.Copy(writer, r); err != nil {
log.ContextLogger(ctx).WithError(err).Error("error writing to GoCloud bucket")
writer.Close()
return err
}
if err := writer.Close(); err != nil {
log.ContextLogger(ctx).WithError(err).Error("error closing GoCloud bucket")
return err
}
return nil
}
func (o *GoCloudObject) ETag() string {
return ""
}
func (o *GoCloudObject) Abort() {
o.Delete()
}
// Delete will always attempt to delete the temporary file.
// According to https://github.com/google/go-cloud/blob/7818961b5c9a112f7e092d3a2d8479cbca80d187/blob/azureblob/azureblob.go#L881-L883,
// if the writer is closed before any Write is called, Close will create an empty file.
func (o *GoCloudObject) Delete() {
if o.bucketURL == "" || o.objectName == "" {
return
}
// Note we can't use the request context because in a successful
// case, the original request has already completed.
deleteCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second) // lint:allow context.Background
defer cancel()
bucket, err := o.mux.OpenBucket(deleteCtx, o.bucketURL)
if err != nil {
log.WithError(err).Error("error opening bucket for delete")
return
}
if err := bucket.Delete(deleteCtx, o.objectName); err != nil {
if gcerrors.Code(err) != gcerrors.NotFound {
log.WithError(err).Error("error deleting object")
}
}
}
|
