feat(listx): add storage listx option

2021-10-23 16:17:29 +08:00 · 2021-10-23 16:17:29 +08:00 · 0b781a178e
parent e96f21399c
commit 0b781a178e
1 changed files with 235 additions and 26 deletions
--- a/storage/server/single/listx.go
+++ b/storage/server/single/listx.go
@ -1,52 +1,261 @@
 package single

-import "gitee.com/timedb/wheatCache/pkg/proto"
+import (
+	context "context"

-func (s *serverSingle) Lindex(ctx context.Context, request *proto.LindexRequest) (*proto.LindexResponse, error) {
-	panic("implement me")
+	"gitee.com/timedb/wheatCache/pkg/event"
+	"gitee.com/timedb/wheatCache/pkg/lru"
+	"gitee.com/timedb/wheatCache/pkg/proto"
+)
+
+func (s *serverSingle) Lindex(
+	ctx context.Context,
+	req *proto.LindexRequest,
+) (*proto.LindexResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return s.dao.LINdex(req.Key, req.Index)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LindexResponse{
+		Result: resp.(string),
+	}, nil
 }

-func (s *serverSingle) Llen(ctx context.Context, request *proto.LlenRequest) (*proto.LlenResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Llen(
+	ctx context.Context,
+	req *proto.LlenRequest,
+) (*proto.LlenResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return s.dao.LLen(req.Key)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LlenResponse{
+		Length: resp.(int32),
+	}, nil
 }

-func (s *serverSingle) Lpop(ctx context.Context, request *proto.LpopRequest) (*proto.LpopResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Lpop(
+	ctx context.Context,
+	request *proto.LpopRequest,
+) (*proto.LpopResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return s.dao.LPop(request.Key, request.Count)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LpopResponse{
+		Results: resp.([]string),
+	}, nil
 }

-func (s *serverSingle) Lpush(ctx context.Context, request *proto.LpushRequest) (*proto.LpushResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Lpush(
+	ctx context.Context,
+	req *proto.LpushRequest,
+) (*proto.LpushResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return nil, s.dao.LPush(req.Key, req.Values...)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	_, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LpushResponse{}, nil
 }

-func (s *serverSingle) Lpushx(ctx context.Context, request *proto.LpushxRequest) (*proto.LpushxResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Lpushx(
+	ctx context.Context,
+	req *proto.LpushxRequest,
+) (*proto.LpushxResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return nil, s.dao.LPush(req.Key, req.Values...)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	_, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LpushxResponse{}, nil
 }

-func (s *serverSingle) Lrange(ctx context.Context, request *proto.LrangeRequest) (*proto.LrangeResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Lrange(
+	ctx context.Context,
+	req *proto.LrangeRequest,
+) (*proto.LrangeResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return s.dao.LRange(req.Key, req.Start, req.End)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LrangeResponse{
+		Values: resp.([]string),
+	}, nil
 }

-func (s *serverSingle) Lrem(ctx context.Context, request *proto.LremRequest) (*proto.LremResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Lrem(
+	ctx context.Context,
+	req *proto.LremRequest,
+) (*proto.LremResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return s.dao.LRemove(req.Key, req.Count, req.Value)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LremResponse{
+		Count: resp.(int32),
+	}, nil
 }

-func (s *serverSingle) Lset(ctx context.Context, request *proto.LsetRequest) (*proto.LsetResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Lset(
+	ctx context.Context,
+	req *proto.LsetRequest,
+) (*proto.LsetResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return nil, s.dao.LSet(req.Key, req.Index, req.Value)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	_, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LsetResponse{}, nil
 }

-func (s *serverSingle) Rpop(ctx context.Context, request *proto.RpopRequest) (*proto.RpopResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Rpop(
+	ctx context.Context,
+	req *proto.RpopRequest,
+) (*proto.RpopResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return s.dao.RPop(req.Key, req.Count)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.RpopResponse{
+		Result: resp.([]string),
+	}, nil
 }

-func (s *serverSingle) Ltrim(ctx context.Context, request *proto.LtrimRequest) (*proto.LtrimResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Ltrim(
+	ctx context.Context,
+	req *proto.LtrimRequest,
+) (*proto.LtrimResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return nil, s.dao.LTrim(req.Key, req.Start, req.End)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	_, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.LtrimResponse{}, nil
 }

-func (s *serverSingle) Rpush(ctx context.Context, request *proto.RpushRequest) (*proto.RpushResponse, error) {
-	panic("implement me")
+func (s *serverSingle) Rpush(
+	ctx context.Context,
+	req *proto.RpushRequest,
+) (*proto.RpushResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return nil, s.dao.RPush(req.Key, req.Values...)
+	})
+
+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	_, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.RpushResponse{}, nil
 }

-func (s *serverSingle) Rpushx(ctx context.Context, request *proto.RpushxRequest) (*proto.RpushxResponse, error) {
-	panic("implement me")
-}
+func (s *serverSingle) Rpushx(
+	ctx context.Context,
+	req *proto.RpushxRequest,
+) (*proto.RpushxResponse, error) {
+	work := event.EventWorkFunc(func() (interface{}, error) {
+		return nil, s.dao.RPushX(req.Key, req.Values...)
+	})

+	lruEvent := event.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(lru.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	_, err := lruEvent.StartWaitEvent(s.timeOut)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.RpushxResponse{}, nil
+}