@cryptotaxi247 / kubo / commits / 35a5ca0ef

update go-datastore to latest

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Sep 15, 2015 at 17:15 UTC 35a5ca0ef5776573e4b161dde315e602d663e5af
43 files changed +1361 -65
Godeps/Godeps.json
+5 -1
@@ -111,6 +111,10 @@
111 "ImportPath": "github.com/inconshreveable/go-update",
112 "Rev": "68f5725818189545231c1fd8694793d45f2fc529"
113 },
114 + {
115 + "ImportPath": "github.com/ipfs/go-log",
116 + "Rev": "ee5cb9834b33bcf29689183e0323e328c8b8de29"
117 + },
118 {
119 "ImportPath": "github.com/jackpal/go-nat-pmp",
120 "Rev": "a45aa3d54aef73b504e15eb71bea0e5565b5e6e1"
@@ -129,7 +133,7 @@
133 },
134 {
135 "ImportPath": "github.com/jbenet/go-datastore",
132 - "Rev": "7d6acaf7c0164c335f2ca4100f8fe30a7e2943dd"
136 + "Rev": "c835c30f206c1e97172e428f052e225adab9abde"
137 },
138 {
139 "ImportPath": "github.com/jbenet/go-detect-race",
Godeps/_workspace/src/github.com/ipfs/go-log/.gxlastpubver new
+1
@@ -0,0 +1 @@
1 +QmXJkcEXB6C9h6Ytb6rrUTFU56Ro62zxgrbxTT3dgjQGA8
\ No newline at end of file
Godeps/_workspace/src/github.com/ipfs/go-log/context.go new
+38
@@ -0,0 +1,38 @@
1 +package log
2 +
3 +import (
4 + "errors"
5 +
6 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
7 +)
8 +
9 +type key int
10 +
11 +const metadataKey key = 0
12 +
13 +// ContextWithLoggable returns a derived context which contains the provided
14 +// Loggable. Any Events logged with the derived context will include the
15 +// provided Loggable.
16 +func ContextWithLoggable(ctx context.Context, l Loggable) context.Context {
17 + existing, err := MetadataFromContext(ctx)
18 + if err != nil {
19 + // context does not contain meta. just set the new metadata
20 + child := context.WithValue(ctx, metadataKey, Metadata(l.Loggable()))
21 + return child
22 + }
23 +
24 + merged := DeepMerge(existing, l.Loggable())
25 + child := context.WithValue(ctx, metadataKey, merged)
26 + return child
27 +}
28 +
29 +func MetadataFromContext(ctx context.Context) (Metadata, error) {
30 + value := ctx.Value(metadataKey)
31 + if value != nil {
32 + metadata, ok := value.(Metadata)
33 + if ok {
34 + return metadata, nil
35 + }
36 + }
37 + return nil, errors.New("context contains no metadata")
38 +}
Godeps/_workspace/src/github.com/ipfs/go-log/context_test.go new
+44
@@ -0,0 +1,44 @@
1 +package log
2 +
3 +import (
4 + "testing"
5 +
6 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
7 +)
8 +
9 +func TestContextContainsMetadata(t *testing.T) {
10 + t.Parallel()
11 +
12 + m := Metadata{"foo": "bar"}
13 + ctx := ContextWithLoggable(context.Background(), m)
14 + got, err := MetadataFromContext(ctx)
15 + if err != nil {
16 + t.Fatal(err)
17 + }
18 +
19 + _, exists := got["foo"]
20 + if !exists {
21 + t.Fail()
22 + }
23 +}
24 +
25 +func TestContextWithPreexistingMetadata(t *testing.T) {
26 + t.Parallel()
27 +
28 + ctx := ContextWithLoggable(context.Background(), Metadata{"hello": "world"})
29 + ctx = ContextWithLoggable(ctx, Metadata{"goodbye": "earth"})
30 +
31 + got, err := MetadataFromContext(ctx)
32 + if err != nil {
33 + t.Fatal(err)
34 + }
35 +
36 + _, exists := got["hello"]
37 + if !exists {
38 + t.Fatal("original key not present")
39 + }
40 + _, exists = got["goodbye"]
41 + if !exists {
42 + t.Fatal("new key not present")
43 + }
44 +}
Godeps/_workspace/src/github.com/ipfs/go-log/entry.go new
+42
@@ -0,0 +1,42 @@
1 +package log
2 +
3 +import (
4 + "time"
5 +
6 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/Sirupsen/logrus"
7 +)
8 +
9 +type entry struct {
10 + loggables []Loggable
11 + system string
12 + event string
13 +}
14 +
15 +// Log logs the event unconditionally (regardless of log level)
16 +// TODO add support for leveled-logs once we decide which levels we want
17 +// for our structured logs
18 +func (e *entry) Log() {
19 + e.log()
20 +}
21 +
22 +// log is a private method invoked by the public Log, Info, Error methods
23 +func (e *entry) log() {
24 + // accumulate metadata
25 + accum := Metadata{}
26 + for _, loggable := range e.loggables {
27 + accum = DeepMerge(accum, loggable.Loggable())
28 + }
29 +
30 + // apply final attributes to reserved keys
31 + // TODO accum["level"] = level
32 + accum["event"] = e.event
33 + accum["system"] = e.system
34 + accum["time"] = FormatRFC3339(time.Now())
35 +
36 + // TODO roll our own event logger
37 + logrus.WithFields(map[string]interface{}(accum)).Info(e.event)
38 +}
39 +
40 +func FormatRFC3339(t time.Time) string {
41 + return t.UTC().Format(time.RFC3339Nano)
42 +}
Godeps/_workspace/src/github.com/ipfs/go-log/example_test.go new
+16
@@ -0,0 +1,16 @@
1 +package log
2 +
3 +import "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
4 +
5 +func ExampleEventLogger() {
6 + {
7 + log := EventLogger(nil)
8 + e := log.EventBegin(context.Background(), "dial")
9 + e.Done()
10 + }
11 + {
12 + log := EventLogger(nil)
13 + e := log.EventBegin(context.Background(), "dial")
14 + _ = e.Close() // implements io.Closer for convenience
15 + }
16 +}
Godeps/_workspace/src/github.com/ipfs/go-log/log.go new
+149
@@ -0,0 +1,149 @@
1 +package log
2 +
3 +import (
4 + "fmt"
5 + "time"
6 +
7 + context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
8 +)
9 +
10 +// StandardLogger provides API compatibility with standard printf loggers
11 +// eg. go-logging
12 +type StandardLogger interface {
13 + Debug(args ...interface{})
14 + Debugf(format string, args ...interface{})
15 + Error(args ...interface{})
16 + Errorf(format string, args ...interface{})
17 + Fatal(args ...interface{})
18 + Fatalf(format string, args ...interface{})
19 + Info(args ...interface{})
20 + Infof(format string, args ...interface{})
21 + Panic(args ...interface{})
22 + Panicf(format string, args ...interface{})
23 + Warning(args ...interface{})
24 + Warningf(format string, args ...interface{})
25 +}
26 +
27 +// EventLogger extends the StandardLogger interface to allow for log items
28 +// containing structured metadata
29 +type EventLogger interface {
30 + StandardLogger
31 +
32 + // Event merges structured data from the provided inputs into a single
33 + // machine-readable log event.
34 + //
35 + // If the context contains metadata, a copy of this is used as the base
36 + // metadata accumulator.
37 + //
38 + // If one or more loggable objects are provided, these are deep-merged into base blob.
39 + //
40 + // Next, the event name is added to the blob under the key "event". If
41 + // the key "event" already exists, it will be over-written.
42 + //
43 + // Finally the timestamp and package name are added to the accumulator and
44 + // the metadata is logged.
45 + Event(ctx context.Context, event string, m ...Loggable)
46 +
47 + EventBegin(ctx context.Context, event string, m ...Loggable) *EventInProgress
48 +}
49 +
50 +// Logger retrieves an event logger by name
51 +func Logger(system string) EventLogger {
52 +
53 + // TODO if we would like to adjust log levels at run-time. Store this event
54 + // logger in a map (just like the util.Logger impl)
55 + if len(system) == 0 {
56 + log.Warnf("Missing name parameter")
57 + system = "undefined"
58 + }
59 + if _, ok := loggers[system]; !ok {
60 + loggers[system] = log.WithField("module", system)
61 + }
62 + logger := loggers[system]
63 +
64 + return &eventLogger{system: system, StandardLogger: logger}
65 +}
66 +
67 +// eventLogger implements the EventLogger and wraps a go-logging Logger
68 +type eventLogger struct {
69 + StandardLogger
70 +
71 + system string
72 + // TODO add log-level
73 +}
74 +
75 +func (el *eventLogger) EventBegin(ctx context.Context, event string, metadata ...Loggable) *EventInProgress {
76 + start := time.Now()
77 + el.Event(ctx, fmt.Sprintf("%sBegin", event), metadata...)
78 +
79 + eip := &EventInProgress{}
80 + eip.doneFunc = func(additional []Loggable) {
81 +
82 + metadata = append(metadata, additional...) // anything added during the operation
83 + metadata = append(metadata, LoggableMap(map[string]interface{}{ // finally, duration of event
84 + "duration": time.Now().Sub(start),
85 + }))
86 +
87 + el.Event(ctx, event, metadata...)
88 + }
89 + return eip
90 +}
91 +
92 +func (el *eventLogger) Event(ctx context.Context, event string, metadata ...Loggable) {
93 +
94 + // short circuit if theres nothing to write to
95 + if !WriterGroup.Active() {
96 + return
97 + }
98 +
99 + // Collect loggables for later logging
100 + var loggables []Loggable
101 +
102 + // get any existing metadata from the context
103 + existing, err := MetadataFromContext(ctx)
104 + if err != nil {
105 + existing = Metadata{}
106 + }
107 + loggables = append(loggables, existing)
108 +
109 + for _, datum := range metadata {
110 + loggables = append(loggables, datum)
111 + }
112 +
113 + e := entry{
114 + loggables: loggables,
115 + system: el.system,
116 + event: event,
117 + }
118 +
119 + e.Log() // TODO replace this when leveled-logs have been implemented
120 +}
121 +
122 +type EventInProgress struct {
123 + loggables []Loggable
124 + doneFunc func([]Loggable)
125 +}
126 +
127 +// Append adds loggables to be included in the call to Done
128 +func (eip *EventInProgress) Append(l Loggable) {
129 + eip.loggables = append(eip.loggables, l)
130 +}
131 +
132 +// SetError includes the provided error
133 +func (eip *EventInProgress) SetError(err error) {
134 + eip.loggables = append(eip.loggables, LoggableMap{
135 + "error": err.Error(),
136 + })
137 +}
138 +
139 +// Done creates a new Event entry that includes the duration and appended
140 +// loggables.
141 +func (eip *EventInProgress) Done() {
142 + eip.doneFunc(eip.loggables) // create final event with extra data
143 +}
144 +
145 +// Close is an alias for done
146 +func (eip *EventInProgress) Close() error {
147 + eip.Done()
148 + return nil
149 +}
Godeps/_workspace/src/github.com/ipfs/go-log/loggable.go new
+34
@@ -0,0 +1,34 @@
1 +package log
2 +
3 +// Loggable describes objects that can be marshalled into Metadata for logging
4 +type Loggable interface {
5 + Loggable() map[string]interface{}
6 +}
7 +
8 +type LoggableMap map[string]interface{}
9 +
10 +func (l LoggableMap) Loggable() map[string]interface{} {
11 + return l
12 +}
13 +
14 +// LoggableF converts a func into a Loggable
15 +type LoggableF func() map[string]interface{}
16 +
17 +func (l LoggableF) Loggable() map[string]interface{} {
18 + return l()
19 +}
20 +
21 +func Deferred(key string, f func() string) Loggable {
22 + function := func() map[string]interface{} {
23 + return map[string]interface{}{
24 + key: f(),
25 + }
26 + }
27 + return LoggableF(function)
28 +}
29 +
30 +func Pair(key string, l Loggable) Loggable {
31 + return LoggableMap{
32 + key: l,
33 + }
34 +}
Godeps/_workspace/src/github.com/ipfs/go-log/metadata.go new
+82
@@ -0,0 +1,82 @@
1 +package log
2 +
3 +import (
4 + "encoding/json"
5 + "errors"
6 + "reflect"
7 +
8 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/satori/go.uuid"
9 +)
10 +
11 +// Metadata is a convenience type for generic maps
12 +type Metadata map[string]interface{}
13 +
14 +// Uuid returns a Metadata with the string key and UUID value
15 +func Uuid(key string) Metadata {
16 + return Metadata{
17 + key: uuid.NewV4().String(),
18 + }
19 +}
20 +
21 +// DeepMerge merges the second Metadata parameter into the first.
22 +// Nested Metadata are merged recursively. Primitives are over-written.
23 +func DeepMerge(b, a Metadata) Metadata {
24 + out := Metadata{}
25 + for k, v := range b {
26 + out[k] = v
27 + }
28 + for k, v := range a {
29 +
30 + maybe, err := Metadatify(v)
31 + if err != nil {
32 + // if the new value is not meta. just overwrite the dest vaue
33 + out[k] = v
34 + continue
35 + }
36 +
37 + // it is meta. What about dest?
38 + outv, exists := out[k]
39 + if !exists {
40 + // the new value is meta, but there's no dest value. just write it
41 + out[k] = v
42 + continue
43 + }
44 +
45 + outMetadataValue, err := Metadatify(outv)
46 + if err != nil {
47 + // the new value is meta and there's a dest value, but the dest
48 + // value isn't meta. just overwrite
49 + out[k] = v
50 + continue
51 + }
52 +
53 + // both are meta. merge them.
54 + out[k] = DeepMerge(outMetadataValue, maybe)
55 + }
56 + return out
57 +}
58 +
59 +// Loggable implements the Loggable interface
60 +func (m Metadata) Loggable() map[string]interface{} {
61 + // NB: method defined on value to avoid de-referencing nil Metadata
62 + return m
63 +}
64 +
65 +func (m Metadata) JsonString() (string, error) {
66 + // NB: method defined on value
67 + b, err := json.Marshal(m)
68 + return string(b), err
69 +}
70 +
71 +// Metadatify converts maps into Metadata
72 +func Metadatify(i interface{}) (Metadata, error) {
73 + value := reflect.ValueOf(i)
74 + if value.Kind() == reflect.Map {
75 + m := map[string]interface{}{}
76 + for _, k := range value.MapKeys() {
77 + m[k.String()] = value.MapIndex(k).Interface()
78 + }
79 + return Metadata(m), nil
80 + }
81 + return nil, errors.New("is not a map")
82 +}
Godeps/_workspace/src/github.com/ipfs/go-log/metadata_test.go new
+50
@@ -0,0 +1,50 @@
1 +package log
2 +
3 +import "testing"
4 +
5 +func TestOverwrite(t *testing.T) {
6 + t.Parallel()
7 +
8 + under := Metadata{
9 + "a": Metadata{
10 + "b": Metadata{
11 + "c": Metadata{
12 + "d": "the original value",
13 + "other": "SURVIVE",
14 + },
15 + },
16 + },
17 + }
18 + over := Metadata{
19 + "a": Metadata{
20 + "b": Metadata{
21 + "c": Metadata{
22 + "d": "a new value",
23 + },
24 + },
25 + },
26 + }
27 +
28 + out := DeepMerge(under, over)
29 +
30 + dval := out["a"].(Metadata)["b"].(Metadata)["c"].(Metadata)["d"].(string)
31 + if dval != "a new value" {
32 + t.Fatal(dval)
33 + }
34 + surv := out["a"].(Metadata)["b"].(Metadata)["c"].(Metadata)["other"].(string)
35 + if surv != "SURVIVE" {
36 + t.Fatal(surv)
37 + }
38 +}
39 +
40 +func TestMarshalJSON(t *testing.T) {
41 + t.Parallel()
42 + bs, _ := Metadata{"a": "b"}.JsonString()
43 + t.Log(bs)
44 +}
45 +
46 +func TestMetadataIsLoggable(t *testing.T) {
47 + t.Parallel()
48 + func(l Loggable) {
49 + }(Metadata{})
50 +}
Godeps/_workspace/src/github.com/ipfs/go-log/oldlog.go new
+97
@@ -0,0 +1,97 @@
1 +package log
2 +
3 +import (
4 + "errors"
5 + "os"
6 +
7 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/Sirupsen/logrus"
8 +)
9 +
10 +func init() {
11 + SetupLogging()
12 +}
13 +
14 +var log = logrus.New()
15 +
16 +// LogFormats is a map of formats used for our logger, keyed by name.
17 +// TODO: write custom TextFormatter (don't print module=name explicitly) and
18 +// fork logrus to add shortfile
19 +var LogFormats = map[string]*logrus.TextFormatter{
20 + "nocolor": {DisableColors: true, FullTimestamp: true, TimestampFormat: "2006-01-02 15:04:05.000000", DisableSorting: true},
21 + "color": {DisableColors: false, FullTimestamp: true, TimestampFormat: "15:04:05:000", DisableSorting: true},
22 +}
23 +var defaultLogFormat = "color"
24 +
25 +// Logging environment variables
26 +const (
27 + envLogging = "IPFS_LOGGING"
28 + envLoggingFmt = "IPFS_LOGGING_FMT"
29 +)
30 +
31 +// ErrNoSuchLogger is returned when the util pkg is asked for a non existant logger
32 +var ErrNoSuchLogger = errors.New("Error: No such logger")
33 +
34 +// loggers is the set of loggers in the system
35 +var loggers = map[string]*logrus.Entry{}
36 +
37 +// SetupLogging will initialize the logger backend and set the flags.
38 +func SetupLogging() {
39 +
40 + format, ok := LogFormats[os.Getenv(envLoggingFmt)]
41 + if !ok {
42 + format = LogFormats[defaultLogFormat]
43 + }
44 +
45 + log.Out = os.Stderr
46 + log.Formatter = format
47 +
48 + lvl := logrus.ErrorLevel
49 +
50 + if logenv := os.Getenv(envLogging); logenv != "" {
51 + var err error
52 + lvl, err = logrus.ParseLevel(logenv)
53 + if err != nil {
54 + log.Debugf("logrus.ParseLevel() Error: %q", err)
55 + lvl = logrus.ErrorLevel // reset to ERROR, could be undefined now(?)
56 + }
57 + }
58 +
59 + SetAllLoggers(lvl)
60 +}
61 +
62 +// SetDebugLogging calls SetAllLoggers with logrus.DebugLevel
63 +func SetDebugLogging() {
64 + SetAllLoggers(logrus.DebugLevel)
65 +}
66 +
67 +// SetAllLoggers changes the logrus.Level of all loggers to lvl
68 +func SetAllLoggers(lvl logrus.Level) {
69 + log.Level = lvl
70 + for _, logger := range loggers {
71 + logger.Level = lvl
72 + }
73 +}
74 +
75 +// SetLogLevel changes the log level of a specific subsystem
76 +// name=="*" changes all subsystems
77 +func SetLogLevel(name, level string) error {
78 + lvl, err := logrus.ParseLevel(level)
79 + if err != nil {
80 + return err
81 + }
82 +
83 + // wildcard, change all
84 + if name == "*" {
85 + SetAllLoggers(lvl)
86 + return nil
87 + }
88 +
89 + // Check if we have a logger by that name
90 + if _, ok := loggers[name]; !ok {
91 + return ErrNoSuchLogger
92 + }
93 +
94 + loggers[name].Level = lvl
95 +
96 + return nil
97 +}
Godeps/_workspace/src/github.com/ipfs/go-log/option.go new
+61
@@ -0,0 +1,61 @@
1 +package log
2 +
3 +import (
4 + "io"
5 + "os"
6 +
7 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/Sirupsen/logrus"
8 +)
9 +
10 +// init sets up sane defaults
11 +func init() {
12 + Configure(TextFormatter)
13 + Configure(Output(os.Stderr))
14 + // has the effect of disabling logging since we log event entries at Info
15 + // level by convention
16 + Configure(LevelError)
17 +}
18 +
19 +// Global writer group for logs to output to
20 +var WriterGroup = new(MirrorWriter)
21 +
22 +type Option func()
23 +
24 +// Configure applies the provided options sequentially from left to right
25 +func Configure(options ...Option) {
26 + for _, f := range options {
27 + f()
28 + }
29 +}
30 +
31 +// LdJSONFormatter Option formats the event log as line-delimited JSON
32 +var LdJSONFormatter = func() {
33 + logrus.SetFormatter(&PoliteJSONFormatter{})
34 +}
35 +
36 +// TextFormatter Option formats the event log as human-readable plain-text
37 +var TextFormatter = func() {
38 + logrus.SetFormatter(&logrus.TextFormatter{})
39 +}
40 +
41 +func Output(w io.Writer) Option {
42 + return func() {
43 + logrus.SetOutput(w)
44 + // TODO return previous Output option
45 + }
46 +}
47 +
48 +// LevelDebug Option sets the log level to debug
49 +var LevelDebug = func() {
50 + logrus.SetLevel(logrus.DebugLevel)
51 +}
52 +
53 +// LevelDebug Option sets the log level to error
54 +var LevelError = func() {
55 + logrus.SetLevel(logrus.ErrorLevel)
56 +}
57 +
58 +// LevelDebug Option sets the log level to info
59 +var LevelInfo = func() {
60 + logrus.SetLevel(logrus.InfoLevel)
61 +}
Godeps/_workspace/src/github.com/ipfs/go-log/package.json new
+5
@@ -0,0 +1,5 @@
1 +{
2 + "name": "go-log",
3 + "version": "1.0.0",
4 + "language": "go"
5 +}
\ No newline at end of file
Godeps/_workspace/src/github.com/ipfs/go-log/polite_json_formatter.go new
+20
@@ -0,0 +1,20 @@
1 +package log
2 +
3 +import (
4 + "encoding/json"
5 + "fmt"
6 +
7 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/Sirupsen/logrus"
8 +)
9 +
10 +// PoliteJSONFormatter marshals entries into JSON encoded slices (without
11 +// overwriting user-provided keys). How polite of it!
12 +type PoliteJSONFormatter struct{}
13 +
14 +func (f *PoliteJSONFormatter) Format(entry *logrus.Entry) ([]byte, error) {
15 + serialized, err := json.Marshal(entry.Data)
16 + if err != nil {
17 + return nil, fmt.Errorf("Failed to marshal fields to JSON, %v", err)
18 + }
19 + return append(serialized, '\n'), nil
20 +}
Godeps/_workspace/src/github.com/ipfs/go-log/writer.go new
+53
@@ -0,0 +1,53 @@
1 +package log
2 +
3 +import (
4 + "io"
5 + "sync"
6 +)
7 +
8 +type MirrorWriter struct {
9 + writers []io.Writer
10 + lk sync.Mutex
11 +}
12 +
13 +func (mw *MirrorWriter) Write(b []byte) (int, error) {
14 + mw.lk.Lock()
15 + // write to all writers, and nil out the broken ones.
16 + for i, w := range mw.writers {
17 + _, err := w.Write(b)
18 + if err != nil {
19 + mw.writers[i] = nil
20 + }
21 + }
22 +
23 + // consolidate the slice
24 + for i := 0; i < len(mw.writers); i++ {
25 + if mw.writers[i] != nil {
26 + continue
27 + }
28 +
29 + j := len(mw.writers) - 1
30 + for ; j > i; j-- {
31 + if mw.writers[j] != nil {
32 + mw.writers[i], mw.writers[j] = mw.writers[j], nil // swap
33 + break
34 + }
35 + }
36 + mw.writers = mw.writers[:j]
37 + }
38 + mw.lk.Unlock()
39 + return len(b), nil
40 +}
41 +
42 +func (mw *MirrorWriter) AddWriter(w io.Writer) {
43 + mw.lk.Lock()
44 + mw.writers = append(mw.writers, w)
45 + mw.lk.Unlock()
46 +}
47 +
48 +func (mw *MirrorWriter) Active() (active bool) {
49 + mw.lk.Lock()
50 + active = len(mw.writers) > 0
51 + mw.lk.Unlock()
52 + return
53 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/Godeps/Godeps.json
+19 -1
@@ -1,10 +1,15 @@
1 {
2 "ImportPath": "github.com/jbenet/go-datastore",
3 - "GoVersion": "go1.4.2",
3 + "GoVersion": "go1.5",
4 "Packages": [
5 "./..."
6 ],
7 "Deps": [
8 + {
9 + "ImportPath": "github.com/Sirupsen/logrus",
10 + "Comment": "v0.8.3-37-g418b41d",
11 + "Rev": "418b41d23a1bf978c06faea5313ba194650ac088"
12 + },
13 {
14 "ImportPath": "github.com/codahale/blake2",
15 "Rev": "3fa823583afba430e8fc7cdbcc670dbf90bfacc4"
@@ -21,10 +26,19 @@
26 "ImportPath": "github.com/dustin/randbo",
27 "Rev": "7f1b564ca7242d22bcc6e2128beb90d9fa38b9f0"
28 },
29 + {
30 + "ImportPath": "github.com/fzzy/radix/redis",
31 + "Comment": "v0.5.1",
32 + "Rev": "27a863cdffdb0998d13e1e11992b18489aeeaa25"
33 + },
34 {
35 "ImportPath": "github.com/hashicorp/golang-lru",
36 "Rev": "4dfff096c4973178c8f35cf6dd1a732a0a139370"
37 },
38 + {
39 + "ImportPath": "github.com/ipfs/go-log",
40 + "Rev": "ee5cb9834b33bcf29689183e0323e328c8b8de29"
41 + },
42 {
43 "ImportPath": "github.com/jbenet/go-os-rename",
44 "Rev": "2d93ae970ba96c41f717036a5bf5494faf1f38c0"
@@ -53,6 +67,10 @@
67 "ImportPath": "github.com/syndtr/gosnappy/snappy",
68 "Rev": "ce8acff4829e0c2458a67ead32390ac0a381c862"
69 },
70 + {
71 + "ImportPath": "golang.org/x/net/context",
72 + "Rev": "dfcbca9c45aeabb8971affa4f76b2d40f6f72328"
73 + },
74 {
75 "ImportPath": "gopkg.in/check.v1",
76 "Rev": "91ae5f88a67b14891cfd43895b01164f6c120420"
Godeps/_workspace/src/github.com/jbenet/go-datastore/basic_ds.go
+20 -4
@@ -1,6 +1,7 @@
1 package datastore
2
3 import (
4 + "io"
5 "log"
6
7 dsq "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
@@ -67,6 +68,10 @@ func (d *MapDatastore) Batch() (Batch, error) {
68 return NewBasicBatch(d), nil
69 }
70
71 +func (d *MapDatastore) Close() error {
72 + return nil
73 +}
74 +
75 // NullDatastore stores nothing, but conforms to the API.
76 // Useful to test with.
77 type NullDatastore struct {
@@ -106,6 +111,10 @@ func (d *NullDatastore) Batch() (Batch, error) {
111 return NewBasicBatch(d), nil
112 }
113
114 +func (d *NullDatastore) Close() error {
115 + return nil
116 +}
117 +
118 // LogDatastore logs all accesses through the datastore.
119 type LogDatastore struct {
120 Name string
@@ -165,9 +174,16 @@ func (d *LogDatastore) Query(q dsq.Query) (dsq.Results, error) {
174
175 func (d *LogDatastore) Batch() (Batch, error) {
176 log.Printf("%s: Batch\n", d.Name)
168 - bds, ok := d.child.(BatchingDatastore)
169 - if !ok {
170 - return nil, ErrBatchUnsupported
177 + if bds, ok := d.child.(Batching); ok {
178 + return bds.Batch()
179 }
172 - return bds.Batch()
180 + return nil, ErrBatchUnsupported
181 +}
182 +
183 +func (d *LogDatastore) Close() error {
184 + log.Printf("%s: Close\n", d.Name)
185 + if cds, ok := d.child.(io.Closer); ok {
186 + return cds.Close()
187 + }
188 + return nil
189 }
Godeps/_workspace/src/github.com/jbenet/go-datastore/coalesce/coalesce.go
+14
@@ -1,6 +1,7 @@
1 package coalesce
2
3 import (
4 + "io"
5 "sync"
6
7 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
@@ -124,3 +125,16 @@ func (d *datastore) Query(q dsq.Query) (dsq.Results, error) {
125 // query not coalesced yet.
126 return d.child.Query(q)
127 }
128 +
129 +func (d *datastore) Close() error {
130 + d.reqmu.Lock()
131 + defer d.reqmu.Unlock()
132 +
133 + for _, s := range d.req {
134 + <-s.done
135 + }
136 + if c, ok := d.child.(io.Closer); ok {
137 + return c.Close()
138 + }
139 + return nil
140 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/datastore.go
+1 -1
@@ -69,7 +69,7 @@ type Datastore interface {
69 Query(q query.Query) (query.Results, error)
70 }
71
72 -type BatchingDatastore interface {
72 +type Batching interface {
73 Datastore
74
75 Batch() (Batch, error)
Godeps/_workspace/src/github.com/jbenet/go-datastore/flatfs/flatfs.go
+28
@@ -10,12 +10,16 @@ import (
10 "os"
11 "path"
12 "strings"
13 + "time"
14
15 + logging "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/ipfs/go-log"
16 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
17 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
18 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-os-rename"
19 )
20
21 +var log = logging.Logger("flatfs")
22 +
23 const (
24 extension = ".data"
25 maxPrefixLen = 16
@@ -93,12 +97,32 @@ func (fs *Datastore) makePrefixDirNoSync(dir string) error {
97 return nil
98 }
99
100 +var putMaxRetries = 3
101 +
102 func (fs *Datastore) Put(key datastore.Key, value interface{}) error {
103 val, ok := value.([]byte)
104 if !ok {
105 return datastore.ErrInvalidType
106 }
107
108 + var err error
109 + for i := 0; i < putMaxRetries; i++ {
110 + err = fs.doPut(key, val)
111 + if err == nil {
112 + return nil
113 + }
114 +
115 + if !strings.Contains(err.Error(), "too many open files") {
116 + return err
117 + }
118 +
119 + log.Error("too many open files, retrying in %dms", 100*i)
120 + time.Sleep(time.Millisecond * 100 * time.Duration(i))
121 + }
122 + return err
123 +}
124 +
125 +func (fs *Datastore) doPut(key datastore.Key, val []byte) error {
126 dir, path := fs.encode(key)
127 if err := fs.makePrefixDir(dir); err != nil {
128 return err
@@ -323,6 +347,10 @@ func (fs *Datastore) enumerateKeys(fi os.FileInfo, res []query.Entry) ([]query.E
347 return res, nil
348 }
349
350 +func (fs *Datastore) Close() error {
351 + return nil
352 +}
353 +
354 type flatfsBatch struct {
355 puts map[datastore.Key]interface{}
356 deletes map[datastore.Key]struct{}
Godeps/_workspace/src/github.com/jbenet/go-datastore/fs/fs.go
+8
@@ -149,3 +149,11 @@ func isFile(path string) bool {
149
150 return !finfo.IsDir()
151 }
152 +
153 +func (d *Datastore) Close() error {
154 + return nil
155 +}
156 +
157 +func (d *Datastore) Batch() (ds.Batch, error) {
158 + return ds.NewBasicBatch(d), nil
159 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/keytransform/interface.go
+1 -3
@@ -16,14 +16,12 @@ type KeyTransform interface {
16 type Datastore interface {
17 ds.Shim
18 KeyTransform
19 -
20 - Batch() (ds.Batch, error)
19 }
20
21 // Wrap wraps a given datastore with a KeyTransform function.
22 // The resulting wrapped datastore will use the transform on all Datastore
23 // operations.
26 -func Wrap(child ds.Datastore, t KeyTransform) Datastore {
24 +func Wrap(child ds.Datastore, t KeyTransform) *ktds {
25 if t == nil {
26 panic("t (KeyTransform) is nil")
27 }
Godeps/_workspace/src/github.com/jbenet/go-datastore/keytransform/keytransform.go
+10 -1
@@ -1,6 +1,8 @@
1 package keytransform
2
3 import (
4 + "io"
5 +
6 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
7 dsq "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
8 )
@@ -74,8 +76,15 @@ func (d *ktds) Query(q dsq.Query) (dsq.Results, error) {
76 return dsq.DerivedResults(qr, ch), nil
77 }
78
79 +func (d *ktds) Close() error {
80 + if c, ok := d.child.(io.Closer); ok {
81 + return c.Close()
82 + }
83 + return nil
84 +}
85 +
86 func (d *ktds) Batch() (ds.Batch, error) {
78 - bds, ok := d.child.(ds.BatchingDatastore)
87 + bds, ok := d.child.(ds.Batching)
88 if !ok {
89 return nil, ds.ErrBatchUnsupported
90 }
Godeps/_workspace/src/github.com/jbenet/go-datastore/leveldb/datastore.go
+6 -8
@@ -1,8 +1,6 @@
1 package leveldb
2
3 import (
4 - "io"
5 -
4 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
5 dsq "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
6 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
@@ -11,18 +9,13 @@ import (
9 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/syndtr/goleveldb/leveldb/util"
10 )
11
14 -type Datastore interface {
15 - ds.ThreadSafeDatastore
16 - io.Closer
17 -}
18 -
12 type datastore struct {
13 DB *leveldb.DB
14 }
15
16 type Options opt.Options
17
25 -func NewDatastore(path string, opts *Options) (Datastore, error) {
18 +func NewDatastore(path string, opts *Options) (*datastore, error) {
19 var nopts opt.Options
20 if opts != nil {
21 nopts = opt.Options(*opts)
@@ -148,6 +141,11 @@ func (d *datastore) runQuery(worker goprocess.Process, qrb *dsq.ResultBuilder) {
141 }
142 }
143
144 +func (d *datastore) Batch() (ds.Batch, error) {
145 + // TODO: implement batch on leveldb
146 + return nil, ds.ErrBatchUnsupported
147 +}
148 +
149 // LevelDB needs to be closed.
150 func (d *datastore) Close() (err error) {
151 return d.DB.Close()
Godeps/_workspace/src/github.com/jbenet/go-datastore/leveldb/ds_test.go
+2 -2
@@ -25,7 +25,7 @@ var testcases = map[string]string{
25 //
26 // d, close := newDS(t)
27 // defer close()
28 -func newDS(t *testing.T) (Datastore, func()) {
28 +func newDS(t *testing.T) (*datastore, func()) {
29 path, err := ioutil.TempDir("/tmp", "testing_leveldb_")
30 if err != nil {
31 t.Fatal(err)
@@ -41,7 +41,7 @@ func newDS(t *testing.T) (Datastore, func()) {
41 }
42 }
43
44 -func addTestCases(t *testing.T, d Datastore, testcases map[string]string) {
44 +func addTestCases(t *testing.T, d *datastore, testcases map[string]string) {
45 for k, v := range testcases {
46 dsk := ds.NewKey(k)
47 if err := d.Put(dsk, []byte(v)); err != nil {
Godeps/_workspace/src/github.com/jbenet/go-datastore/lru/datastore.go
+8
@@ -54,3 +54,11 @@ func (d *Datastore) Delete(key ds.Key) (err error) {
54 func (d *Datastore) Query(q dsq.Query) (dsq.Results, error) {
55 return nil, errors.New("KeyList not implemented.")
56 }
57 +
58 +func (d *Datastore) Close() error {
59 + return nil
60 +}
61 +
62 +func (d *Datastore) Batch() (ds.Batch, error) {
63 + return nil, ds.ErrBatchUnsupported
64 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/measure/measure.go
+7 -8
@@ -3,6 +3,7 @@
3 package measure
4
5 import (
6 + "io"
7 "time"
8
9 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/codahale/metrics"
@@ -18,17 +19,12 @@ const (
19 maxSize = int64(1 << 32)
20 )
21
21 -type DatastoreCloser interface {
22 - datastore.Datastore
23 - Close() error
24 -}
25 -
22 // New wraps the datastore, providing metrics on the operations. The
23 // metrics are registered with names starting with prefix and a dot.
24 //
25 // If prefix is not unique, New will panic. Call Close to release the
26 // prefix.
31 -func New(prefix string, ds datastore.Datastore) DatastoreCloser {
27 +func New(prefix string, ds datastore.Datastore) *measure {
28 m := &measure{
29 backend: ds,
30
@@ -84,7 +80,6 @@ type measure struct {
80 }
81
82 var _ datastore.Datastore = (*measure)(nil)
87 -var _ DatastoreCloser = (*measure)(nil)
83
84 func recordLatency(h *metrics.Histogram, start time.Time) {
85 elapsed := time.Now().Sub(start) / time.Microsecond
@@ -159,7 +154,7 @@ type measuredBatch struct {
154 }
155
156 func (m *measure) Batch() (datastore.Batch, error) {
162 - bds, ok := m.backend.(datastore.BatchingDatastore)
157 + bds, ok := m.backend.(datastore.Batching)
158 if !ok {
159 return nil, datastore.ErrBatchUnsupported
160 }
@@ -245,5 +240,9 @@ func (m *measure) Close() error {
240 m.queryNum.Remove()
241 m.queryErr.Remove()
242 m.queryLatency.Remove()
243 +
244 + if c, ok := m.backend.(io.Closer); ok {
245 + return c.Close()
246 + }
247 return nil
248 }
Godeps/_workspace/src/github.com/jbenet/go-datastore/mount/mount.go
+14 -1
@@ -4,6 +4,7 @@ package mount
4
5 import (
6 "errors"
7 + "io"
8 "strings"
9
10 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
@@ -115,6 +116,18 @@ func (d *Datastore) Query(q query.Query) (query.Results, error) {
116 return r, nil
117 }
118
119 +func (d *Datastore) Close() error {
120 + for _, d := range d.mounts {
121 + if c, ok := d.Datastore.(io.Closer); ok {
122 + err := c.Close()
123 + if err != nil {
124 + return err
125 + }
126 + }
127 + }
128 + return nil
129 +}
130 +
131 type mountBatch struct {
132 mounts map[string]datastore.Batch
133
@@ -132,7 +145,7 @@ func (mt *mountBatch) lookupBatch(key datastore.Key) (datastore.Batch, datastore
145 child, loc, rest := mt.d.lookup(key)
146 t, ok := mt.mounts[loc.String()]
147 if !ok {
135 - bds, ok := child.(datastore.BatchingDatastore)
148 + bds, ok := child.(datastore.Batching)
149 if !ok {
150 return nil, datastore.NewKey(""), datastore.ErrBatchUnsupported
151 }
Godeps/_workspace/src/github.com/jbenet/go-datastore/namespace/namespace.go
+9 -1
@@ -36,7 +36,7 @@ func PrefixTransform(prefix ds.Key) ktds.KeyTransform {
36 }
37
38 // Wrap wraps a given datastore with a key-prefix.
39 -func Wrap(child ds.Datastore, prefix ds.Key) ktds.Datastore {
39 +func Wrap(child ds.Datastore, prefix ds.Key) *datastore {
40 if child == nil {
41 panic("child (ds.Datastore) is nil")
42 }
@@ -81,3 +81,11 @@ func (d *datastore) Query(q dsq.Query) (dsq.Results, error) {
81
82 return dsq.DerivedResults(qr, ch), nil
83 }
84 +
85 +func (d *datastore) Batch() (ds.Batch, error) {
86 + if bds, ok := d.Datastore.(ds.Batching); ok {
87 + return bds.Batch()
88 + }
89 +
90 + return nil, ds.ErrBatchUnsupported
91 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/panic/panic.go
+21
@@ -2,6 +2,7 @@ package sync
2
3 import (
4 "fmt"
5 + "io"
6 "os"
7
8 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
@@ -67,6 +68,26 @@ func (d *datastore) Query(q dsq.Query) (dsq.Results, error) {
68 return r, nil
69 }
70
71 +func (d *datastore) Close() error {
72 + if c, ok := d.child.(io.Closer); ok {
73 + err := c.Close()
74 + if err != nil {
75 + fmt.Fprintf(os.Stdout, "panic datastore: %s", err)
76 + panic("panic datastore: Close failed")
77 + }
78 + }
79 + return nil
80 +}
81 +
82 +func (d *datastore) Batch() (ds.Batch, error) {
83 + b, err := d.child.(ds.Batching).Batch()
84 + if err != nil {
85 + return nil, err
86 + }
87 +
88 + return &panicBatch{b}, nil
89 +}
90 +
91 type panicBatch struct {
92 t ds.Batch
93 }
Godeps/_workspace/src/github.com/jbenet/go-datastore/redis/redis.go
+10 -3
@@ -7,7 +7,6 @@ import (
7 "time"
8
9 "github.com/fzzy/radix/redis"
10 -
10 datastore "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
11 query "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
12 )
@@ -17,14 +16,14 @@ var _ datastore.ThreadSafeDatastore = &Datastore{}
16
17 var ErrInvalidType = errors.New("redis datastore: invalid type error. this datastore only supports []byte values")
18
20 -func NewExpiringDatastore(client *redis.Client, ttl time.Duration) (datastore.ThreadSafeDatastore, error) {
19 +func NewExpiringDatastore(client *redis.Client, ttl time.Duration) (*Datastore, error) {
20 return &Datastore{
21 client: client,
22 ttl: ttl,
23 }, nil
24 }
25
27 -func NewDatastore(client *redis.Client) (datastore.ThreadSafeDatastore, error) {
26 +func NewDatastore(client *redis.Client) (*Datastore, error) {
27 return &Datastore{
28 client: client,
29 }, nil
@@ -83,3 +82,11 @@ func (ds *Datastore) Query(q query.Query) (query.Results, error) {
82 }
83
84 func (ds *Datastore) IsThreadSafe() {}
85 +
86 +func (ds *Datastore) Batch() (datastore.Batch, error) {
87 + return nil, datastore.ErrBatchUnsupported
88 +}
89 +
90 +func (ds *Datastore) Close() error {
91 + return ds.client.Close()
92 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/redis/redis_test.go
+1
@@ -8,6 +8,7 @@ import (
8
9 "github.com/fzzy/radix/redis"
10 datastore "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
11 +
12 dstest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/test"
13 )
14
Godeps/_workspace/src/github.com/jbenet/go-datastore/sync/sync.go
+12 -2
@@ -1,6 +1,7 @@
1 package sync
2
3 import (
4 + "io"
5 "sync"
6
7 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
@@ -17,7 +18,7 @@ type MutexDatastore struct {
18
19 // MutexWrap constructs a datastore with a coarse lock around
20 // the entire datastore, for every single operation
20 -func MutexWrap(d ds.Datastore) ds.ThreadSafeDatastore {
21 +func MutexWrap(d ds.Datastore) *MutexDatastore {
22 return &MutexDatastore{child: d}
23 }
24
@@ -67,7 +68,7 @@ func (d *MutexDatastore) Query(q dsq.Query) (dsq.Results, error) {
68 func (d *MutexDatastore) Batch() (ds.Batch, error) {
69 d.RLock()
70 defer d.RUnlock()
70 - bds, ok := d.child.(ds.BatchingDatastore)
71 + bds, ok := d.child.(ds.Batching)
72 if !ok {
73 return nil, ds.ErrBatchUnsupported
74 }
@@ -81,6 +82,15 @@ func (d *MutexDatastore) Batch() (ds.Batch, error) {
82 }, nil
83 }
84
85 +func (d *MutexDatastore) Close() error {
86 + d.RWMutex.Lock()
87 + defer d.RWMutex.Unlock()
88 + if c, ok := d.child.(io.Closer); ok {
89 + return c.Close()
90 + }
91 + return nil
92 +}
93 +
94 type syncBatch struct {
95 lk sync.Mutex
96 batch ds.Batch
Godeps/_workspace/src/github.com/jbenet/go-datastore/syncmount/mount.go new
+198
@@ -0,0 +1,198 @@
1 +// Package mount provides a Datastore that has other Datastores
2 +// mounted at various key prefixes and is threadsafe
3 +package syncmount
4 +
5 +import (
6 + "errors"
7 + "io"
8 + "strings"
9 + "sync"
10 +
11 + ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
12 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/keytransform"
13 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
14 +)
15 +
16 +var (
17 + ErrNoMount = errors.New("no datastore mounted for this key")
18 +)
19 +
20 +type Mount struct {
21 + Prefix ds.Key
22 + Datastore ds.Datastore
23 +}
24 +
25 +func New(mounts []Mount) *Datastore {
26 + // make a copy so we're sure it doesn't mutate
27 + m := make([]Mount, len(mounts))
28 + for i, v := range mounts {
29 + m[i] = v
30 + }
31 + return &Datastore{mounts: m}
32 +}
33 +
34 +type Datastore struct {
35 + mounts []Mount
36 + lk sync.Mutex
37 +}
38 +
39 +var _ ds.Datastore = (*Datastore)(nil)
40 +
41 +func (d *Datastore) lookup(key ds.Key) (ds.Datastore, ds.Key, ds.Key) {
42 + d.lk.Lock()
43 + defer d.lk.Unlock()
44 + for _, m := range d.mounts {
45 + if m.Prefix.Equal(key) || m.Prefix.IsAncestorOf(key) {
46 + s := strings.TrimPrefix(key.String(), m.Prefix.String())
47 + k := ds.NewKey(s)
48 + return m.Datastore, m.Prefix, k
49 + }
50 + }
51 + return nil, ds.NewKey("/"), key
52 +}
53 +
54 +func (d *Datastore) Put(key ds.Key, value interface{}) error {
55 + cds, _, k := d.lookup(key)
56 + if cds == nil {
57 + return ErrNoMount
58 + }
59 + return cds.Put(k, value)
60 +}
61 +
62 +func (d *Datastore) Get(key ds.Key) (value interface{}, err error) {
63 + cds, _, k := d.lookup(key)
64 + if cds == nil {
65 + return nil, ds.ErrNotFound
66 + }
67 + return cds.Get(k)
68 +}
69 +
70 +func (d *Datastore) Has(key ds.Key) (exists bool, err error) {
71 + cds, _, k := d.lookup(key)
72 + if cds == nil {
73 + return false, nil
74 + }
75 + return cds.Has(k)
76 +}
77 +
78 +func (d *Datastore) Delete(key ds.Key) error {
79 + cds, _, k := d.lookup(key)
80 + if cds == nil {
81 + return ds.ErrNotFound
82 + }
83 + return cds.Delete(k)
84 +}
85 +
86 +func (d *Datastore) Query(q query.Query) (query.Results, error) {
87 + if len(q.Filters) > 0 ||
88 + len(q.Orders) > 0 ||
89 + q.Limit > 0 ||
90 + q.Offset > 0 {
91 + // TODO this is overly simplistic, but the only caller is
92 + // `ipfs refs local` for now, and this gets us moving.
93 + return nil, errors.New("mount only supports listing all prefixed keys in random order")
94 + }
95 + key := ds.NewKey(q.Prefix)
96 + cds, mount, k := d.lookup(key)
97 + if cds == nil {
98 + return nil, errors.New("mount only supports listing a mount point")
99 + }
100 + // TODO support listing cross mount points too
101 +
102 + // delegate the query to the mounted datastore, while adjusting
103 + // keys in and out
104 + q2 := q
105 + q2.Prefix = k.String()
106 + wrapDS := keytransform.Wrap(cds, &keytransform.Pair{
107 + Convert: func(ds.Key) ds.Key {
108 + panic("this should never be called")
109 + },
110 + Invert: func(k ds.Key) ds.Key {
111 + return mount.Child(k)
112 + },
113 + })
114 +
115 + r, err := wrapDS.Query(q2)
116 + if err != nil {
117 + return nil, err
118 + }
119 + r = query.ResultsReplaceQuery(r, q)
120 + return r, nil
121 +}
122 +
123 +func (d *Datastore) IsThreadSafe() {}
124 +
125 +func (d *Datastore) Close() error {
126 + for _, d := range d.mounts {
127 + if c, ok := d.Datastore.(io.Closer); ok {
128 + err := c.Close()
129 + if err != nil {
130 + return err
131 + }
132 + }
133 + }
134 + return nil
135 +}
136 +
137 +type mountBatch struct {
138 + mounts map[string]ds.Batch
139 + lk sync.Mutex
140 +
141 + d *Datastore
142 +}
143 +
144 +func (d *Datastore) Batch() (ds.Batch, error) {
145 + return &mountBatch{
146 + mounts: make(map[string]ds.Batch),
147 + d: d,
148 + }, nil
149 +}
150 +
151 +func (mt *mountBatch) lookupBatch(key ds.Key) (ds.Batch, ds.Key, error) {
152 + mt.lk.Lock()
153 + defer mt.lk.Unlock()
154 +
155 + child, loc, rest := mt.d.lookup(key)
156 + t, ok := mt.mounts[loc.String()]
157 + if !ok {
158 + bds, ok := child.(ds.Batching)
159 + if !ok {
160 + return nil, ds.NewKey(""), ds.ErrBatchUnsupported
161 + }
162 + var err error
163 + t, err = bds.Batch()
164 + if err != nil {
165 + return nil, ds.NewKey(""), err
166 + }
167 + mt.mounts[loc.String()] = t
168 + }
169 + return t, rest, nil
170 +}
171 +
172 +func (mt *mountBatch) Put(key ds.Key, val interface{}) error {
173 + t, rest, err := mt.lookupBatch(key)
174 + if err != nil {
175 + return err
176 + }
177 +
178 + return t.Put(rest, val)
179 +}
180 +
181 +func (mt *mountBatch) Delete(key ds.Key) error {
182 + t, rest, err := mt.lookupBatch(key)
183 + if err != nil {
184 + return err
185 + }
186 +
187 + return t.Delete(rest)
188 +}
189 +
190 +func (mt *mountBatch) Commit() error {
191 + for _, t := range mt.mounts {
192 + err := t.Commit()
193 + if err != nil {
194 + return err
195 + }
196 + }
197 + return nil
198 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/syncmount/mount_test.go new
+241
@@ -0,0 +1,241 @@
1 +package syncmount_test
2 +
3 +import (
4 + "testing"
5 +
6 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
7 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/mount"
8 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
9 +)
10 +
11 +func TestPutBadNothing(t *testing.T) {
12 + m := mount.New(nil)
13 +
14 + err := m.Put(datastore.NewKey("quux"), []byte("foobar"))
15 + if g, e := err, mount.ErrNoMount; g != e {
16 + t.Fatalf("Put got wrong error: %v != %v", g, e)
17 + }
18 +}
19 +
20 +func TestPutBadNoMount(t *testing.T) {
21 + mapds := datastore.NewMapDatastore()
22 + m := mount.New([]mount.Mount{
23 + {Prefix: datastore.NewKey("/redherring"), Datastore: mapds},
24 + })
25 +
26 + err := m.Put(datastore.NewKey("/quux/thud"), []byte("foobar"))
27 + if g, e := err, mount.ErrNoMount; g != e {
28 + t.Fatalf("expected ErrNoMount, got: %v\n", g)
29 + }
30 +}
31 +
32 +func TestPut(t *testing.T) {
33 + mapds := datastore.NewMapDatastore()
34 + m := mount.New([]mount.Mount{
35 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
36 + })
37 +
38 + if err := m.Put(datastore.NewKey("/quux/thud"), []byte("foobar")); err != nil {
39 + t.Fatalf("Put error: %v", err)
40 + }
41 +
42 + val, err := mapds.Get(datastore.NewKey("/thud"))
43 + if err != nil {
44 + t.Fatalf("Get error: %v", err)
45 + }
46 + buf, ok := val.([]byte)
47 + if !ok {
48 + t.Fatalf("Get value is not []byte: %T %v", val, val)
49 + }
50 + if g, e := string(buf), "foobar"; g != e {
51 + t.Errorf("wrong value: %q != %q", g, e)
52 + }
53 +}
54 +
55 +func TestGetBadNothing(t *testing.T) {
56 + m := mount.New([]mount.Mount{})
57 +
58 + _, err := m.Get(datastore.NewKey("/quux/thud"))
59 + if g, e := err, datastore.ErrNotFound; g != e {
60 + t.Fatalf("expected ErrNotFound, got: %v\n", g)
61 + }
62 +}
63 +
64 +func TestGetBadNoMount(t *testing.T) {
65 + mapds := datastore.NewMapDatastore()
66 + m := mount.New([]mount.Mount{
67 + {Prefix: datastore.NewKey("/redherring"), Datastore: mapds},
68 + })
69 +
70 + _, err := m.Get(datastore.NewKey("/quux/thud"))
71 + if g, e := err, datastore.ErrNotFound; g != e {
72 + t.Fatalf("expected ErrNotFound, got: %v\n", g)
73 + }
74 +}
75 +
76 +func TestGetNotFound(t *testing.T) {
77 + mapds := datastore.NewMapDatastore()
78 + m := mount.New([]mount.Mount{
79 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
80 + })
81 +
82 + _, err := m.Get(datastore.NewKey("/quux/thud"))
83 + if g, e := err, datastore.ErrNotFound; g != e {
84 + t.Fatalf("expected ErrNotFound, got: %v\n", g)
85 + }
86 +}
87 +
88 +func TestGet(t *testing.T) {
89 + mapds := datastore.NewMapDatastore()
90 + m := mount.New([]mount.Mount{
91 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
92 + })
93 +
94 + if err := mapds.Put(datastore.NewKey("/thud"), []byte("foobar")); err != nil {
95 + t.Fatalf("Get error: %v", err)
96 + }
97 +
98 + val, err := m.Get(datastore.NewKey("/quux/thud"))
99 + if err != nil {
100 + t.Fatalf("Put error: %v", err)
101 + }
102 +
103 + buf, ok := val.([]byte)
104 + if !ok {
105 + t.Fatalf("Get value is not []byte: %T %v", val, val)
106 + }
107 + if g, e := string(buf), "foobar"; g != e {
108 + t.Errorf("wrong value: %q != %q", g, e)
109 + }
110 +}
111 +
112 +func TestHasBadNothing(t *testing.T) {
113 + m := mount.New([]mount.Mount{})
114 +
115 + found, err := m.Has(datastore.NewKey("/quux/thud"))
116 + if err != nil {
117 + t.Fatalf("Has error: %v", err)
118 + }
119 + if g, e := found, false; g != e {
120 + t.Fatalf("wrong value: %v != %v", g, e)
121 + }
122 +}
123 +
124 +func TestHasBadNoMount(t *testing.T) {
125 + mapds := datastore.NewMapDatastore()
126 + m := mount.New([]mount.Mount{
127 + {Prefix: datastore.NewKey("/redherring"), Datastore: mapds},
128 + })
129 +
130 + found, err := m.Has(datastore.NewKey("/quux/thud"))
131 + if err != nil {
132 + t.Fatalf("Has error: %v", err)
133 + }
134 + if g, e := found, false; g != e {
135 + t.Fatalf("wrong value: %v != %v", g, e)
136 + }
137 +}
138 +
139 +func TestHasNotFound(t *testing.T) {
140 + mapds := datastore.NewMapDatastore()
141 + m := mount.New([]mount.Mount{
142 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
143 + })
144 +
145 + found, err := m.Has(datastore.NewKey("/quux/thud"))
146 + if err != nil {
147 + t.Fatalf("Has error: %v", err)
148 + }
149 + if g, e := found, false; g != e {
150 + t.Fatalf("wrong value: %v != %v", g, e)
151 + }
152 +}
153 +
154 +func TestHas(t *testing.T) {
155 + mapds := datastore.NewMapDatastore()
156 + m := mount.New([]mount.Mount{
157 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
158 + })
159 +
160 + if err := mapds.Put(datastore.NewKey("/thud"), []byte("foobar")); err != nil {
161 + t.Fatalf("Put error: %v", err)
162 + }
163 +
164 + found, err := m.Has(datastore.NewKey("/quux/thud"))
165 + if err != nil {
166 + t.Fatalf("Has error: %v", err)
167 + }
168 + if g, e := found, true; g != e {
169 + t.Fatalf("wrong value: %v != %v", g, e)
170 + }
171 +}
172 +
173 +func TestDeleteNotFound(t *testing.T) {
174 + mapds := datastore.NewMapDatastore()
175 + m := mount.New([]mount.Mount{
176 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
177 + })
178 +
179 + err := m.Delete(datastore.NewKey("/quux/thud"))
180 + if g, e := err, datastore.ErrNotFound; g != e {
181 + t.Fatalf("expected ErrNotFound, got: %v\n", g)
182 + }
183 +}
184 +
185 +func TestDelete(t *testing.T) {
186 + mapds := datastore.NewMapDatastore()
187 + m := mount.New([]mount.Mount{
188 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
189 + })
190 +
191 + if err := mapds.Put(datastore.NewKey("/thud"), []byte("foobar")); err != nil {
192 + t.Fatalf("Put error: %v", err)
193 + }
194 +
195 + err := m.Delete(datastore.NewKey("/quux/thud"))
196 + if err != nil {
197 + t.Fatalf("Delete error: %v", err)
198 + }
199 +
200 + // make sure it disappeared
201 + found, err := mapds.Has(datastore.NewKey("/thud"))
202 + if err != nil {
203 + t.Fatalf("Has error: %v", err)
204 + }
205 + if g, e := found, false; g != e {
206 + t.Fatalf("wrong value: %v != %v", g, e)
207 + }
208 +}
209 +
210 +func TestQuerySimple(t *testing.T) {
211 + mapds := datastore.NewMapDatastore()
212 + m := mount.New([]mount.Mount{
213 + {Prefix: datastore.NewKey("/quux"), Datastore: mapds},
214 + })
215 +
216 + const myKey = "/quux/thud"
217 + if err := m.Put(datastore.NewKey(myKey), []byte("foobar")); err != nil {
218 + t.Fatalf("Put error: %v", err)
219 + }
220 +
221 + res, err := m.Query(query.Query{Prefix: "/quux"})
222 + if err != nil {
223 + t.Fatalf("Query fail: %v\n", err)
224 + }
225 + entries, err := res.Rest()
226 + if err != nil {
227 + t.Fatalf("Query Results.Rest fail: %v\n", err)
228 + }
229 + seen := false
230 + for _, e := range entries {
231 + switch e.Key {
232 + case datastore.NewKey(myKey).String():
233 + seen = true
234 + default:
235 + t.Errorf("saw unexpected key: %q", e.Key)
236 + }
237 + }
238 + if !seen {
239 + t.Errorf("did not see wanted key %q in %+v", myKey, entries)
240 + }
241 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/test/test_util.go
+2 -2
@@ -9,7 +9,7 @@ import (
9 dstore "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
10 )
11
12 -func RunBatchTest(t *testing.T, ds dstore.BatchingDatastore) {
12 +func RunBatchTest(t *testing.T, ds dstore.Batching) {
13 batch, err := ds.Batch()
14 if err != nil {
15 t.Fatal(err)
@@ -58,7 +58,7 @@ func RunBatchTest(t *testing.T, ds dstore.BatchingDatastore) {
58 }
59 }
60
61 -func RunBatchDeleteTest(t *testing.T, ds dstore.BatchingDatastore) {
61 +func RunBatchDeleteTest(t *testing.T, ds dstore.Batching) {
62 r := rand.New()
63 var keys []dstore.Key
64 for i := 0; i < 20; i++ {
Godeps/_workspace/src/github.com/jbenet/go-datastore/tiered/tiered.go
+1 -1
@@ -13,7 +13,7 @@ type tiered []ds.Datastore
13 // New returns a tiered datastore. Puts and Deletes will write-through to
14 // all datastores, Has and Get will try each datastore sequentially, and
15 // Query will always try the last one (most complete) first.
16 -func New(dses ...ds.Datastore) ds.Datastore {
16 +func New(dses ...ds.Datastore) tiered {
17 return tiered(dses)
18 }
19
Godeps/_workspace/src/github.com/jbenet/go-datastore/tiered/tiered_test.go
+4 -4
@@ -49,19 +49,19 @@ func TestTiered(t *testing.T) {
49 td := New(d1, d2, d3, d4)
50 td.Put(ds.NewKey("foo"), "bar")
51 testHas(t, []ds.Datastore{td}, ds.NewKey("foo"), "bar")
52 - testHas(t, td.(tiered), ds.NewKey("foo"), "bar") // all children
52 + testHas(t, td, ds.NewKey("foo"), "bar") // all children
53
54 // remove it from, say, caches.
55 d1.Delete(ds.NewKey("foo"))
56 d2.Delete(ds.NewKey("foo"))
57 testHas(t, []ds.Datastore{td}, ds.NewKey("foo"), "bar")
58 - testHas(t, td.(tiered)[2:], ds.NewKey("foo"), "bar")
59 - testNotHas(t, td.(tiered)[:2], ds.NewKey("foo"))
58 + testHas(t, td[2:], ds.NewKey("foo"), "bar")
59 + testNotHas(t, td[:2], ds.NewKey("foo"))
60
61 // write it again.
62 td.Put(ds.NewKey("foo"), "bar2")
63 testHas(t, []ds.Datastore{td}, ds.NewKey("foo"), "bar2")
64 - testHas(t, td.(tiered), ds.NewKey("foo"), "bar2")
64 + testHas(t, td, ds.NewKey("foo"), "bar2")
65 }
66
67 func TestQueryCallsLast(t *testing.T) {
Godeps/_workspace/src/github.com/jbenet/go-datastore/timecache/timecache.go
+10 -2
@@ -1,6 +1,7 @@
1 package timecache
2
3 import (
4 + "io"
5 "sync"
6 "time"
7
@@ -24,13 +25,13 @@ type datastore struct {
25 ttls map[ds.Key]time.Time
26 }
27
27 -func WithTTL(ttl time.Duration) ds.Datastore {
28 +func WithTTL(ttl time.Duration) *datastore {
29 return WithCache(ds.NewMapDatastore(), ttl)
30 }
31
32 // WithCache wraps a given datastore as a timecache.
33 // Get + Has requests are considered expired after a TTL.
33 -func WithCache(d ds.Datastore, ttl time.Duration) ds.Datastore {
34 +func WithCache(d ds.Datastore, ttl time.Duration) *datastore {
35 return &datastore{cache: d, ttl: ttl, ttls: make(map[ds.Key]time.Time)}
36 }
37
@@ -94,3 +95,10 @@ func (d *datastore) Delete(key ds.Key) (err error) {
95 func (d *datastore) Query(q dsq.Query) (dsq.Results, error) {
96 return d.cache.Query(q)
97 }
98 +
99 +func (d *datastore) Close() error {
100 + if c, ok := d.cache.(io.Closer); ok {
101 + return c.Close()
102 + }
103 + return nil
104 +}
blocks/blockstore/blockstore.go
+1 -1
@@ -43,7 +43,7 @@ func NewBlockstore(d ds.ThreadSafeDatastore) Blockstore {
43 }
44
45 type blockstore struct {
46 - datastore ds.BatchingDatastore
46 + datastore ds.Batching
47 // cant be ThreadSafeDatastore cause namespace.Datastore doesnt support it.
48 // we do check it on `NewBlockstore` though.
49 }
repo/fsrepo/fsrepo.go
+7 -17
@@ -94,10 +94,6 @@ type FSRepo struct {
94 lockfile io.Closer
95 config *config.Config
96 ds ds.ThreadSafeDatastore
97 - // tracked separately for use in Close; do not use directly.
98 - leveldbDS levelds.Datastore
99 - metricsBlocks measure.DatastoreCloser
100 - metricsLevelDB measure.DatastoreCloser
97 }
98
99 var _ repo.Repo = (*FSRepo)(nil)
@@ -352,7 +348,7 @@ func (r *FSRepo) openDatastore() error {
348 leveldbPath := path.Join(r.path, leveldbDirectory)
349 var err error
350 // save leveldb reference so it can be neatly closed afterward
355 - r.leveldbDS, err = levelds.NewDatastore(leveldbPath, &levelds.Options{
351 + leveldbDS, err := levelds.NewDatastore(leveldbPath, &levelds.Options{
352 Compression: ldbopts.NoCompression,
353 })
354 if err != nil {
@@ -382,16 +378,16 @@ func (r *FSRepo) openDatastore() error {
378 id = fmt.Sprintf("uninitialized_%p", r)
379 }
380 prefix := "fsrepo." + id + ".datastore."
385 - r.metricsBlocks = measure.New(prefix+"blocks", blocksDS)
386 - r.metricsLevelDB = measure.New(prefix+"leveldb", r.leveldbDS)
381 + metricsBlocks := measure.New(prefix+"blocks", blocksDS)
382 + metricsLevelDB := measure.New(prefix+"leveldb", leveldbDS)
383 mountDS := mount.New([]mount.Mount{
384 {
385 Prefix: ds.NewKey("/blocks"),
390 - Datastore: r.metricsBlocks,
386 + Datastore: metricsBlocks,
387 },
388 {
389 Prefix: ds.NewKey("/"),
394 - Datastore: r.metricsLevelDB,
390 + Datastore: metricsLevelDB,
391 },
392 })
393 // Make sure it's ok to claim the virtual datastore from mount as
@@ -400,7 +396,7 @@ func (r *FSRepo) openDatastore() error {
396 // variants. This is the same dilemma as the `[].byte` attempt at
397 // introducing const types to Go.
398 var _ ds.ThreadSafeDatastore = blocksDS
403 - var _ ds.ThreadSafeDatastore = r.leveldbDS
399 + var _ ds.ThreadSafeDatastore = leveldbDS
400 r.ds = ds2.ClaimThreadSafe{mountDS}
401 return nil
402 }
@@ -420,13 +416,7 @@ func (r *FSRepo) Close() error {
416 return errors.New("repo is closed")
417 }
418
423 - if err := r.metricsBlocks.Close(); err != nil {
424 - return err
425 - }
426 - if err := r.metricsLevelDB.Close(); err != nil {
427 - return err
428 - }
429 - if err := r.leveldbDS.Close(); err != nil {
419 + if err := r.ds.(io.Closer).Close(); err != nil {
420 return err
421 }
422
util/datastore2/datastore_closer.go
+1 -1
@@ -26,7 +26,7 @@ func (w *datastoreCloserWrapper) Close() error {
26 }
27
28 func (w *datastoreCloserWrapper) Batch() (datastore.Batch, error) {
29 - bds, ok := w.ThreadSafeDatastore.(datastore.BatchingDatastore)
29 + bds, ok := w.ThreadSafeDatastore.(datastore.Batching)
30 if !ok {
31 return nil, datastore.ErrBatchUnsupported
32 }
util/datastore2/threadsafe.go
+8 -1
@@ -1,15 +1,22 @@
1 package datastore2
2
3 import (
4 + "io"
5 +
6 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
7 )
8
9 // ClaimThreadSafe claims that a Datastore is threadsafe, even when
10 // it's type does not guarantee this. Use carefully.
11 type ClaimThreadSafe struct {
10 - datastore.BatchingDatastore
12 + datastore.Batching
13 }
14
15 var _ datastore.ThreadSafeDatastore = ClaimThreadSafe{}
16
17 func (ClaimThreadSafe) IsThreadSafe() {}
18 +
19 +// TEMP UNTIL dev0.4.0 merges and solves this ugly interface stuff
20 +func (c ClaimThreadSafe) Close() error {
21 + return c.Batching.(io.Closer).Close()
22 +}