forked from perkeep/perkeep
/
ns.go
170 lines (152 loc) · 4.28 KB
/
ns.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
/*
Copyright 2014 The Camlistore Authors
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 namespace implements the "namespace" blobserver storage type.
//
// A namespace storage is backed by another storage target but only
// has access and visibility to a subset of the blobs which have been
// uploaded through this namespace. The list of accessible blobs are
// stored in the provided "inventory" sorted key/value target.
package namespace
import (
"bytes"
"fmt"
"io"
"log"
"os"
"strconv"
"camlistore.org/pkg/blob"
"camlistore.org/pkg/blobserver"
"camlistore.org/pkg/context"
"camlistore.org/pkg/jsonconfig"
"camlistore.org/pkg/sorted"
"camlistore.org/pkg/strutil"
)
type nsto struct {
inventory sorted.KeyValue
master blobserver.Storage
}
func init() {
blobserver.RegisterStorageConstructor("namespace", blobserver.StorageConstructor(newFromConfig))
}
func newFromConfig(ld blobserver.Loader, config jsonconfig.Obj) (storage blobserver.Storage, err error) {
sto := &nsto{}
invConf := config.RequiredObject("inventory")
masterName := config.RequiredString("storage")
if err := config.Validate(); err != nil {
return nil, err
}
sto.inventory, err = sorted.NewKeyValue(invConf)
if err != nil {
return nil, fmt.Errorf("Invalid 'inventory' configuration: %v", err)
}
sto.master, err = ld.GetStorage(masterName)
if err != nil {
return nil, fmt.Errorf("Invalid 'storage' configuration: %v", err)
}
return sto, nil
}
func (ns *nsto) EnumerateBlobs(ctx *context.Context, dest chan<- blob.SizedRef, after string, limit int) error {
defer close(dest)
done := ctx.Done()
it := ns.inventory.Find(after, "")
first := true
for limit > 0 && it.Next() {
if first {
first = false
if after != "" && it.Key() == after {
continue
}
}
br, ok := blob.ParseBytes(it.KeyBytes())
size, err := strutil.ParseUintBytes(it.ValueBytes(), 10, 32)
if !ok || err != nil {
log.Printf("Bogus namespace key %q / value %q", it.Key(), it.Value())
continue
}
select {
case dest <- blob.SizedRef{br, uint32(size)}:
case <-done:
return context.ErrCanceled
}
limit--
}
if err := it.Close(); err != nil {
return err
}
return nil
}
func (ns *nsto) Fetch(br blob.Ref) (rc io.ReadCloser, size uint32, err error) {
invSizeStr, err := ns.inventory.Get(br.String())
if err == sorted.ErrNotFound {
err = os.ErrNotExist
return
}
if err != nil {
return
}
invSize, err := strconv.ParseUint(invSizeStr, 10, 32)
if err != nil {
return
}
rc, size, err = ns.master.Fetch(br)
if err != nil {
return
}
if size != uint32(invSize) {
log.Printf("namespace: on blob %v, unexpected inventory size %d for master size %d", br, invSize, size)
return nil, 0, os.ErrNotExist
}
return rc, size, nil
}
func (ns *nsto) ReceiveBlob(br blob.Ref, src io.Reader) (sb blob.SizedRef, err error) {
var buf bytes.Buffer
size, err := io.Copy(&buf, src)
if err != nil {
return
}
// Check if a duplicate blob, already uploaded previously.
if _, ierr := ns.inventory.Get(br.String()); ierr == nil {
return blob.SizedRef{br, uint32(size)}, nil
}
sb, err = ns.master.ReceiveBlob(br, &buf)
if err != nil {
return
}
err = ns.inventory.Set(br.String(), strconv.Itoa(int(size)))
return
}
func (ns *nsto) RemoveBlobs(blobs []blob.Ref) error {
for _, br := range blobs {
if err := ns.inventory.Delete(br.String()); err != nil {
return err
}
}
return nil
}
func (ns *nsto) StatBlobs(dest chan<- blob.SizedRef, blobs []blob.Ref) error {
for _, br := range blobs {
invSizeStr, err := ns.inventory.Get(br.String())
if err == sorted.ErrNotFound {
continue
}
if err != nil {
return err
}
invSize, err := strconv.ParseUint(invSizeStr, 10, 32)
if err != nil {
log.Printf("Bogus namespace key %q / value %q", br.String(), invSizeStr)
}
dest <- blob.SizedRef{br, uint32(invSize)}
}
return nil
}