!101 feat-channex

Merge pull request !101 from bandl/feat-storage-channelx
2021-11-30 10:55:07 +00:00 · 2021-11-30 10:55:07 +00:00 · c6b14ced27
parent a37097b3dd b87799a88e
commit c6b14ced27
18 changed files with 1775 additions and 96 deletions
--- a/README.md
+++ b/README.md
@ -95,9 +95,9 @@ func TestClient(t *testing.T) {
 	- [x] listx
 	- [x] stringx 
 	- [x] hashx
-	- [ ] setx
+	- [x] setx
 	- [ ] zsetx
-	- [ ] channelx
+	- [x] channelx
 - storage 层
 	- [x] log 消息推送
 	- [x] 服务自动生成工具
--- a/mock/storage/mock_client.gen.go
+++ b/mock/storage/mock_client.gen.go
@ -56,6 +56,106 @@ func (mr *MockCommServerClientMockRecorder) Add(ctx, in interface{}, opts ...int
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Add", reflect.TypeOf((*MockCommServerClient)(nil).Add), varargs...)
 }

+// CClean mocks base method.
+func (m *MockCommServerClient) CClean(ctx context.Context, in *proto.CCleanRequest, opts ...grpc.CallOption) (*proto.CCleanResponse, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, in}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "CClean", varargs...)
+	ret0, _ := ret[0].(*proto.CCleanResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CClean indicates an expected call of CClean.
+func (mr *MockCommServerClientMockRecorder) CClean(ctx, in interface{}, opts ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, in}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CClean", reflect.TypeOf((*MockCommServerClient)(nil).CClean), varargs...)
+}
+
+// CLen mocks base method.
+func (m *MockCommServerClient) CLen(ctx context.Context, in *proto.CLenRequest, opts ...grpc.CallOption) (*proto.CLenResponse, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, in}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "CLen", varargs...)
+	ret0, _ := ret[0].(*proto.CLenResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CLen indicates an expected call of CLen.
+func (mr *MockCommServerClientMockRecorder) CLen(ctx, in interface{}, opts ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, in}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CLen", reflect.TypeOf((*MockCommServerClient)(nil).CLen), varargs...)
+}
+
+// CMake mocks base method.
+func (m *MockCommServerClient) CMake(ctx context.Context, in *proto.CMakeRequest, opts ...grpc.CallOption) (*proto.CMakeResponse, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, in}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "CMake", varargs...)
+	ret0, _ := ret[0].(*proto.CMakeResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CMake indicates an expected call of CMake.
+func (mr *MockCommServerClientMockRecorder) CMake(ctx, in interface{}, opts ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, in}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CMake", reflect.TypeOf((*MockCommServerClient)(nil).CMake), varargs...)
+}
+
+// CPop mocks base method.
+func (m *MockCommServerClient) CPop(ctx context.Context, in *proto.CPopRequest, opts ...grpc.CallOption) (*proto.CPopResponse, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, in}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "CPop", varargs...)
+	ret0, _ := ret[0].(*proto.CPopResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CPop indicates an expected call of CPop.
+func (mr *MockCommServerClientMockRecorder) CPop(ctx, in interface{}, opts ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, in}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CPop", reflect.TypeOf((*MockCommServerClient)(nil).CPop), varargs...)
+}
+
+// CPush mocks base method.
+func (m *MockCommServerClient) CPush(ctx context.Context, in *proto.CPushRequest, opts ...grpc.CallOption) (*proto.CPushResponse, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, in}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "CPush", varargs...)
+	ret0, _ := ret[0].(*proto.CPushResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CPush indicates an expected call of CPush.
+func (mr *MockCommServerClientMockRecorder) CPush(ctx, in interface{}, opts ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, in}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CPush", reflect.TypeOf((*MockCommServerClient)(nil).CPush), varargs...)
+}
+
 // Get mocks base method.
 func (m *MockCommServerClient) Get(ctx context.Context, in *proto.GetRequest, opts ...grpc.CallOption) (*proto.GetResponse, error) {
 	m.ctrl.T.Helper()
@ -954,6 +1054,81 @@ func (mr *MockCommServerServerMockRecorder) Add(arg0, arg1 interface{}) *gomock.
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Add", reflect.TypeOf((*MockCommServerServer)(nil).Add), arg0, arg1)
 }

+// CClean mocks base method.
+func (m *MockCommServerServer) CClean(arg0 context.Context, arg1 *proto.CCleanRequest) (*proto.CCleanResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CClean", arg0, arg1)
+	ret0, _ := ret[0].(*proto.CCleanResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CClean indicates an expected call of CClean.
+func (mr *MockCommServerServerMockRecorder) CClean(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CClean", reflect.TypeOf((*MockCommServerServer)(nil).CClean), arg0, arg1)
+}
+
+// CLen mocks base method.
+func (m *MockCommServerServer) CLen(arg0 context.Context, arg1 *proto.CLenRequest) (*proto.CLenResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CLen", arg0, arg1)
+	ret0, _ := ret[0].(*proto.CLenResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CLen indicates an expected call of CLen.
+func (mr *MockCommServerServerMockRecorder) CLen(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CLen", reflect.TypeOf((*MockCommServerServer)(nil).CLen), arg0, arg1)
+}
+
+// CMake mocks base method.
+func (m *MockCommServerServer) CMake(arg0 context.Context, arg1 *proto.CMakeRequest) (*proto.CMakeResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CMake", arg0, arg1)
+	ret0, _ := ret[0].(*proto.CMakeResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CMake indicates an expected call of CMake.
+func (mr *MockCommServerServerMockRecorder) CMake(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CMake", reflect.TypeOf((*MockCommServerServer)(nil).CMake), arg0, arg1)
+}
+
+// CPop mocks base method.
+func (m *MockCommServerServer) CPop(arg0 context.Context, arg1 *proto.CPopRequest) (*proto.CPopResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CPop", arg0, arg1)
+	ret0, _ := ret[0].(*proto.CPopResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CPop indicates an expected call of CPop.
+func (mr *MockCommServerServerMockRecorder) CPop(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CPop", reflect.TypeOf((*MockCommServerServer)(nil).CPop), arg0, arg1)
+}
+
+// CPush mocks base method.
+func (m *MockCommServerServer) CPush(arg0 context.Context, arg1 *proto.CPushRequest) (*proto.CPushResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CPush", arg0, arg1)
+	ret0, _ := ret[0].(*proto.CPushResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CPush indicates an expected call of CPush.
+func (mr *MockCommServerServerMockRecorder) CPush(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CPush", reflect.TypeOf((*MockCommServerServer)(nil).CPush), arg0, arg1)
+}
+
 // Get mocks base method.
 func (m *MockCommServerServer) Get(arg0 context.Context, arg1 *proto.GetRequest) (*proto.GetResponse, error) {
 	m.ctrl.T.Helper()
--- a/pkg/lru/lru.go
+++ b/pkg/lru/lru.go
@ -123,6 +123,9 @@ func (lru *SingleCache) Add(key *proto.BaseKey, val structure.KeyBaseInterface)
 	}
 	if elVal, ok := lru.lruMap[key.Key]; ok {
 		lru.li.MoveToFront(elVal)
+		oldSize := elVal.Value.(structure.KeyBaseInterface).SizeByte()
+
+		lru.UpdateLruSize(structure.UpdateLength(val.SizeByte() - oldSize))
 		elVal.Value = keyBaseVal
 		return nil
 	}
--- a/pkg/proto/channelx.pb.go
+++ b/pkg/proto/channelx.pb.go
@ -0,0 +1,733 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.26.0
+// 	protoc        v3.17.3
+// source: channelx.proto
+
+package proto
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type CPushRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Key   *BaseKey `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
+	Value []string `protobuf:"bytes,2,rep,name=value,proto3" json:"value,omitempty"`
+}
+
+func (x *CPushRequest) Reset() {
+	*x = CPushRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CPushRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CPushRequest) ProtoMessage() {}
+
+func (x *CPushRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CPushRequest.ProtoReflect.Descriptor instead.
+func (*CPushRequest) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *CPushRequest) GetKey() *BaseKey {
+	if x != nil {
+		return x.Key
+	}
+	return nil
+}
+
+func (x *CPushRequest) GetValue() []string {
+	if x != nil {
+		return x.Value
+	}
+	return nil
+}
+
+type CPushResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	E *External `protobuf:"bytes,1,opt,name=e,proto3" json:"e,omitempty"`
+}
+
+func (x *CPushResponse) Reset() {
+	*x = CPushResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CPushResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CPushResponse) ProtoMessage() {}
+
+func (x *CPushResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CPushResponse.ProtoReflect.Descriptor instead.
+func (*CPushResponse) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *CPushResponse) GetE() *External {
+	if x != nil {
+		return x.E
+	}
+	return nil
+}
+
+type CPopRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Key   *BaseKey `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
+	Count int32    `protobuf:"varint,2,opt,name=count,proto3" json:"count,omitempty"`
+}
+
+func (x *CPopRequest) Reset() {
+	*x = CPopRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CPopRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CPopRequest) ProtoMessage() {}
+
+func (x *CPopRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CPopRequest.ProtoReflect.Descriptor instead.
+func (*CPopRequest) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *CPopRequest) GetKey() *BaseKey {
+	if x != nil {
+		return x.Key
+	}
+	return nil
+}
+
+func (x *CPopRequest) GetCount() int32 {
+	if x != nil {
+		return x.Count
+	}
+	return 0
+}
+
+type CPopResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	E      *External `protobuf:"bytes,1,opt,name=e,proto3" json:"e,omitempty"`
+	Result []string  `protobuf:"bytes,2,rep,name=result,proto3" json:"result,omitempty"`
+}
+
+func (x *CPopResponse) Reset() {
+	*x = CPopResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CPopResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CPopResponse) ProtoMessage() {}
+
+func (x *CPopResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CPopResponse.ProtoReflect.Descriptor instead.
+func (*CPopResponse) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *CPopResponse) GetE() *External {
+	if x != nil {
+		return x.E
+	}
+	return nil
+}
+
+func (x *CPopResponse) GetResult() []string {
+	if x != nil {
+		return x.Result
+	}
+	return nil
+}
+
+type CMakeRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Key    *BaseKey `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
+	Length int32    `protobuf:"varint,2,opt,name=length,proto3" json:"length,omitempty"`
+}
+
+func (x *CMakeRequest) Reset() {
+	*x = CMakeRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[4]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CMakeRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CMakeRequest) ProtoMessage() {}
+
+func (x *CMakeRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[4]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CMakeRequest.ProtoReflect.Descriptor instead.
+func (*CMakeRequest) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{4}
+}
+
+func (x *CMakeRequest) GetKey() *BaseKey {
+	if x != nil {
+		return x.Key
+	}
+	return nil
+}
+
+func (x *CMakeRequest) GetLength() int32 {
+	if x != nil {
+		return x.Length
+	}
+	return 0
+}
+
+type CMakeResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *CMakeResponse) Reset() {
+	*x = CMakeResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[5]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CMakeResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CMakeResponse) ProtoMessage() {}
+
+func (x *CMakeResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[5]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CMakeResponse.ProtoReflect.Descriptor instead.
+func (*CMakeResponse) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{5}
+}
+
+type CLenRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Key *BaseKey `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
+}
+
+func (x *CLenRequest) Reset() {
+	*x = CLenRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[6]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CLenRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CLenRequest) ProtoMessage() {}
+
+func (x *CLenRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[6]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CLenRequest.ProtoReflect.Descriptor instead.
+func (*CLenRequest) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{6}
+}
+
+func (x *CLenRequest) GetKey() *BaseKey {
+	if x != nil {
+		return x.Key
+	}
+	return nil
+}
+
+type CLenResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Length int32 `protobuf:"varint,2,opt,name=length,proto3" json:"length,omitempty"`
+}
+
+func (x *CLenResponse) Reset() {
+	*x = CLenResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[7]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CLenResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CLenResponse) ProtoMessage() {}
+
+func (x *CLenResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[7]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CLenResponse.ProtoReflect.Descriptor instead.
+func (*CLenResponse) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{7}
+}
+
+func (x *CLenResponse) GetLength() int32 {
+	if x != nil {
+		return x.Length
+	}
+	return 0
+}
+
+type CCleanRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Key *BaseKey `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
+}
+
+func (x *CCleanRequest) Reset() {
+	*x = CCleanRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[8]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CCleanRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CCleanRequest) ProtoMessage() {}
+
+func (x *CCleanRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[8]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CCleanRequest.ProtoReflect.Descriptor instead.
+func (*CCleanRequest) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{8}
+}
+
+func (x *CCleanRequest) GetKey() *BaseKey {
+	if x != nil {
+		return x.Key
+	}
+	return nil
+}
+
+type CCleanResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *CCleanResponse) Reset() {
+	*x = CCleanResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_channelx_proto_msgTypes[9]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CCleanResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CCleanResponse) ProtoMessage() {}
+
+func (x *CCleanResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_channelx_proto_msgTypes[9]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CCleanResponse.ProtoReflect.Descriptor instead.
+func (*CCleanResponse) Descriptor() ([]byte, []int) {
+	return file_channelx_proto_rawDescGZIP(), []int{9}
+}
+
+var File_channelx_proto protoreflect.FileDescriptor
+
+var file_channelx_proto_rawDesc = []byte{
+	0x0a, 0x0e, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x78, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x1a, 0x0a, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x40, 0x0a, 0x0c,
+	0x43, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x03,
+	0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x42, 0x61, 0x73, 0x65,
+	0x4b, 0x65, 0x79, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75,
+	0x65, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x28,
+	0x0a, 0x0d, 0x43, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12,
+	0x17, 0x0a, 0x01, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x45, 0x78, 0x74,
+	0x65, 0x72, 0x6e, 0x61, 0x6c, 0x52, 0x01, 0x65, 0x22, 0x3f, 0x0a, 0x0b, 0x43, 0x50, 0x6f, 0x70,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x52, 0x03,
+	0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x05, 0x52, 0x05, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x22, 0x3f, 0x0a, 0x0c, 0x43, 0x50, 0x6f,
+	0x70, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x17, 0x0a, 0x01, 0x65, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x45, 0x78, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x52,
+	0x01, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x18, 0x02, 0x20, 0x03,
+	0x28, 0x09, 0x52, 0x06, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x22, 0x42, 0x0a, 0x0c, 0x43, 0x4d,
+	0x61, 0x6b, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x03, 0x6b, 0x65,
+	0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x4b, 0x65,
+	0x79, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x16, 0x0a, 0x06, 0x6c, 0x65, 0x6e, 0x67, 0x74, 0x68,
+	0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x6c, 0x65, 0x6e, 0x67, 0x74, 0x68, 0x22, 0x0f,
+	0x0a, 0x0d, 0x43, 0x4d, 0x61, 0x6b, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
+	0x29, 0x0a, 0x0b, 0x43, 0x4c, 0x65, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a,
+	0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e, 0x42, 0x61,
+	0x73, 0x65, 0x4b, 0x65, 0x79, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0x26, 0x0a, 0x0c, 0x43, 0x4c,
+	0x65, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x6c, 0x65,
+	0x6e, 0x67, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x6c, 0x65, 0x6e, 0x67,
+	0x74, 0x68, 0x22, 0x2b, 0x0a, 0x0d, 0x43, 0x43, 0x6c, 0x65, 0x61, 0x6e, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x08, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22,
+	0x10, 0x0a, 0x0e, 0x43, 0x43, 0x6c, 0x65, 0x61, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x42, 0x0b, 0x5a, 0x09, 0x70, 0x6b, 0x67, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_channelx_proto_rawDescOnce sync.Once
+	file_channelx_proto_rawDescData = file_channelx_proto_rawDesc
+)
+
+func file_channelx_proto_rawDescGZIP() []byte {
+	file_channelx_proto_rawDescOnce.Do(func() {
+		file_channelx_proto_rawDescData = protoimpl.X.CompressGZIP(file_channelx_proto_rawDescData)
+	})
+	return file_channelx_proto_rawDescData
+}
+
+var file_channelx_proto_msgTypes = make([]protoimpl.MessageInfo, 10)
+var file_channelx_proto_goTypes = []interface{}{
+	(*CPushRequest)(nil),   // 0: CPushRequest
+	(*CPushResponse)(nil),  // 1: CPushResponse
+	(*CPopRequest)(nil),    // 2: CPopRequest
+	(*CPopResponse)(nil),   // 3: CPopResponse
+	(*CMakeRequest)(nil),   // 4: CMakeRequest
+	(*CMakeResponse)(nil),  // 5: CMakeResponse
+	(*CLenRequest)(nil),    // 6: CLenRequest
+	(*CLenResponse)(nil),   // 7: CLenResponse
+	(*CCleanRequest)(nil),  // 8: CCleanRequest
+	(*CCleanResponse)(nil), // 9: CCleanResponse
+	(*BaseKey)(nil),        // 10: BaseKey
+	(*External)(nil),       // 11: External
+}
+var file_channelx_proto_depIdxs = []int32{
+	10, // 0: CPushRequest.key:type_name -> BaseKey
+	11, // 1: CPushResponse.e:type_name -> External
+	10, // 2: CPopRequest.key:type_name -> BaseKey
+	11, // 3: CPopResponse.e:type_name -> External
+	10, // 4: CMakeRequest.key:type_name -> BaseKey
+	10, // 5: CLenRequest.key:type_name -> BaseKey
+	10, // 6: CCleanRequest.key:type_name -> BaseKey
+	7,  // [7:7] is the sub-list for method output_type
+	7,  // [7:7] is the sub-list for method input_type
+	7,  // [7:7] is the sub-list for extension type_name
+	7,  // [7:7] is the sub-list for extension extendee
+	0,  // [0:7] is the sub-list for field type_name
+}
+
+func init() { file_channelx_proto_init() }
+func file_channelx_proto_init() {
+	if File_channelx_proto != nil {
+		return
+	}
+	file_base_proto_init()
+	if !protoimpl.UnsafeEnabled {
+		file_channelx_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CPushRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CPushResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CPopRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CPopResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CMakeRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CMakeResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CLenRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CLenResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CCleanRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_channelx_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CCleanResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_channelx_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   10,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_channelx_proto_goTypes,
+		DependencyIndexes: file_channelx_proto_depIdxs,
+		MessageInfos:      file_channelx_proto_msgTypes,
+	}.Build()
+	File_channelx_proto = out.File
+	file_channelx_proto_rawDesc = nil
+	file_channelx_proto_goTypes = nil
+	file_channelx_proto_depIdxs = nil
+}
--- a/pkg/proto/storage.pb.go
+++ b/pkg/proto/storage.pb.go
@ -33,7 +33,8 @@ var file_storage_proto_rawDesc = []byte{
 	0x0d, 0x73, 0x74, 0x72, 0x69, 0x6e, 0x67, 0x78, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x0b,
 	0x6c, 0x69, 0x73, 0x74, 0x78, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x0b, 0x68, 0x61, 0x73,
 	0x68, 0x78, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x0a, 0x73, 0x65, 0x74, 0x78, 0x2e, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x32, 0xb1, 0x0e, 0x0a, 0x0a, 0x43, 0x6f, 0x6d, 0x6d, 0x53, 0x65, 0x72,
+	0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x0e, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x78, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x32, 0xf6, 0x0f, 0x0a, 0x0a, 0x43, 0x6f, 0x6d, 0x6d, 0x53, 0x65, 0x72,
 	0x76, 0x65, 0x72, 0x12, 0x20, 0x0a, 0x03, 0x53, 0x65, 0x74, 0x12, 0x0b, 0x2e, 0x53, 0x65, 0x74,
 	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0c, 0x2e, 0x53, 0x65, 0x74, 0x52, 0x65, 0x73,
 	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x20, 0x0a, 0x03, 0x47, 0x65, 0x74, 0x12, 0x0b, 0x2e, 0x47,
@ -148,8 +149,21 @@ var file_storage_proto_rawDesc = []byte{
 	0x69, 0x6f, 0x6e, 0x53, 0x74, 0x6f, 0x72, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
 	0x12, 0x26, 0x0a, 0x05, 0x53, 0x53, 0x63, 0x61, 0x6e, 0x12, 0x0d, 0x2e, 0x53, 0x53, 0x63, 0x61,
 	0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0e, 0x2e, 0x53, 0x53, 0x63, 0x61, 0x6e,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x0b, 0x5a, 0x09, 0x70, 0x6b, 0x67, 0x2f,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x26, 0x0a, 0x05, 0x43, 0x50, 0x75, 0x73,
+	0x68, 0x12, 0x0d, 0x2e, 0x43, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x0e, 0x2e, 0x43, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x12, 0x23, 0x0a, 0x04, 0x43, 0x50, 0x6f, 0x70, 0x12, 0x0c, 0x2e, 0x43, 0x50, 0x6f, 0x70, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0d, 0x2e, 0x43, 0x50, 0x6f, 0x70, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x26, 0x0a, 0x05, 0x43, 0x4d, 0x61, 0x6b, 0x65, 0x12, 0x0d,
+	0x2e, 0x43, 0x4d, 0x61, 0x6b, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0e, 0x2e,
+	0x43, 0x4d, 0x61, 0x6b, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x23, 0x0a,
+	0x04, 0x43, 0x4c, 0x65, 0x6e, 0x12, 0x0c, 0x2e, 0x43, 0x4c, 0x65, 0x6e, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x1a, 0x0d, 0x2e, 0x43, 0x4c, 0x65, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x12, 0x29, 0x0a, 0x06, 0x43, 0x43, 0x6c, 0x65, 0x61, 0x6e, 0x12, 0x0e, 0x2e, 0x43,
+	0x43, 0x6c, 0x65, 0x61, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0f, 0x2e, 0x43,
+	0x43, 0x6c, 0x65, 0x61, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x0b, 0x5a,
+	0x09, 0x70, 0x6b, 0x67, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x33,
 }

 var file_storage_proto_goTypes = []interface{}{
@ -197,50 +211,60 @@ var file_storage_proto_goTypes = []interface{}{
 	(*SUnionRequest)(nil),       // 41: SUnionRequest
 	(*SUnionStoreRequest)(nil),  // 42: SUnionStoreRequest
 	(*SScanRequest)(nil),        // 43: SScanRequest
-	(*SetResponse)(nil),         // 44: SetResponse
-	(*GetResponse)(nil),         // 45: GetResponse
-	(*AddResponse)(nil),         // 46: AddResponse
-	(*ReduceResponse)(nil),      // 47: ReduceResponse
-	(*SetBitResponse)(nil),      // 48: SetBitResponse
-	(*GetBitResponse)(nil),      // 49: GetBitResponse
-	(*GetRangeResponse)(nil),    // 50: GetRangeResponse
-	(*GetSetResponse)(nil),      // 51: GetSetResponse
-	(*StrLenResponse)(nil),      // 52: StrLenResponse
-	(*SetnxResponse)(nil),       // 53: SetnxResponse
-	(*LIndexResponse)(nil),      // 54: LIndexResponse
-	(*LLenResponse)(nil),        // 55: LLenResponse
-	(*LPopResponse)(nil),        // 56: LPopResponse
-	(*LPushResponse)(nil),       // 57: LPushResponse
-	(*LPushXResponse)(nil),      // 58: LPushXResponse
-	(*LRangeResponse)(nil),      // 59: LRangeResponse
-	(*LRemResponse)(nil),        // 60: LRemResponse
-	(*LSetResponse)(nil),        // 61: LSetResponse
-	(*RPopResponse)(nil),        // 62: RPopResponse
-	(*LTrimResponse)(nil),       // 63: LTrimResponse
-	(*RPushResponse)(nil),       // 64: RPushResponse
-	(*RPushXResponse)(nil),      // 65: RPushXResponse
-	(*HDelResponse)(nil),        // 66: HDelResponse
-	(*HExistsResponse)(nil),     // 67: HExistsResponse
-	(*HGetResponse)(nil),        // 68: HGetResponse
-	(*HGetAllResponse)(nil),     // 69: HGetAllResponse
-	(*HIncrByResponse)(nil),     // 70: HIncrByResponse
-	(*HKeysResponse)(nil),       // 71: HKeysResponse
-	(*HLenResponse)(nil),        // 72: HLenResponse
-	(*HSetResponse)(nil),        // 73: HSetResponse
-	(*HSetXResponse)(nil),       // 74: HSetXResponse
-	(*SAddResponse)(nil),        // 75: SAddResponse
-	(*SCardResponse)(nil),       // 76: SCardResponse
-	(*SDiffResponse)(nil),       // 77: SDiffResponse
-	(*SDiffStoreResponse)(nil),  // 78: SDiffStoreResponse
-	(*SInterResponse)(nil),      // 79: SInterResponse
-	(*SInterStoreResponse)(nil), // 80: SInterStoreResponse
-	(*SIsMemberResponse)(nil),   // 81: SIsMemberResponse
-	(*SMoveResponse)(nil),       // 82: SMoveResponse
-	(*SPopResponse)(nil),        // 83: SPopResponse
-	(*SRemResponse)(nil),        // 84: SRemResponse
-	(*SUnionResponse)(nil),      // 85: SUnionResponse
-	(*SUnionStoreResponse)(nil), // 86: SUnionStoreResponse
-	(*SScanResponse)(nil),       // 87: SScanResponse
+	(*CPushRequest)(nil),        // 44: CPushRequest
+	(*CPopRequest)(nil),         // 45: CPopRequest
+	(*CMakeRequest)(nil),        // 46: CMakeRequest
+	(*CLenRequest)(nil),         // 47: CLenRequest
+	(*CCleanRequest)(nil),       // 48: CCleanRequest
+	(*SetResponse)(nil),         // 49: SetResponse
+	(*GetResponse)(nil),         // 50: GetResponse
+	(*AddResponse)(nil),         // 51: AddResponse
+	(*ReduceResponse)(nil),      // 52: ReduceResponse
+	(*SetBitResponse)(nil),      // 53: SetBitResponse
+	(*GetBitResponse)(nil),      // 54: GetBitResponse
+	(*GetRangeResponse)(nil),    // 55: GetRangeResponse
+	(*GetSetResponse)(nil),      // 56: GetSetResponse
+	(*StrLenResponse)(nil),      // 57: StrLenResponse
+	(*SetnxResponse)(nil),       // 58: SetnxResponse
+	(*LIndexResponse)(nil),      // 59: LIndexResponse
+	(*LLenResponse)(nil),        // 60: LLenResponse
+	(*LPopResponse)(nil),        // 61: LPopResponse
+	(*LPushResponse)(nil),       // 62: LPushResponse
+	(*LPushXResponse)(nil),      // 63: LPushXResponse
+	(*LRangeResponse)(nil),      // 64: LRangeResponse
+	(*LRemResponse)(nil),        // 65: LRemResponse
+	(*LSetResponse)(nil),        // 66: LSetResponse
+	(*RPopResponse)(nil),        // 67: RPopResponse
+	(*LTrimResponse)(nil),       // 68: LTrimResponse
+	(*RPushResponse)(nil),       // 69: RPushResponse
+	(*RPushXResponse)(nil),      // 70: RPushXResponse
+	(*HDelResponse)(nil),        // 71: HDelResponse
+	(*HExistsResponse)(nil),     // 72: HExistsResponse
+	(*HGetResponse)(nil),        // 73: HGetResponse
+	(*HGetAllResponse)(nil),     // 74: HGetAllResponse
+	(*HIncrByResponse)(nil),     // 75: HIncrByResponse
+	(*HKeysResponse)(nil),       // 76: HKeysResponse
+	(*HLenResponse)(nil),        // 77: HLenResponse
+	(*HSetResponse)(nil),        // 78: HSetResponse
+	(*HSetXResponse)(nil),       // 79: HSetXResponse
+	(*SAddResponse)(nil),        // 80: SAddResponse
+	(*SCardResponse)(nil),       // 81: SCardResponse
+	(*SDiffResponse)(nil),       // 82: SDiffResponse
+	(*SDiffStoreResponse)(nil),  // 83: SDiffStoreResponse
+	(*SInterResponse)(nil),      // 84: SInterResponse
+	(*SInterStoreResponse)(nil), // 85: SInterStoreResponse
+	(*SIsMemberResponse)(nil),   // 86: SIsMemberResponse
+	(*SMoveResponse)(nil),       // 87: SMoveResponse
+	(*SPopResponse)(nil),        // 88: SPopResponse
+	(*SRemResponse)(nil),        // 89: SRemResponse
+	(*SUnionResponse)(nil),      // 90: SUnionResponse
+	(*SUnionStoreResponse)(nil), // 91: SUnionStoreResponse
+	(*SScanResponse)(nil),       // 92: SScanResponse
+	(*CPushResponse)(nil),       // 93: CPushResponse
+	(*CPopResponse)(nil),        // 94: CPopResponse
+	(*CMakeResponse)(nil),       // 95: CMakeResponse
+	(*CLenResponse)(nil),        // 96: CLenResponse
+	(*CCleanResponse)(nil),      // 97: CCleanResponse
 }
 var file_storage_proto_depIdxs = []int32{
 	0,  // 0: CommServer.Set:input_type -> SetRequest
@ -287,52 +311,62 @@ var file_storage_proto_depIdxs = []int32{
 	41, // 41: CommServer.SUnion:input_type -> SUnionRequest
 	42, // 42: CommServer.SUnionStore:input_type -> SUnionStoreRequest
 	43, // 43: CommServer.SScan:input_type -> SScanRequest
-	44, // 44: CommServer.Set:output_type -> SetResponse
-	45, // 45: CommServer.Get:output_type -> GetResponse
-	46, // 46: CommServer.Add:output_type -> AddResponse
-	47, // 47: CommServer.Reduce:output_type -> ReduceResponse
-	48, // 48: CommServer.SetBit:output_type -> SetBitResponse
-	49, // 49: CommServer.GetBit:output_type -> GetBitResponse
-	50, // 50: CommServer.GetRange:output_type -> GetRangeResponse
-	51, // 51: CommServer.GetSet:output_type -> GetSetResponse
-	52, // 52: CommServer.StrLen:output_type -> StrLenResponse
-	53, // 53: CommServer.Setnx:output_type -> SetnxResponse
-	54, // 54: CommServer.LIndex:output_type -> LIndexResponse
-	55, // 55: CommServer.LLen:output_type -> LLenResponse
-	56, // 56: CommServer.LPop:output_type -> LPopResponse
-	57, // 57: CommServer.LPush:output_type -> LPushResponse
-	58, // 58: CommServer.LPushX:output_type -> LPushXResponse
-	59, // 59: CommServer.LRange:output_type -> LRangeResponse
-	60, // 60: CommServer.LRem:output_type -> LRemResponse
-	61, // 61: CommServer.LSet:output_type -> LSetResponse
-	62, // 62: CommServer.RPop:output_type -> RPopResponse
-	63, // 63: CommServer.LTrim:output_type -> LTrimResponse
-	64, // 64: CommServer.RPush:output_type -> RPushResponse
-	65, // 65: CommServer.RPushX:output_type -> RPushXResponse
-	66, // 66: CommServer.HDel:output_type -> HDelResponse
-	67, // 67: CommServer.HExists:output_type -> HExistsResponse
-	68, // 68: CommServer.HGet:output_type -> HGetResponse
-	69, // 69: CommServer.HGetAll:output_type -> HGetAllResponse
-	70, // 70: CommServer.HIncrBy:output_type -> HIncrByResponse
-	71, // 71: CommServer.HKeys:output_type -> HKeysResponse
-	72, // 72: CommServer.HLen:output_type -> HLenResponse
-	73, // 73: CommServer.HSet:output_type -> HSetResponse
-	74, // 74: CommServer.HSetX:output_type -> HSetXResponse
-	75, // 75: CommServer.SAdd:output_type -> SAddResponse
-	76, // 76: CommServer.SCard:output_type -> SCardResponse
-	77, // 77: CommServer.SDiff:output_type -> SDiffResponse
-	78, // 78: CommServer.SDiffStore:output_type -> SDiffStoreResponse
-	79, // 79: CommServer.SInter:output_type -> SInterResponse
-	80, // 80: CommServer.SInterStore:output_type -> SInterStoreResponse
-	81, // 81: CommServer.SIsMember:output_type -> SIsMemberResponse
-	82, // 82: CommServer.SMove:output_type -> SMoveResponse
-	83, // 83: CommServer.SPop:output_type -> SPopResponse
-	84, // 84: CommServer.SRem:output_type -> SRemResponse
-	85, // 85: CommServer.SUnion:output_type -> SUnionResponse
-	86, // 86: CommServer.SUnionStore:output_type -> SUnionStoreResponse
-	87, // 87: CommServer.SScan:output_type -> SScanResponse
-	44, // [44:88] is the sub-list for method output_type
-	0,  // [0:44] is the sub-list for method input_type
+	44, // 44: CommServer.CPush:input_type -> CPushRequest
+	45, // 45: CommServer.CPop:input_type -> CPopRequest
+	46, // 46: CommServer.CMake:input_type -> CMakeRequest
+	47, // 47: CommServer.CLen:input_type -> CLenRequest
+	48, // 48: CommServer.CClean:input_type -> CCleanRequest
+	49, // 49: CommServer.Set:output_type -> SetResponse
+	50, // 50: CommServer.Get:output_type -> GetResponse
+	51, // 51: CommServer.Add:output_type -> AddResponse
+	52, // 52: CommServer.Reduce:output_type -> ReduceResponse
+	53, // 53: CommServer.SetBit:output_type -> SetBitResponse
+	54, // 54: CommServer.GetBit:output_type -> GetBitResponse
+	55, // 55: CommServer.GetRange:output_type -> GetRangeResponse
+	56, // 56: CommServer.GetSet:output_type -> GetSetResponse
+	57, // 57: CommServer.StrLen:output_type -> StrLenResponse
+	58, // 58: CommServer.Setnx:output_type -> SetnxResponse
+	59, // 59: CommServer.LIndex:output_type -> LIndexResponse
+	60, // 60: CommServer.LLen:output_type -> LLenResponse
+	61, // 61: CommServer.LPop:output_type -> LPopResponse
+	62, // 62: CommServer.LPush:output_type -> LPushResponse
+	63, // 63: CommServer.LPushX:output_type -> LPushXResponse
+	64, // 64: CommServer.LRange:output_type -> LRangeResponse
+	65, // 65: CommServer.LRem:output_type -> LRemResponse
+	66, // 66: CommServer.LSet:output_type -> LSetResponse
+	67, // 67: CommServer.RPop:output_type -> RPopResponse
+	68, // 68: CommServer.LTrim:output_type -> LTrimResponse
+	69, // 69: CommServer.RPush:output_type -> RPushResponse
+	70, // 70: CommServer.RPushX:output_type -> RPushXResponse
+	71, // 71: CommServer.HDel:output_type -> HDelResponse
+	72, // 72: CommServer.HExists:output_type -> HExistsResponse
+	73, // 73: CommServer.HGet:output_type -> HGetResponse
+	74, // 74: CommServer.HGetAll:output_type -> HGetAllResponse
+	75, // 75: CommServer.HIncrBy:output_type -> HIncrByResponse
+	76, // 76: CommServer.HKeys:output_type -> HKeysResponse
+	77, // 77: CommServer.HLen:output_type -> HLenResponse
+	78, // 78: CommServer.HSet:output_type -> HSetResponse
+	79, // 79: CommServer.HSetX:output_type -> HSetXResponse
+	80, // 80: CommServer.SAdd:output_type -> SAddResponse
+	81, // 81: CommServer.SCard:output_type -> SCardResponse
+	82, // 82: CommServer.SDiff:output_type -> SDiffResponse
+	83, // 83: CommServer.SDiffStore:output_type -> SDiffStoreResponse
+	84, // 84: CommServer.SInter:output_type -> SInterResponse
+	85, // 85: CommServer.SInterStore:output_type -> SInterStoreResponse
+	86, // 86: CommServer.SIsMember:output_type -> SIsMemberResponse
+	87, // 87: CommServer.SMove:output_type -> SMoveResponse
+	88, // 88: CommServer.SPop:output_type -> SPopResponse
+	89, // 89: CommServer.SRem:output_type -> SRemResponse
+	90, // 90: CommServer.SUnion:output_type -> SUnionResponse
+	91, // 91: CommServer.SUnionStore:output_type -> SUnionStoreResponse
+	92, // 92: CommServer.SScan:output_type -> SScanResponse
+	93, // 93: CommServer.CPush:output_type -> CPushResponse
+	94, // 94: CommServer.CPop:output_type -> CPopResponse
+	95, // 95: CommServer.CMake:output_type -> CMakeResponse
+	96, // 96: CommServer.CLen:output_type -> CLenResponse
+	97, // 97: CommServer.CClean:output_type -> CCleanResponse
+	49, // [49:98] is the sub-list for method output_type
+	0,  // [0:49] is the sub-list for method input_type
 	0,  // [0:0] is the sub-list for extension type_name
 	0,  // [0:0] is the sub-list for extension extendee
 	0,  // [0:0] is the sub-list for field type_name
@ -347,6 +381,7 @@ func file_storage_proto_init() {
 	file_listx_proto_init()
 	file_hashx_proto_init()
 	file_setx_proto_init()
+	file_channelx_proto_init()
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
@ -422,6 +457,11 @@ type CommServerClient interface {
 	SUnion(ctx context.Context, in *SUnionRequest, opts ...grpc.CallOption) (*SUnionResponse, error)
 	SUnionStore(ctx context.Context, in *SUnionStoreRequest, opts ...grpc.CallOption) (*SUnionStoreResponse, error)
 	SScan(ctx context.Context, in *SScanRequest, opts ...grpc.CallOption) (*SScanResponse, error)
+	CPush(ctx context.Context, in *CPushRequest, opts ...grpc.CallOption) (*CPushResponse, error)
+	CPop(ctx context.Context, in *CPopRequest, opts ...grpc.CallOption) (*CPopResponse, error)
+	CMake(ctx context.Context, in *CMakeRequest, opts ...grpc.CallOption) (*CMakeResponse, error)
+	CLen(ctx context.Context, in *CLenRequest, opts ...grpc.CallOption) (*CLenResponse, error)
+	CClean(ctx context.Context, in *CCleanRequest, opts ...grpc.CallOption) (*CCleanResponse, error)
 }

 type commServerClient struct {
@ -828,6 +868,51 @@ func (c *commServerClient) SScan(ctx context.Context, in *SScanRequest, opts ...
 	return out, nil
 }

+func (c *commServerClient) CPush(ctx context.Context, in *CPushRequest, opts ...grpc.CallOption) (*CPushResponse, error) {
+	out := new(CPushResponse)
+	err := c.cc.Invoke(ctx, "/CommServer/CPush", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *commServerClient) CPop(ctx context.Context, in *CPopRequest, opts ...grpc.CallOption) (*CPopResponse, error) {
+	out := new(CPopResponse)
+	err := c.cc.Invoke(ctx, "/CommServer/CPop", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *commServerClient) CMake(ctx context.Context, in *CMakeRequest, opts ...grpc.CallOption) (*CMakeResponse, error) {
+	out := new(CMakeResponse)
+	err := c.cc.Invoke(ctx, "/CommServer/CMake", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *commServerClient) CLen(ctx context.Context, in *CLenRequest, opts ...grpc.CallOption) (*CLenResponse, error) {
+	out := new(CLenResponse)
+	err := c.cc.Invoke(ctx, "/CommServer/CLen", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *commServerClient) CClean(ctx context.Context, in *CCleanRequest, opts ...grpc.CallOption) (*CCleanResponse, error) {
+	out := new(CCleanResponse)
+	err := c.cc.Invoke(ctx, "/CommServer/CClean", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // CommServerServer is the server API for CommServer service.
 type CommServerServer interface {
 	Set(context.Context, *SetRequest) (*SetResponse, error)
@ -874,6 +959,11 @@ type CommServerServer interface {
 	SUnion(context.Context, *SUnionRequest) (*SUnionResponse, error)
 	SUnionStore(context.Context, *SUnionStoreRequest) (*SUnionStoreResponse, error)
 	SScan(context.Context, *SScanRequest) (*SScanResponse, error)
+	CPush(context.Context, *CPushRequest) (*CPushResponse, error)
+	CPop(context.Context, *CPopRequest) (*CPopResponse, error)
+	CMake(context.Context, *CMakeRequest) (*CMakeResponse, error)
+	CLen(context.Context, *CLenRequest) (*CLenResponse, error)
+	CClean(context.Context, *CCleanRequest) (*CCleanResponse, error)
 }

 // UnimplementedCommServerServer can be embedded to have forward compatible implementations.
@ -1012,6 +1102,21 @@ func (*UnimplementedCommServerServer) SUnionStore(context.Context, *SUnionStoreR
 func (*UnimplementedCommServerServer) SScan(context.Context, *SScanRequest) (*SScanResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method SScan not implemented")
 }
+func (*UnimplementedCommServerServer) CPush(context.Context, *CPushRequest) (*CPushResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CPush not implemented")
+}
+func (*UnimplementedCommServerServer) CPop(context.Context, *CPopRequest) (*CPopResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CPop not implemented")
+}
+func (*UnimplementedCommServerServer) CMake(context.Context, *CMakeRequest) (*CMakeResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CMake not implemented")
+}
+func (*UnimplementedCommServerServer) CLen(context.Context, *CLenRequest) (*CLenResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CLen not implemented")
+}
+func (*UnimplementedCommServerServer) CClean(context.Context, *CCleanRequest) (*CCleanResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CClean not implemented")
+}

 func RegisterCommServerServer(s *grpc.Server, srv CommServerServer) {
 	s.RegisterService(&_CommServer_serviceDesc, srv)
@ -1809,6 +1914,96 @@ func _CommServer_SScan_Handler(srv interface{}, ctx context.Context, dec func(in
 	return interceptor(ctx, in, info, handler)
 }

+func _CommServer_CPush_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(CPushRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CommServerServer).CPush(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/CommServer/CPush",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CommServerServer).CPush(ctx, req.(*CPushRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _CommServer_CPop_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(CPopRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CommServerServer).CPop(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/CommServer/CPop",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CommServerServer).CPop(ctx, req.(*CPopRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _CommServer_CMake_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(CMakeRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CommServerServer).CMake(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/CommServer/CMake",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CommServerServer).CMake(ctx, req.(*CMakeRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _CommServer_CLen_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(CLenRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CommServerServer).CLen(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/CommServer/CLen",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CommServerServer).CLen(ctx, req.(*CLenRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _CommServer_CClean_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(CCleanRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CommServerServer).CClean(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/CommServer/CClean",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CommServerServer).CClean(ctx, req.(*CCleanRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 var _CommServer_serviceDesc = grpc.ServiceDesc{
 	ServiceName: "CommServer",
 	HandlerType: (*CommServerServer)(nil),
@ -1989,6 +2184,26 @@ var _CommServer_serviceDesc = grpc.ServiceDesc{
 			MethodName: "SScan",
 			Handler:    _CommServer_SScan_Handler,
 		},
+		{
+			MethodName: "CPush",
+			Handler:    _CommServer_CPush_Handler,
+		},
+		{
+			MethodName: "CPop",
+			Handler:    _CommServer_CPop_Handler,
+		},
+		{
+			MethodName: "CMake",
+			Handler:    _CommServer_CMake_Handler,
+		},
+		{
+			MethodName: "CLen",
+			Handler:    _CommServer_CLen_Handler,
+		},
+		{
+			MethodName: "CClean",
+			Handler:    _CommServer_CClean_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "storage.proto",
--- a/pkg/structure/channelx/channelx.go
+++ b/pkg/structure/channelx/channelx.go
@ -0,0 +1,66 @@
+package channelx
+
+import (
+	"sync/atomic"
+
+	"gitee.com/wheat-os/wheatCache/pkg/structure"
+)
+
+type ChannelX struct {
+	channel  chan *structure.Value
+	sizeByte int64
+}
+
+func MakeChannelX(length int) structure.ChannelXInterface {
+	return &ChannelX{
+		channel:  make(chan *structure.Value, length),
+		sizeByte: 0,
+	}
+}
+
+func (c *ChannelX) SizeByte() int64 {
+	return c.sizeByte
+}
+
+// RollBack TODO 事务相关, V2 实现
+func (c *ChannelX) RollBack() error {
+	panic("not implemented") // TODO: Implement
+}
+
+// Begin 事务相关, V2 实现
+func (c *ChannelX) Begin() error {
+	panic("not implemented") // TODO: Implement
+}
+
+// Comment 事务相关, V2 实现
+func (c *ChannelX) Comment() error {
+	panic("not implemented") // TODO: Implement
+}
+
+func (c *ChannelX) Encode() ([]byte, error) {
+	panic("not implemented") // TODO: Implement
+}
+
+func (c *ChannelX) Push(value string) structure.UpdateLength {
+	val := structure.NewValue(value)
+	up := val.GetSize()
+	c.channel <- val
+	atomic.AddInt64(&c.sizeByte, int64(up))
+	return structure.UpdateLength(up)
+}
+
+func (c *ChannelX) Pop() (string, structure.UpdateLength) {
+	val := <-c.channel
+	return val.ToString(), structure.UpdateLength(val.GetSize()) * -1
+}
+
+func (c *ChannelX) Length() int {
+	return len(c.channel)
+}
+
+func (c *ChannelX) Clean() structure.UpdateLength {
+	c.channel = make(chan *structure.Value, cap(c.channel))
+	up := c.sizeByte
+	c.sizeByte = 0
+	return structure.UpdateLength(up) * -1
+}
--- a/pkg/structure/channelx/channelx_test.go
+++ b/pkg/structure/channelx/channelx_test.go
@ -0,0 +1,24 @@
+package channelx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestChannelX_Push(t *testing.T) {
+	c := MakeChannelX(10)
+	require.Equal(t, c.Length(), 0)
+
+	up := c.Push("111")
+	require.Equal(t, 24, int(up))
+
+	res, up := c.Pop()
+	require.Equal(t, -24, int(up))
+	require.Equal(t, res, "111")
+
+	up = c.Push("111")
+	c.CLean()
+
+	require.Equal(t, c.Length(), 0)
+}
--- a/pkg/structure/define.go
+++ b/pkg/structure/define.go
@ -67,3 +67,11 @@ type HashXInterface interface {
 	Length() int
 	Range(consur, count int, regex string) []string
 }
+
+type ChannelXInterface interface {
+	KeyBaseInterface
+	Push(value string) UpdateLength
+	Pop() (string, UpdateLength)
+	Length() int
+	Clean() UpdateLength
+}
--- a/protobuf/channelx.proto
+++ b/protobuf/channelx.proto
@ -0,0 +1,45 @@
+syntax = "proto3";
+import "base.proto";
+option go_package = "pkg/proto";
+
+message CPushRequest {
+  BaseKey key = 1;
+  repeated string value = 2;
+}
+
+message CPushResponse {
+  External e = 1;
+}
+
+message CPopRequest {
+  BaseKey key = 1;
+  int32 count = 2;
+}
+
+message CPopResponse {
+  External e = 1;
+  repeated string result = 2;
+}
+
+message CMakeRequest {
+  BaseKey key = 1;
+  int32 length = 2;
+}
+
+message CMakeResponse {
+}
+
+message CLenRequest {
+  BaseKey key = 1;
+}
+
+message CLenResponse {
+  int32 length = 2;
+}
+
+message CCleanRequest {
+  BaseKey key = 1;
+}
+
+message CCleanResponse {
+}
--- a/protobuf/storage.proto
+++ b/protobuf/storage.proto
@ -9,6 +9,7 @@ import "stringx.proto";
 import "listx.proto";
 import "hashx.proto";
 import "setx.proto";
+import "channelx.proto";


 service CommServer {
@ -56,4 +57,9 @@ service CommServer {
  rpc SUnion (SUnionRequest) returns (SUnionResponse);
  rpc SUnionStore (SUnionStoreRequest) returns (SUnionStoreResponse);
  rpc SScan (SScanRequest) returns (SScanResponse);
+  rpc CPush (CPushRequest) returns (CPushResponse);
+  rpc CPop (CPopRequest) returns (CPopResponse);
+  rpc CMake (CMakeRequest) returns (CMakeResponse);
+  rpc CLen (CLenRequest) returns (CLenResponse);
+  rpc CClean (CCleanRequest) returns (CCleanResponse);
 }
--- a/storage/dao/channelx.go
+++ b/storage/dao/channelx.go
@ -0,0 +1,94 @@
+package dao
+
+import (
+	"gitee.com/wheat-os/wheatCache/pkg/errorx"
+	"gitee.com/wheat-os/wheatCache/pkg/event2"
+	"gitee.com/wheat-os/wheatCache/pkg/proto"
+	"gitee.com/wheat-os/wheatCache/pkg/structure"
+	"gitee.com/wheat-os/wheatCache/pkg/structure/channelx"
+)
+
+func (d *Dao) CPush(key *proto.BaseKey, Value []string) (interface{}, error) {
+	val, ok := d.lru.Get(key)
+	if !ok {
+		return nil, errorx.NotKeyErr(key.Key)
+	}
+	chanVal, ok := val.(structure.ChannelXInterface)
+	if !ok {
+		return nil, errorx.DaoTypeErr("channelx")
+	}
+
+	return event2.EventAwaitFunc(func() (interface{}, error) {
+		var sumUp structure.UpdateLength
+		for _, v := range Value {
+			sumUp += chanVal.Push(v)
+		}
+
+		d.lru.UpdateLruSize(sumUp)
+
+		return &proto.CPushResponse{}, nil
+	}), nil
+
+}
+
+func (d *Dao) CPop(key *proto.BaseKey, count int32) (interface{}, error) {
+	val, ok := d.lru.Get(key)
+	if !ok {
+		return nil, errorx.NotKeyErr(key.Key)
+	}
+	chanVal, ok := val.(structure.ChannelXInterface)
+	if !ok {
+		return nil, errorx.DaoTypeErr("channelx")
+	}
+
+	return event2.EventAwaitFunc(func() (interface{}, error) {
+		var sumUp structure.UpdateLength
+		result := make([]string, 0, count)
+		for i := int32(0); i < count; i++ {
+			v, up := chanVal.Pop()
+			sumUp += up
+			result = append(result, v)
+		}
+
+		d.lru.UpdateLruSize(sumUp)
+		return &proto.CPopResponse{Result: result}, nil
+	}), nil
+}
+
+func (d *Dao) CMake(key *proto.BaseKey, length int32) (*proto.CMakeResponse, error) {
+	chanVal := channelx.MakeChannelX(int(length))
+	err := d.lru.Add(key, chanVal)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.CMakeResponse{}, nil
+}
+
+func (d *Dao) CLen(key *proto.BaseKey) (*proto.CLenResponse, error) {
+	val, ok := d.lru.Get(key)
+	if !ok {
+		return nil, errorx.NotKeyErr(key.Key)
+	}
+	chanVal, ok := val.(structure.ChannelXInterface)
+	if !ok {
+		return nil, errorx.DaoTypeErr("channelx")
+	}
+
+	return &proto.CLenResponse{Length: int32(chanVal.Length())}, nil
+}
+
+func (d *Dao) CClean(key *proto.BaseKey) (*proto.CCleanResponse, error) {
+	val, ok := d.lru.Get(key)
+	if !ok {
+		return nil, errorx.NotKeyErr(key.Key)
+	}
+	chanVal, ok := val.(structure.ChannelXInterface)
+	if !ok {
+		return nil, errorx.DaoTypeErr("channelx")
+	}
+	up := chanVal.Clean()
+	d.lru.UpdateLruSize(up)
+
+	return &proto.CCleanResponse{}, nil
+}
--- a/storage/dao/channelx_test.go
+++ b/storage/dao/channelx_test.go
@ -0,0 +1,70 @@
+package dao
+
+import (
+	"strconv"
+	"testing"
+
+	"gitee.com/wheat-os/wheatCache/pkg/event2"
+	"gitee.com/wheat-os/wheatCache/pkg/lru"
+	"gitee.com/wheat-os/wheatCache/pkg/proto"
+	"github.com/stretchr/testify/require"
+)
+
+func execWaitFunc(t *testing.T, work interface{}) (interface{}, error) {
+	wait, ok := work.(event2.EventAwaitFunc)
+	require.True(t, ok)
+
+	return wait()
+}
+
+func TestDao_CPush_CPop(t *testing.T) {
+	testBaseKey := proto.NewBaseKey("123")
+	l := lru.NewLRUCache()
+	dao := NewDao(l)
+
+	_, err := dao.CMake(testBaseKey, 200)
+	require.NoError(t, err)
+
+	wait, err := dao.CPush(testBaseKey, []string{"1", "2", "3"})
+	require.NoError(t, err)
+	waitFunc, ok := wait.(event2.EventAwaitFunc)
+	require.True(t, ok)
+	_, err = waitFunc()
+	require.NoError(t, err)
+
+	wait, err = dao.CPop(testBaseKey, 3)
+	require.NoError(t, err)
+
+	res, err := execWaitFunc(t, wait)
+	require.NoError(t, err)
+	require.Equal(t, res.(*proto.CPopResponse).Result, []string{"1", "2", "3"})
+}
+
+func TestDao_Async_Push(t *testing.T) {
+	testBaseKey := proto.NewBaseKey("123")
+	l := lru.NewLRUCache()
+	dao := NewDao(l)
+
+	_, err := dao.CMake(testBaseKey, 200)
+	require.NoError(t, err)
+
+	rq := make([]string, 0)
+	for i := 0; i < 300; i++ {
+		rq = append(rq, strconv.Itoa(i))
+	}
+
+	go func() {
+
+		wait, err := dao.CPush(testBaseKey, rq)
+		require.NoError(t, err)
+		_, err = execWaitFunc(t, wait)
+		require.NoError(t, err)
+	}()
+
+	wait, err := dao.CPop(testBaseKey, 300)
+	require.NoError(t, err)
+	resp, err := execWaitFunc(t, wait)
+	require.NoError(t, err)
+
+	require.Equal(t, resp.(*proto.CPopResponse).Result, rq)
+}
--- a/storage/dao/dao.gen.go
+++ b/storage/dao/dao.gen.go
@ -86,6 +86,21 @@ func (d *Dao) ExecMessage(message protobuf.Message) error {
 	case *proto.HSetXRequest:
 		_, err := d.HSetX(req.Key, req.Items)
 		return err
+	case *proto.CPushRequest:
+		_, err := d.CPush(req.Key, req.Value)
+		return err
+	case *proto.CPopRequest:
+		_, err := d.CPop(req.Key, req.Count)
+		return err
+	case *proto.CMakeRequest:
+		_, err := d.CMake(req.Key, req.Length)
+		return err
+	case *proto.CLenRequest:
+		_, err := d.CLen(req.Key)
+		return err
+	case *proto.CCleanRequest:
+		_, err := d.CClean(req.Key)
+		return err
 	case *proto.SetRequest:
 		_, err := d.Set(req.Key, req.Val)
 		return err
--- a/storage/dao/interface.gen.go
+++ b/storage/dao/interface.gen.go
@ -30,6 +30,11 @@ type Interface interface {
 	HLen(*proto.BaseKey) (*proto.HLenResponse, error)
 	HSet(*proto.BaseKey, map[string]string) (*proto.HSetResponse, error)
 	HSetX(*proto.BaseKey, map[string]string) (*proto.HSetXResponse, error)
+	CPush(*proto.BaseKey, []string) (interface{}, error)
+	CPop(*proto.BaseKey, int32) (interface{}, error)
+	CMake(*proto.BaseKey, int32) (*proto.CMakeResponse, error)
+	CLen(*proto.BaseKey) (*proto.CLenResponse, error)
+	CClean(*proto.BaseKey) (*proto.CCleanResponse, error)
 	Set(*proto.BaseKey, string) (*proto.SetResponse, error)
 	Get(*proto.BaseKey) (*proto.GetResponse, error)
 	Add(*proto.BaseKey, int32) (*proto.AddResponse, error)
--- a/storage/persisted/codec.gen.go
+++ b/storage/persisted/codec.gen.go
@ -90,6 +90,16 @@ func decode(method string, buf []byte) (protobuf.Message, error) {
 		return decodeHSet(buf)
 	case "HSetX":
 		return decodeHSetX(buf)
+	case "CPush":
+		return decodeCPush(buf)
+	case "CPop":
+		return decodeCPop(buf)
+	case "CMake":
+		return decodeCMake(buf)
+	case "CLen":
+		return decodeCLen(buf)
+	case "CClean":
+		return decodeCClean(buf)
 	case "Set":
 		return decodeSet(buf)
 	case "Get":
@ -331,6 +341,51 @@ func decodeHSetX(buf []byte) (*proto.HSetXRequest, error) {
 	return req, nil
 }

+func decodeCPush(buf []byte) (*proto.CPushRequest, error) {
+	req := &proto.CPushRequest{}
+	err := protobuf.Unmarshal(buf, req)
+	if err != nil {
+		return nil, err
+	}
+	return req, nil
+}
+
+func decodeCPop(buf []byte) (*proto.CPopRequest, error) {
+	req := &proto.CPopRequest{}
+	err := protobuf.Unmarshal(buf, req)
+	if err != nil {
+		return nil, err
+	}
+	return req, nil
+}
+
+func decodeCMake(buf []byte) (*proto.CMakeRequest, error) {
+	req := &proto.CMakeRequest{}
+	err := protobuf.Unmarshal(buf, req)
+	if err != nil {
+		return nil, err
+	}
+	return req, nil
+}
+
+func decodeCLen(buf []byte) (*proto.CLenRequest, error) {
+	req := &proto.CLenRequest{}
+	err := protobuf.Unmarshal(buf, req)
+	if err != nil {
+		return nil, err
+	}
+	return req, nil
+}
+
+func decodeCClean(buf []byte) (*proto.CCleanRequest, error) {
+	req := &proto.CCleanRequest{}
+	err := protobuf.Unmarshal(buf, req)
+	if err != nil {
+		return nil, err
+	}
+	return req, nil
+}
+
 func decodeSet(buf []byte) (*proto.SetRequest, error) {
 	req := &proto.SetRequest{}
 	err := protobuf.Unmarshal(buf, req)
--- a/storage/service/single_service.gen.go
+++ b/storage/service/single_service.gen.go
@ -598,6 +598,146 @@ func (s *singleService) HSetX(
 	return resp.(*proto.HSetXResponse), nil
 }

+func (s *singleService) CPush(
+	ctx context.Context,
+	req *proto.CPushRequest,
+) (*proto.CPushResponse, error) {
+	work := event2.EventWorkFunc(func() (interface{}, error) {
+		resp, err := s.dao.CPush(req.Key, req.Value)
+		if err != nil {
+			return nil, err
+		}
+		if s.aof != nil {
+			s.aof.SendRequest("CPush", req)
+		}
+		return resp, nil
+	})
+
+	lruEvent := s.lruProduce.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(event2.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	lruEvent.Recovery()
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.(*proto.CPushResponse), nil
+}
+
+func (s *singleService) CPop(
+	ctx context.Context,
+	req *proto.CPopRequest,
+) (*proto.CPopResponse, error) {
+	work := event2.EventWorkFunc(func() (interface{}, error) {
+		resp, err := s.dao.CPop(req.Key, req.Count)
+		if err != nil {
+			return nil, err
+		}
+		if s.aof != nil {
+			s.aof.SendRequest("CPop", req)
+		}
+		return resp, nil
+	})
+
+	lruEvent := s.lruProduce.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(event2.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	lruEvent.Recovery()
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.(*proto.CPopResponse), nil
+}
+
+func (s *singleService) CMake(
+	ctx context.Context,
+	req *proto.CMakeRequest,
+) (*proto.CMakeResponse, error) {
+	work := event2.EventWorkFunc(func() (interface{}, error) {
+		resp, err := s.dao.CMake(req.Key, req.Length)
+		if err != nil {
+			return nil, err
+		}
+		if s.aof != nil {
+			s.aof.SendRequest("CMake", req)
+		}
+		return resp, nil
+	})
+
+	lruEvent := s.lruProduce.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(event2.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	lruEvent.Recovery()
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.(*proto.CMakeResponse), nil
+}
+
+func (s *singleService) CLen(
+	ctx context.Context,
+	req *proto.CLenRequest,
+) (*proto.CLenResponse, error) {
+	work := event2.EventWorkFunc(func() (interface{}, error) {
+		resp, err := s.dao.CLen(req.Key)
+		if err != nil {
+			return nil, err
+		}
+		if s.aof != nil {
+			s.aof.SendRequest("CLen", req)
+		}
+		return resp, nil
+	})
+
+	lruEvent := s.lruProduce.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(event2.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	lruEvent.Recovery()
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.(*proto.CLenResponse), nil
+}
+
+func (s *singleService) CClean(
+	ctx context.Context,
+	req *proto.CCleanRequest,
+) (*proto.CCleanResponse, error) {
+	work := event2.EventWorkFunc(func() (interface{}, error) {
+		resp, err := s.dao.CClean(req.Key)
+		if err != nil {
+			return nil, err
+		}
+		if s.aof != nil {
+			s.aof.SendRequest("CClean", req)
+		}
+		return resp, nil
+	})
+
+	lruEvent := s.lruProduce.NewEvent(lru.OptionEventName)
+	lruEvent.InitWaitEvent()
+	lruEvent.SetValue(event2.WorkFuncEventKey, work)
+	s.lruProduce.Call(ctx, lruEvent)
+	resp, err := lruEvent.StartWaitEvent(s.timeOut)
+	lruEvent.Recovery()
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.(*proto.CCleanResponse), nil
+}
+
 func (s *singleService) Set(
 	ctx context.Context,
 	req *proto.SetRequest,
--- a/storage/temp/const.gen.go
+++ b/storage/temp/const.gen.go
@ -10,6 +10,7 @@ const (
 	LIST_X
 	HASH_X
 	SET_X
+	CHANNEL_X
 )

 const (
@ -58,6 +59,11 @@ const (
 	SUnion
 	SUnionStore
 	SScan
+	CPush
+	CPop
+	CMake
+	CLen
+	CClean
 )

 var CommKeyString = map[string]int{
@ -108,6 +114,12 @@ var CommKeyString = map[string]int{
 	"SUnion":      SET_X,
 	"SUnionStore": SET_X,
 	"SScan":       SET_X,
+
+	"CPush":  CHANNEL_X,
+	"CPop":   CHANNEL_X,
+	"CMake":  CHANNEL_X,
+	"CLen":   CHANNEL_X,
+	"CClean": CHANNEL_X,
 }

 var CommKey = map[int]int{
@ -158,4 +170,10 @@ var CommKey = map[int]int{
 	SUnion:      SET_X,
 	SUnionStore: SET_X,
 	SScan:       SET_X,
+
+	CPush:  CHANNEL_X,
+	CPop:   CHANNEL_X,
+	CMake:  CHANNEL_X,
+	CLen:   CHANNEL_X,
+	CClean: CHANNEL_X,
 }
--- a/storage/temp/tem.yaml
+++ b/storage/temp/tem.yaml
@ -51,4 +51,11 @@ SET_X:
 - s_rem
 - s_union
 - s_union_store
- - s_scan
+ - s_scan
+
+CHANNEL_X:
+ - c_push
+ - c_pop
+ - c_make
+ - c_len
+ - c_clean