x

core/grpc_bucket.go

@@ -2,7 +2,11 @@ package core
 
 import (
 	"context"
+	"errors"
+	"sync"
+	"time"
 
+	"github.com/cdle/sillyplus/core/storage"
 	"github.com/cdle/sillyplus/proto3/srpc"
 	"github.com/cdle/sillyplus/utils"
 )
@@ -11,6 +15,57 @@ type SillyGirlService struct {
 	srpc.UnsafeSillyGirlServiceServer
 }
 
+func (sg *SillyGirlService) BucketWatch(stream srpc.SillyGirlService_BucketWatchServer) error {
+	var watcher func(old, new, key string) *storage.Final
+	var echos sync.Map
+	for {
+		req, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		if watcher == nil {
+			watcher = func(old, new, key string) *storage.Final {
+				echo := utils.GenUUID()
+				ch := make(chan *storage.Final)
+				echos.Store(echo, ch)
+				defer echos.Delete(echo)
+				stream.Send(&srpc.BucketWatchResponse{
+					Echo: echo,
+					Old:  old,
+					Now:  new,
+					Key:  key,
+				})
+				select {
+				case v := <-ch:
+					return v
+				case <-time.After(time.Second * 5):
+				}
+				return nil
+			}
+			storage.Watch(MakeBucket(req.Name), req.Key, watcher, req.PluginId)
+		} else {
+			echo := req.Echo
+			v, ok := echos.Load(echo)
+			var fn *storage.Final
+			if req.Error != "VOID" {
+				fn = &storage.Final{
+					Now:     req.Now,
+					Message: req.Message,
+				}
+				if req.Error != "" {
+					fn.Error = errors.New(req.Error)
+				}
+			}
+			if ok {
+				select {
+				case v.(chan *storage.Final) <- fn:
+				case <-time.After(time.Millisecond):
+				}
+			}
+		}
+	}
+}
+
 // Get implements BucketServiceServer.Get.
 func (sg *SillyGirlService) BucketGet(ctx context.Context, req *srpc.BucketKeyRequest) (*srpc.Default, error) {
 	value := MakeBucket(req.Name).GetString(req.Key)