feat: add in-memory sequencer (#398)

Author: stulzq

cmd/layotto/main.go

@@ -115,6 +115,7 @@ import (
 
 	// Sequencer
 	sequencer_etcd "mosn.io/layotto/components/sequencer/etcd"
+	sequencer_inmemory "mosn.io/layotto/components/sequencer/in-memory"
 	sequencer_mongo "mosn.io/layotto/components/sequencer/mongo"
 	sequencer_redis "mosn.io/layotto/components/sequencer/redis"
 	sequencer_zookeeper "mosn.io/layotto/components/sequencer/zookeeper"
@@ -375,6 +376,9 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 			runtime_sequencer.NewFactory("mongo", func() sequencer.Store {
 				return sequencer_mongo.NewMongoSequencer(log.DefaultLogger)
 			}),
+			runtime_sequencer.NewFactory("in-memory", func() sequencer.Store {
+				return sequencer_inmemory.NewInMemorySequencer()
+			}),
 		),
 		// secretstores
 		runtime.WithSecretStoresFactory(

cmd/layotto_multiple_api/main.go

@@ -108,6 +108,7 @@ import (
 
 	// Sequencer
 	sequencer_etcd "mosn.io/layotto/components/sequencer/etcd"
+	sequencer_inmemory "mosn.io/layotto/components/sequencer/in-memory"
 	sequencer_redis "mosn.io/layotto/components/sequencer/redis"
 	sequencer_zookeeper "mosn.io/layotto/components/sequencer/zookeeper"
 
@@ -366,6 +367,9 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 			runtime_sequencer.NewFactory("zookeeper", func() sequencer.Store {
 				return sequencer_zookeeper.NewZookeeperSequencer(log.DefaultLogger)
 			}),
+			runtime_sequencer.NewFactory("in-memory", func() sequencer.Store {
+				return sequencer_inmemory.NewInMemorySequencer()
+			}),
 		))
 	return server, err
 }

components/go.mod

@@ -32,6 +32,7 @@ require (
 	go.etcd.io/etcd/client/v3 v3.5.0
 	go.etcd.io/etcd/server/v3 v3.5.0
 	go.mongodb.org/mongo-driver v1.8.0
+	go.uber.org/atomic v1.7.0
 	golang.org/x/net v0.0.0-20210614182718-04defd469f4e // indirect
 	golang.org/x/oauth2 v0.0.0-20201208152858-08078c50e5b5 // indirect
 	google.golang.org/grpc v1.38.0

components/sequencer/in-memory/in_memory_sequencer.go

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2021 Layotto Authors
+ *
+ * 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 in_memory
+
+import (
+	"go.uber.org/atomic"
+	"mosn.io/layotto/components/sequencer"
+	"sync"
+)
+
+type InMemorySequencer struct {
+	data *sync.Map
+}
+
+func NewInMemorySequencer() *InMemorySequencer {
+	return &InMemorySequencer{
+		data: &sync.Map{},
+	}
+}
+
+func (s *InMemorySequencer) Init(_ sequencer.Configuration) error {
+	return nil
+}
+
+func (s *InMemorySequencer) GetNextId(req *sequencer.GetNextIdRequest) (*sequencer.GetNextIdResponse, error) {
+	seed, ok := s.data.Load(req.Key)
+	if !ok {
+		seed, _ = s.data.LoadOrStore(req.Key, &atomic.Int64{})
+	}
+
+	nextId := seed.(*atomic.Int64).Inc()
+	return &sequencer.GetNextIdResponse{NextId: nextId}, nil
+
+}
+
+func (s *InMemorySequencer) GetSegment(req *sequencer.GetSegmentRequest) (bool, *sequencer.GetSegmentResponse, error) {
+	seed, ok := s.data.Load(req.Key)
+	if !ok {
+		seed, _ = s.data.LoadOrStore(req.Key, &atomic.Int64{})
+	}
+
+	res := seed.(*atomic.Int64).Add(int64(req.Size))
+	return true, &sequencer.GetSegmentResponse{
+		From: res - int64(req.Size) + 1,
+		To:   res,
+	}, nil
+}

components/sequencer/in-memory/in_memory_sequencer_test.go

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2021 Layotto Authors
+ *
+ * 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 in_memory
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"mosn.io/layotto/components/sequencer"
+)
+
+func TestNew(t *testing.T) {
+	s := NewInMemorySequencer()
+
+	assert.NotNil(t, s)
+}
+
+func TestInit(t *testing.T) {
+	s := NewInMemorySequencer()
+	assert.NotNil(t, s)
+
+	err := s.Init(sequencer.Configuration{})
+	assert.NoError(t, err)
+}
+
+func TestGetNextId(t *testing.T) {
+	s := NewInMemorySequencer()
+	assert.NotNil(t, s)
+
+	err := s.Init(sequencer.Configuration{})
+	assert.NoError(t, err)
+
+	var resp *sequencer.GetNextIdResponse
+	resp, err = s.GetNextId(&sequencer.GetNextIdRequest{Key: "666"})
+	assert.NoError(t, err)
+	assert.Equal(t, int64(1), resp.NextId)
+
+	resp, err = s.GetNextId(&sequencer.GetNextIdRequest{Key: "666"})
+	assert.NoError(t, err)
+	assert.Equal(t, int64(2), resp.NextId)
+
+	resp, err = s.GetNextId(&sequencer.GetNextIdRequest{Key: "777"})
+	assert.NoError(t, err)
+	assert.Equal(t, int64(1), resp.NextId)
+}
+
+func TestGetSegment(t *testing.T) {
+	s := NewInMemorySequencer()
+	assert.NotNil(t, s)
+
+	err := s.Init(sequencer.Configuration{})
+	assert.NoError(t, err)
+
+	var resp *sequencer.GetSegmentResponse
+	var res bool
+	res, resp, err = s.GetSegment(&sequencer.GetSegmentRequest{Key: "666", Size: 5})
+	assert.NoError(t, err)
+	assert.Equal(t, int64(1), resp.From)
+	assert.Equal(t, int64(5), resp.To)
+	assert.True(t, res)
+
+	res, resp, err = s.GetSegment(&sequencer.GetSegmentRequest{Key: "666", Size: 5})
+	assert.NoError(t, err)
+	assert.Equal(t, int64(6), resp.From)
+	assert.Equal(t, int64(10), resp.To)
+	assert.True(t, res)
+
+	res, resp, err = s.GetSegment(&sequencer.GetSegmentRequest{Key: "777", Size: 5})
+	assert.NoError(t, err)
+	assert.Equal(t, int64(1), resp.From)
+	assert.Equal(t, int64(5), resp.To)
+	assert.True(t, res)
+
+}

configs/config_in_memory.json

@@ -39,6 +39,11 @@
                           }
                         }
                       },
+                      "sequencer": {
+                        "in-memory": {
+                          "metadata": {}
+                        }
+                      },
                       "app": {
                         "app_id": "app1",
                         "grpc_callback_port": 9999

demo/sequencer/in-memory/client.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	client "mosn.io/layotto/sdk/go-sdk/client"
+	runtimev1pb "mosn.io/layotto/spec/proto/runtime/v1"
+)
+
+const (
+	key       = "key666"
+	storeName = "in-memory"
+)
+
+func main() {
+
+	cli, err := client.NewClient()
+	if err != nil {
+		panic(err)
+	}
+	defer cli.Close()
+	ctx := context.Background()
+	fmt.Printf("Try to get next id.Key:%s \n", key)
+	for i := 0; i < 10; i++ {
+		id, err := cli.GetNextId(ctx, &runtimev1pb.GetNextIdRequest{
+			StoreName: storeName,
+			Key:       key,
+			Options:   nil,
+			Metadata:  nil,
+		})
+		if err != nil {
+			fmt.Print(err)
+			return
+		}
+		fmt.Printf("Next id:%v \n", id)
+	}
+	fmt.Println("Demo success!")
+}

docs/zh/component_specs/sequencer/in-memory.md

@@ -0,0 +1,25 @@
+# In-Memory
+
+## 配置项说明
+
+直接使用配置：configs/config_in_memory.json
+
+
+## 启动 layotto
+
+````shell
+cd ${projectpath}/cmd/layotto
+go build
+````
+编译成功后执行:
+````shell
+./layotto start -c ../../configs/config_in_memory.json
+````
+
+## 运行 Demo
+
+````shell
+cd ${projectpath}/demo/sequencer/in-memory/
+ go build -o client
+ ./client
+````