etcd3/store: watcher implementation

pkg/storage/etcd3/event.go

@@ -0,0 +1,50 @@
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+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 etcd3
+
+import (
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/storage/storagepb"
+)
+
+type event struct {
+	key       string
+	value     []byte
+	rev       int64
+	isDeleted bool
+	isCreated bool
+}
+
+func parseKV(kv *storagepb.KeyValue) *event {
+	return &event{
+		key:       string(kv.Key),
+		value:     kv.Value,
+		rev:       kv.ModRevision,
+		isDeleted: false,
+		isCreated: kv.ModRevision == kv.CreateRevision,
+	}
+}
+
+func parseEvent(e *clientv3.Event) *event {
+	return &event{
+		key:       string(e.Kv.Key),
+		value:     e.Kv.Value,
+		rev:       e.Kv.ModRevision,
+		isDeleted: e.Type == clientv3.EventTypeDelete,
+		isCreated: e.IsCreate(),
+	}
+}

pkg/storage/etcd3/store.go

@@ -42,6 +42,7 @@ type store struct {
 	codec      runtime.Codec
 	versioner  storage.Versioner
 	pathPrefix string
+	watcher    *watcher
 }
 
 type elemForDecode struct {
@@ -57,11 +58,13 @@ type objState struct {
 }
 
 func newStore(c *clientv3.Client, codec runtime.Codec, prefix string) *store {
+	versioner := etcd.APIObjectVersioner{}
 	return &store{
 		client:     c,
-		versioner:  etcd.APIObjectVersioner{},
+		versioner:  versioner,
 		codec:      codec,
 		pathPrefix: prefix,
+		watcher:    newWatcher(c, codec, versioner),
 	}
 }
 
@@ -315,12 +318,21 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, filter st
 
 // Watch implements storage.Interface.Watch.
 func (s *store) Watch(ctx context.Context, key string, resourceVersion string, filter storage.FilterFunc) (watch.Interface, error) {
-	panic("TODO: unimplemented")
+	return s.watch(ctx, key, resourceVersion, filter, false)
 }
 
 // WatchList implements storage.Interface.WatchList.
 func (s *store) WatchList(ctx context.Context, key string, resourceVersion string, filter storage.FilterFunc) (watch.Interface, error) {
-	panic("TODO: unimplemented")
+	return s.watch(ctx, key, resourceVersion, filter, true)
+}
+
+func (s *store) watch(ctx context.Context, key string, rv string, filter storage.FilterFunc, recursive bool) (watch.Interface, error) {
+	rev, err := storage.ParseWatchResourceVersion(rv)
+	if err != nil {
+		return nil, err
+	}
+	key = keyWithPrefix(s.pathPrefix, key)
+	return s.watcher.Watch(ctx, key, int64(rev), recursive, filter)
 }
 
 func (s *store) getState(getResp *clientv3.GetResponse, key string, v reflect.Value, ignoreNotFound bool) (*objState, error) {