feat: package module (#188)

Author: RobertIndie

.github/workflows/bench.yaml

@@ -27,15 +27,15 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [ '1.21' ]
+        go-version: [ '1.22' ]
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v3
         with:
           go-version: ${{ matrix.go-version }}
       - uses: acifani/setup-tinygo@v2
         with:
-          tinygo-version: '0.30.0'
+          tinygo-version: '0.31.2'
       - run: docker compose -f ./tests/docker-compose.yaml up -d
       - run: make build-all
       - name: Wait for Pulsar service
@@ -45,7 +45,7 @@ jobs:
         if: failure()
         run: docker compose -f ./tests/docker-compose.yaml logs || true
       - name: Upload artifacts
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         with:
           name: BenchmarkStressForBasicFunc Profile
           path: ./benchmark/*.pprof

.github/workflows/ci.yaml

@@ -27,15 +27,15 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [ '1.21' ]
+        go-version: [ '1.22' ]
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v3
         with:
           go-version: ${{ matrix.go-version }}
       - uses: acifani/setup-tinygo@v2
         with:
-          tinygo-version: '0.30.0'
+          tinygo-version: '0.31.2'
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3
         with:

clients/gofs/gofs.go

@@ -1,6 +1,3 @@
-//go:build wasi
-// +build wasi
-
 /*
  * Copyright 2024 Function Stream Org.
  *
@@ -19,55 +16,220 @@
 
 package gofs
 
-import "C"
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/wirelessr/avroschema"
 	"os"
-	"syscall"
+	"sync"
+	"time"
+
+	"github.com/wirelessr/avroschema"
+)
+
+const (
+	StateInit int32 = iota
+	StateRunning
+)
+
+const (
+	FSSocketPath   = "FS_SOCKET_PATH"
+	FSFunctionName = "FS_FUNCTION_NAME"
+	FSModuleName   = "FS_MODULE_NAME"
+	DefaultModule  = "default"
 )
 
-var processFd int
-var registerSchemaFd int
+var (
+	ErrRegisterModuleDuringRunning = fmt.Errorf("cannot register module during running")
+	ErrAlreadyRunning              = fmt.Errorf("already running")
+)
 
-func init() {
-	processFd, _ = syscall.Open("/process", syscall.O_RDWR, 0)
-	registerSchemaFd, _ = syscall.Open("/registerSchema", syscall.O_RDWR, 0)
+type FSClient interface {
+	error
+	Register(module string, wrapper *moduleWrapper) FSClient
+	Run() error
 }
 
-var processFunc func([]byte) []byte
+type fsClient struct {
+	rpc        *fsRPCClient
+	modules    map[string]*moduleWrapper
+	state      int32
+	registerMu sync.Mutex
+	err        error
+}
 
-func Register[I any, O any](process func(*I) *O) error {
-	outputSchema, err := avroschema.Reflect(new(O))
-	if err != nil {
-		return err
+func NewFSClient() FSClient {
+	return &fsClient{
+		modules: make(map[string]*moduleWrapper),
+		state:   StateInit,
 	}
-	processFunc = func(payload []byte) []byte {
+}
+
+type moduleWrapper struct {
+	*fsClient
+	processFunc func([]byte) []byte // Only for Function
+	executeFunc func() error
+	initFunc    func() error
+	registerErr error
+}
+
+func (c *fsClient) Register(module string, wrapper *moduleWrapper) FSClient {
+	if c.err != nil {
+		return c
+	}
+	c.registerMu.Lock()
+	defer c.registerMu.Unlock()
+	if c.state == StateRunning {
+		c.err = ErrRegisterModuleDuringRunning
+		return c
+	}
+	if wrapper.registerErr != nil {
+		c.err = wrapper.registerErr
+		return c
+	}
+	c.modules[module] = wrapper
+	return c
+}
+
+func Function[I any, O any](process func(*I) *O) *moduleWrapper {
+	processFunc := func(payload []byte) []byte {
 		input := new(I)
-		err = json.Unmarshal(payload, input)
+		err := json.Unmarshal(payload, input)
 		if err != nil {
-			fmt.Fprintf(os.Stderr, "Failed to parse JSON: %s %s", err, payload)
-			return nil
+			_, _ = fmt.Fprintf(os.Stderr, "Failed to parse JSON: %s %s\n", err, payload)
 		}
 		output := process(input)
 		outputPayload, _ := json.Marshal(output)
 		return outputPayload
 	}
-	syscall.Write(registerSchemaFd, []byte(outputSchema))
-	return nil
+	m := &moduleWrapper{}
+	m.initFunc = func() error {
+		outputSchema, err := avroschema.Reflect(new(O))
+		if err != nil {
+			return err
+		}
+		err = m.rpc.RegisterSchema(outputSchema)
+		if err != nil {
+			return fmt.Errorf("failed to register schema: %w", err)
+		}
+		return nil
+	}
+	m.executeFunc = func() error {
+		for {
+			inputPayload, err := m.rpc.Read()
+			if err != nil {
+				_, _ = fmt.Fprintf(os.Stderr, "Failed to read: %s\n", err)
+				time.Sleep(3 * time.Second)
+				continue
+			}
+			outputPayload := processFunc(inputPayload)
+			err = m.rpc.Write(outputPayload)
+			if err != nil {
+				_, _ = fmt.Fprintf(os.Stderr, "Failed to write: %s\n", err)
+			}
+		}
+	}
+	m.processFunc = processFunc
+	return m
+}
+
+func Source[O any](process func(emit func(*O) error)) *moduleWrapper {
+	m := &moduleWrapper{}
+	emit := func(event *O) error {
+		outputPayload, _ := json.Marshal(event)
+		return m.rpc.Write(outputPayload)
+	}
+	m.initFunc = func() error {
+		outputSchema, err := avroschema.Reflect(new(O))
+		if err != nil {
+			return err
+		}
+		err = m.rpc.RegisterSchema(outputSchema)
+		if err != nil {
+			return fmt.Errorf("failed to register schema: %w", err)
+		}
+		return nil
+	}
+	m.executeFunc = func() error {
+		process(emit)
+		return nil
+	}
+	return m
 }
 
-//export process
-func process() {
-	var stat syscall.Stat_t
-	syscall.Fstat(processFd, &stat)
-	payload := make([]byte, stat.Size)
-	_, _ = syscall.Read(processFd, payload)
-	outputPayload := processFunc(payload)
-	_, _ = syscall.Write(processFd, outputPayload)
+func Sink[I any](process func(*I)) *moduleWrapper {
+	processFunc := func(payload []byte) {
+		input := new(I)
+		err := json.Unmarshal(payload, input)
+		if err != nil {
+			_, _ = fmt.Fprintf(os.Stderr, "Failed to parse JSON: %s %s\n", err, payload)
+		}
+		process(input)
+	}
+	m := &moduleWrapper{}
+	m.initFunc = func() error {
+		inputSchema, err := avroschema.Reflect(new(I))
+		if err != nil {
+			return err
+		}
+		err = m.rpc.RegisterSchema(inputSchema)
+		if err != nil {
+			return fmt.Errorf("failed to register schema: %w", err)
+		}
+		return nil
+	}
+	m.executeFunc = func() error {
+		for {
+			inputPayload, err := m.rpc.Read()
+			if err != nil {
+				_, _ = fmt.Fprintf(os.Stderr, "Failed to read: %s\n", err)
+				time.Sleep(3 * time.Second)
+				continue
+			}
+			processFunc(inputPayload)
+		}
+	}
+	return m
+}
+
+func (c *fsClient) Run() error {
+	if c.err != nil {
+		return c.err
+	}
+	c.registerMu.Lock()
+	if c.state == StateRunning {
+		c.registerMu.Unlock()
+		return ErrAlreadyRunning
+	}
+	c.state = StateRunning
+	c.registerMu.Unlock()
+
+	if c.rpc == nil {
+		rpc, err := newFSRPCClient()
+		if err != nil {
+			return err
+		}
+		c.rpc = rpc
+	}
+	module := os.Getenv(FSModuleName)
+	if module == "" {
+		module = DefaultModule
+	}
+	m, ok := c.modules[module]
+	if !ok {
+		return fmt.Errorf("module %s not found", module)
+	}
+	m.fsClient = c
+	err := m.initFunc()
+	if err != nil {
+		return err
+	}
+	c.rpc.loadModule(m)
+	if c.rpc.skipExecuting() {
+		return nil
+	}
+	return m.executeFunc()
 }
 
-func Run() {
-	// Leave it empty
+func (c *fsClient) Error() string {
+	return c.err.Error()
 }