feat: support pluggable state store (#173)

Makefile

@@ -20,6 +20,9 @@ build_example:
 lint:
 	golangci-lint run
 
+lint-fix:
+	golangci-lint run --fix
+
 build_all: build build_example
 
 test:

admin/utils/utils.go

@@ -18,7 +18,8 @@ package utils
 
 import (
 	"fmt"
-	"github.com/functionstream/function-stream/admin/client"
+
+	adminclient "github.com/functionstream/function-stream/admin/client"
 )
 
 func MakeQueueSourceTubeConfig(subName string, topics ...string) []adminclient.ModelTubeConfig {

benchmark/bench_test.go

@@ -18,21 +18,22 @@ package benchmark
 
 import (
 	"context"
+	"math/rand"
+	"os"
+	"runtime/pprof"
+	"strconv"
+	"testing"
+	"time"
+
 	"github.com/apache/pulsar-client-go/pulsaradmin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
-	"github.com/functionstream/function-stream/admin/client"
+	adminclient "github.com/functionstream/function-stream/admin/client"
 	adminutils "github.com/functionstream/function-stream/admin/utils"
 	"github.com/functionstream/function-stream/common"
 	"github.com/functionstream/function-stream/fs"
 	"github.com/functionstream/function-stream/fs/contube"
 	"github.com/functionstream/function-stream/perf"
 	"github.com/functionstream/function-stream/server"
-	"math/rand"
-	"os"
-	"runtime/pprof"
-	"strconv"
-	"testing"
-	"time"
 )
 
 func BenchmarkStressForBasicFunc(b *testing.B) {

cmd/client/cmd.go

@@ -35,7 +35,8 @@ var (
 )
 
 func init() {
-	Cmd.PersistentFlags().StringVarP(&c.Config.ServiceAddr, "service-address", "s", "http://localhost:7300", "Service address")
+	Cmd.PersistentFlags().StringVarP(&c.Config.ServiceAddr, "service-address", "s",
+		"http://localhost:7300", "Service address")
 
 	Cmd.AddCommand(create.Cmd)
 	Cmd.AddCommand(list.Cmd)

cmd/client/consume/cmd.go

@@ -18,11 +18,12 @@ package consume
 
 import (
 	"fmt"
-	"github.com/functionstream/function-stream/admin/client"
+	"os"
+
+	adminclient "github.com/functionstream/function-stream/admin/client"
 	"github.com/functionstream/function-stream/cmd/client/common"
 	"github.com/spf13/cobra"
 	"golang.org/x/net/context"
-	"os"
 )
 
 var (

cmd/client/create/cmd.go

@@ -19,13 +19,14 @@ package create
 import (
 	"context"
 	"fmt"
-	"github.com/functionstream/function-stream/admin/client"
+	"io"
+	"os"
+
+	adminclient "github.com/functionstream/function-stream/admin/client"
 	"github.com/functionstream/function-stream/admin/utils"
 	"github.com/functionstream/function-stream/cmd/client/common"
 	fs_cmmon "github.com/functionstream/function-stream/common"
 	"github.com/spf13/cobra"
-	"io"
-	"os"
 )
 
 var (

cmd/client/delete/cmd.go

@@ -17,7 +17,7 @@
 package del
 
 import (
-	"github.com/functionstream/function-stream/admin/client"
+	adminclient "github.com/functionstream/function-stream/admin/client"
 	"github.com/functionstream/function-stream/cmd/client/common"
 	"github.com/spf13/cobra"
 )

cmd/client/list/cmd.go

@@ -19,10 +19,11 @@ package list
 import (
 	"context"
 	"fmt"
-	"github.com/functionstream/function-stream/admin/client"
+	"os"
+
+	adminclient "github.com/functionstream/function-stream/admin/client"
 	"github.com/functionstream/function-stream/cmd/client/common"
 	"github.com/spf13/cobra"
-	"os"
 )
 
 var Cmd = &cobra.Command{

cmd/client/produce/cmd.go

@@ -18,11 +18,12 @@ package produce
 
 import (
 	"fmt"
-	"github.com/functionstream/function-stream/admin/client"
+	"os"
+
+	adminclient "github.com/functionstream/function-stream/admin/client"
 	"github.com/functionstream/function-stream/cmd/client/common"
 	"github.com/spf13/cobra"
 	"golang.org/x/net/context"
-	"os"
 )
 
 var (

cmd/main.go

@@ -18,12 +18,13 @@ package main
 
 import (
 	"fmt"
+	"os"
+
 	"github.com/functionstream/function-stream/cmd/client"
 	"github.com/functionstream/function-stream/cmd/perf"
 	"github.com/functionstream/function-stream/cmd/server"
 	"github.com/functionstream/function-stream/cmd/standalone"
 	"github.com/spf13/cobra"
-	"os"
 )
 
 var (

cmd/perf/cmd.go

@@ -18,10 +18,11 @@ package perf
 
 import (
 	"context"
+	"io"
+
 	"github.com/functionstream/function-stream/common"
 	"github.com/functionstream/function-stream/perf"
 	"github.com/spf13/cobra"
-	"io"
 )
 
 var (

cmd/server/cmd.go

@@ -18,10 +18,11 @@ package server
 
 import (
 	"context"
+	"io"
+
 	"github.com/functionstream/function-stream/common"
 	"github.com/functionstream/function-stream/server"
 	"github.com/spf13/cobra"
-	"io"
 )
 
 var (
@@ -45,7 +46,8 @@ var (
 func init() {
 	Cmd.Flags().StringVarP(&config.configFile, "config-file", "c", "conf/function-stream.yaml",
 		"path to the config file (default is conf/function-stream.yaml)")
-	Cmd.Flags().BoolVarP(&config.loadConfigFromEnv, "load-config-from-env", "e", false, "load config from env (default is false)")
+	Cmd.Flags().BoolVarP(&config.loadConfigFromEnv, "load-config-from-env", "e", false,
+		"load config from env (default is false)")
 }
 
 func exec(*cobra.Command, []string) {

cmd/standalone/cmd.go

@@ -18,10 +18,11 @@ package standalone
 
 import (
 	"context"
+	"io"
+
 	"github.com/functionstream/function-stream/common"
 	"github.com/functionstream/function-stream/server"
 	"github.com/spf13/cobra"
-	"io"
 )
 
 var (

common/constants.go

@@ -21,9 +21,9 @@ const (
 	MemoryTubeType = "memory"
 	HttpTubeType   = "http"
 
-	DefaultAddr      = "localhost:7300"
-	DefaultPulsarURL = "pulsar://localhost:6650"
-	DefaultTubeType  = PulsarTubeType
+	WASMRuntime = "wasm"
 
 	RuntimeArchiveConfigKey = "archive"
+
+	StateStorePebble = "pebble"
 )

common/model/function.go

@@ -17,9 +17,10 @@
 package model
 
 import (
+	"strings"
+
 	"github.com/functionstream/function-stream/fs/contube"
 	"github.com/pkg/errors"
-	"strings"
 )
 
 type TubeConfig struct {

common/model/function_serde_test.go

@@ -19,10 +19,11 @@ package model
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/functionstream/function-stream/common"
-	"gopkg.in/yaml.v3"
 	"reflect"
 	"testing"
+
+	"github.com/functionstream/function-stream/common"
+	"gopkg.in/yaml.v3"
 )
 
 func TestFunctionSerde(t *testing.T) {
@@ -110,7 +111,8 @@ func TestFunctionSerdeWithNil(t *testing.T) {
 
 	fmt.Println(string(data))
 
-	f.Sources = []*TubeConfig{} // The nil would be expected to be converted to a zero-length array for the YAML serialization
+	f.Sources = []*TubeConfig{} // The nil would be expected to be converted to a zero-length array for the YAML
+	// serialization
 
 	// YAML Deserialization
 	err = yaml.Unmarshal(data, &f2)

fs/api/instance.go

@@ -17,10 +17,11 @@
 package api
 
 import (
+	"log/slog"
+
 	"github.com/functionstream/function-stream/common/model"
 	"github.com/functionstream/function-stream/fs/contube"
 	"golang.org/x/net/context"
-	"log/slog"
 )
 
 type FunctionInstance interface {

fs/contube/contube.go

@@ -19,6 +19,7 @@ package contube
 import (
 	"context"
 	"fmt"
+
 	"github.com/pkg/errors"
 )
 

fs/contube/http.go

@@ -17,13 +17,14 @@
 package contube
 
 import (
-	"github.com/pkg/errors"
-	"golang.org/x/net/context"
 	"io"
 	"log/slog"
 	"net/http"
 	"sync"
 	"sync/atomic"
+
+	"github.com/pkg/errors"
+	"golang.org/x/net/context"
 )
 
 type state int
@@ -133,7 +134,8 @@ func (f *HttpTubeFactory) NewSinkTube(_ context.Context, _ ConfigMap) (chan<- Re
 	return nil, ErrSinkTubeNotImplemented
 }
 
-func (f *HttpTubeFactory) GetHandleFunc(getEndpoint func(r *http.Request) (string, error), logger *slog.Logger) func(http.ResponseWriter, *http.Request) {
+func (f *HttpTubeFactory) GetHandleFunc(getEndpoint func(r *http.Request) (string, error),
+	logger *slog.Logger) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		endpoint, err := getEndpoint(r)
 		if err != nil {

fs/contube/http_test.go

@@ -17,9 +17,10 @@
 package contube
 
 import (
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/net/context"
-	"testing"
 )
 
 func TestHttpTubeHandleRecord(t *testing.T) {

fs/contube/memory_test.go

@@ -35,7 +35,8 @@ func TestMemoryTube(t *testing.T) {
 	var events []Record
 
 	topics := []string{"topic1", "topic2", "topic3"}
-	source, err := memoryQueueFactory.NewSourceTube(ctx, (&SourceQueueConfig{Topics: topics, SubName: "consume-" + strconv.Itoa(rand.Int())}).ToConfigMap())
+	source, err := memoryQueueFactory.NewSourceTube(ctx, (&SourceQueueConfig{Topics: topics,
+		SubName: "consume-" + strconv.Itoa(rand.Int())}).ToConfigMap())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -72,7 +73,8 @@ func TestMemoryTube(t *testing.T) {
 	wg.Wait()
 	cancel()
 
-	// Give enough time to ensure that the goroutine execution within NewSource Tube and NewSinkTube is complete and the released queue is successful.
+	// Give enough time to ensure that the goroutine execution within NewSource Tube and NewSinkTube is complete and
+	// the released queue is successful.
 	time.Sleep(100 * time.Millisecond)
 
 	// assert the memoryQueueFactory.queues is empty.

fs/contube/pulsar.go

@@ -18,10 +18,11 @@ package contube
 
 import (
 	"context"
-	"github.com/apache/pulsar-client-go/pulsar"
-	"github.com/pkg/errors"
 	"log/slog"
 	"sync/atomic"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+	"github.com/pkg/errors"
 )
 
 const (

fs/func_ctx_impl_test.go

@@ -17,8 +17,9 @@
 package fs
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestFuncCtx_NilStore(t *testing.T) {

fs/instance_impl.go

@@ -18,13 +18,14 @@ package fs
 
 import (
 	"context"
+	"log/slog"
+	"reflect"
+
 	"github.com/functionstream/function-stream/common"
 	"github.com/functionstream/function-stream/common/model"
 	"github.com/functionstream/function-stream/fs/api"
 	"github.com/functionstream/function-stream/fs/contube"
 	"github.com/pkg/errors"
-	"log/slog"
-	"reflect"
 )
 
 type FunctionInstanceImpl struct {
@@ -53,7 +54,8 @@ func NewDefaultInstanceFactory() api.FunctionInstanceFactory {
 	return &DefaultInstanceFactory{}
 }
 
-func (f *DefaultInstanceFactory) NewFunctionInstance(definition *model.Function, funcCtx api.FunctionContext, index int32, logger *slog.Logger) api.FunctionInstance {
+func (f *DefaultInstanceFactory) NewFunctionInstance(definition *model.Function, funcCtx api.FunctionContext,
+	index int32, logger *slog.Logger) api.FunctionInstance {
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	ctx = context.WithValue(ctx, CtxKeyFunctionName, definition.Name)
 	ctx = context.WithValue(ctx, CtxKeyInstanceIndex, index)
@@ -69,7 +71,8 @@ func (f *DefaultInstanceFactory) NewFunctionInstance(definition *model.Function,
 	}
 }
 
-func (instance *FunctionInstanceImpl) Run(runtimeFactory api.FunctionRuntimeFactory, sources []<-chan contube.Record, sink chan<- contube.Record) {
+func (instance *FunctionInstanceImpl) Run(runtimeFactory api.FunctionRuntimeFactory, sources []<-chan contube.Record,
+	sink chan<- contube.Record) {
 	runtime, err := runtimeFactory.NewFunctionRuntime(instance)
 	if err != nil {
 		instance.readyCh <- errors.Wrap(err, "Error creating runtime")