store/noop.go

@@ -2,14 +2,18 @@ package store
 
 import (
 	"context"
+	"sync"
 	"sync/atomic"
 
 	"go.unistack.org/micro/v3/options"
+	"go.unistack.org/micro/v3/util/id"
 )
 
 var _ Store = (*noopStore)(nil)
 
 type noopStore struct {
+	mu          sync.Mutex
+	watchers    map[string]Watcher
 	funcRead    FuncRead
 	funcWrite   FuncWrite
 	funcExists  FuncExists
@@ -182,3 +186,41 @@ func (n *noopStore) connect(ctx context.Context) error {
 
 	return nil
 }
+
+type watcher struct {
+	exit chan bool
+	id   string
+	ch   chan Event
+	opts WatchOptions
+}
+
+func (m *noopStore) Watch(ctx context.Context, opts ...WatchOption) (Watcher, error) {
+	id, err := id.New()
+	if err != nil {
+		return nil, err
+	}
+	wo, err := NewWatchOptions(opts...)
+	if err != nil {
+		return nil, err
+	}
+	// construct the watcher
+	w := &watcher{
+		exit: make(chan bool),
+		ch:   make(chan Event),
+		id:   id,
+		opts: wo,
+	}
+
+	m.mu.Lock()
+	m.watchers[w.id] = w
+	m.mu.Unlock()
+
+	return w, nil
+}
+
+func (w *watcher) Next() (Event, error) {
+	return nil, nil
+}
+
+func (w *watcher) Stop() {
+}

store/store.go

@@ -4,9 +4,11 @@ package store
 import (
 	"context"
 	"errors"
+	"time"
 )
 
 var (
+	ErrWatcherStopped = errors.New("watcher stopped")
 	// ErrNotConnected is returned when a store is not connected
 	ErrNotConnected = errors.New("not conected")
 	// ErrNotFound is returned when a key doesn't exist
@@ -43,6 +45,7 @@ type Store interface {
 	Disconnect(ctx context.Context) error
 	// String returns the name of the implementation.
 	String() string
+	Watch(ctx context.Context, opts ...WatchOption) (Watcher, error)
 }
 
 type (
@@ -57,3 +60,41 @@ type (
 	FuncList   func(ctx context.Context, opts ...ListOption) ([]string, error)
 	HookList   func(next FuncList) FuncList
 )
+
+type EventType int
+
+const (
+	EventTypeUnknown = iota
+	EventTypeConnect
+	EventTypeDisconnect
+	EventTypeOpError
+)
+
+type Event interface {
+	Timestamp() time.Time
+	Error() error
+	Type() EventType
+}
+
+type Watcher interface {
+	// Next is a blocking call
+	Next() (Event, error)
+	// Stop stops the watcher
+	Stop()
+}
+
+type WatchOption func(*WatchOptions) error
+
+type WatchOptions struct{}
+
+func NewWatchOptions(opts ...WatchOption) (WatchOptions, error) {
+	options := WatchOptions{}
+	var err error
+	for _, o := range opts {
+		if err = o(&options); err != nil {
+			break
+		}
+	}
+
+	return options, err
+}

store/wrapper.go

@@ -67,16 +67,6 @@ func (w *NamespaceStore) String() string {
 	return w.s.String()
 }
 
-// type NamespaceWrapper struct{}
-
-// func NewNamespaceWrapper() Wrapper {
-//	return &NamespaceWrapper{}
-// }
-
-/*
-func (w *OmitWrapper) Logf(fn LogfFunc) LogfFunc {
-	return func(ctx context.Context, level Level, msg string, args ...interface{}) {
-		fn(ctx, level, msg, getArgs(args)...)
-	}
+func (w *NamespaceStore) Watch(ctx context.Context, opts ...WatchOption) (Watcher, error) {
+	return w.s.Watch(ctx, opts...)
 }
-*/