feat: make parameters of API plugin extendable (#369)

Author: seeflood

.github/workflows/github-actions.yml

@@ -15,7 +15,7 @@ jobs:
       # If you want to matrix build , you can append the following list.
       matrix:
         go_version:
-          - 1.15
+          - 1.14.13
         os:
           - ubuntu-latest
 
@@ -69,6 +69,17 @@ jobs:
       - name: Post Coverage
         run: bash <(curl -s https://codecov.io/bash)
 
-      - name: integrate
-        run: make wasm-integrate
+  integrate:
+    name: integrate
+    runs-on: ubuntu-latest
+    steps:
+      - name: Set up Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: 1.14.13
 
+      - name: Check out code
+        uses: actions/checkout@v2
+
+      - name: Run Integrate tests.
+        run: make wasm-integrate

pkg/integrate/api/helloworld/grpc_api.go renamed to cmd/layotto_multiple_api/helloworld/grpc_api.go

@@ -18,33 +18,14 @@ package helloworld
 
 import (
 	"context"
-	"github.com/dapr/components-contrib/bindings"
-	"github.com/dapr/components-contrib/pubsub"
-	"github.com/dapr/components-contrib/state"
 	rawGRPC "google.golang.org/grpc"
 	pb "google.golang.org/grpc/examples/helloworld/helloworld"
-	"mosn.io/layotto/components/configstores"
-	"mosn.io/layotto/components/file"
-	"mosn.io/layotto/components/hello"
-	"mosn.io/layotto/components/lock"
-	"mosn.io/layotto/components/rpc"
-	"mosn.io/layotto/components/sequencer"
 	"mosn.io/layotto/pkg/grpc"
+	grpc_api "mosn.io/layotto/pkg/grpc"
 	mgrpc "mosn.io/mosn/pkg/filter/network/grpc"
 )
 
-func NewHelloWorldAPI(
-	appId string,
-	hellos map[string]hello.HelloService,
-	configStores map[string]configstores.Store,
-	rpcs map[string]rpc.Invoker,
-	pubSubs map[string]pubsub.PubSub,
-	stateStores map[string]state.Store,
-	files map[string]file.File,
-	lockStores map[string]lock.LockStore,
-	sequencers map[string]sequencer.Store,
-	sendToOutputBindingFn func(name string, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error),
-) grpc.GrpcAPI {
+func NewHelloWorldAPI(ac *grpc_api.ApplicationContext) grpc.GrpcAPI {
 	return &server{}
 }
 

cmd/layotto_multiple_api/main.go

@@ -19,15 +19,13 @@ package main
 import (
 	"encoding/json"
 	"fmt"
-	"os"
-	"strconv"
-	"time"
-
 	"mosn.io/api"
-	"mosn.io/layotto/diagnostics"
+	helloworld_api "mosn.io/layotto/cmd/layotto_multiple_api/helloworld"
 	"mosn.io/layotto/pkg/grpc/dapr"
 	"mosn.io/layotto/pkg/grpc/default_api"
-	helloworld_api "mosn.io/layotto/pkg/integrate/api/helloworld"
+	"os"
+	"strconv"
+	"time"
 
 	mock_state "mosn.io/layotto/pkg/mock/components/state"
 	_ "mosn.io/layotto/pkg/wasm"
@@ -121,6 +119,7 @@ import (
 
 	"github.com/urfave/cli"
 	"google.golang.org/grpc"
+	"mosn.io/layotto/diagnostics"
 	_ "mosn.io/layotto/pkg/filter/network/tcpcopy"
 	"mosn.io/layotto/pkg/runtime"
 	"mosn.io/mosn/pkg/featuregate"

codecov.yml

@@ -12,7 +12,7 @@ coverage:
 
 ignore:
   # Configure what to ignore.
-  - "pkg/mock"         # - Testing mock.
+  - "pkg/mock/**/*"         # - Testing mock.
   # Ignore non-deterministic methods and tests
   # codecov might report unexpected changes even if no code has been changed.
   # see https://community.codecov.com/t/reported-change-in-coverage-when-no-code-changes-made/1346
@@ -24,4 +24,5 @@ ignore:
   - "pkg/grpc/dapr/proto/runtime/v1"
   - "pkg/grpc/dapr/dapr_api_unimplement.go"
   - "pkg/wasm/watcher.go"
+  - "cmd/**/*"
   - "components/lock/consul/consul_lock_task.go"

components/rpc/callback/callback.go

@@ -57,7 +57,7 @@ var (
 	// to storage BeforeFactory
 	beforeInvokeRegistry = map[string]BeforeFactory{}
 	// to storage AfterFactory
-	afterInvokeRegistry  = map[string]AfterFactory{}
+	afterInvokeRegistry = map[string]AfterFactory{}
 )
 
 // NewCallback is created Callback
@@ -84,7 +84,7 @@ func (c *callback) AddBeforeInvoke(conf rpc.CallbackFunc) {
 	c.beforeInvoke = append(c.beforeInvoke, f.Create())
 }
 
-// AddAfterInvoke is add beforeInvoke into callback.afterInvoke
+// AddAfterInvoke is used to add beforeInvoke into callback.afterInvoke
 func (c *callback) AddAfterInvoke(conf rpc.CallbackFunc) {
 	f, ok := afterInvokeRegistry[conf.Name]
 	if !ok {
@@ -98,7 +98,7 @@ func (c *callback) AddAfterInvoke(conf rpc.CallbackFunc) {
 	c.afterInvoke = append(c.afterInvoke, f.Create())
 }
 
-// BeforeInvoke is get RPCRequest in callback.beforeInvoke
+// BeforeInvoke is used to invoke beforeInvoke callbacks
 func (c *callback) BeforeInvoke(request *rpc.RPCRequest) (*rpc.RPCRequest, error) {
 	var err error
 	for _, cb := range c.beforeInvoke {
@@ -109,7 +109,7 @@ func (c *callback) BeforeInvoke(request *rpc.RPCRequest) (*rpc.RPCRequest, error
 	return request, err
 }
 
-// AfterInvoke is get RPCResponse in callback.afterInvoke
+// AfterInvoke is used to invoke afterInvoke callbacks
 func (c *callback) AfterInvoke(response *rpc.RPCResponse) (*rpc.RPCResponse, error) {
 	var err error
 	for _, cb := range c.afterInvoke {

components/rpc/invoker/mosn/channel/connpool.go

@@ -88,7 +88,6 @@ func newConnPool(
 	return p
 }
 
-
 // connPool is connected pool
 type connPool struct {
 	maxActive   int
@@ -131,6 +130,7 @@ func (p *connPool) Get(ctx context.Context) (*wrapConn, error) {
 	if p.stateFunc != nil {
 		wc.state = p.stateFunc()
 	}
+	// start a readloop gorountine to read and handle data
 	if p.onDataFunc != nil {
 		utils.GoWithRecover(func() {
 			p.readloop(wc)
@@ -171,6 +171,7 @@ func (p *connPool) readloop(c *wrapConn) {
 
 	c.buf = buffer.NewIoBuffer(defaultBufSize)
 	for {
+		// read data from connection
 		n, readErr := c.buf.ReadOnce(c)
 		if readErr != nil {
 			err = readErr
@@ -182,6 +183,8 @@ func (p *connPool) readloop(c *wrapConn) {
 		}
 
 		if n > 0 {
+			// handle data.
+			// it will delegate to hstate if it's constructed by httpchannel
 			if onDataErr := p.onDataFunc(c); onDataErr != nil {
 				err = onDataErr
 				log.DefaultLogger.Errorf("[runtime][rpc]connpool onData err: %s", onDataErr.Error())

components/rpc/invoker/mosn/channel/httpchannel.go

@@ -36,8 +36,11 @@ func init() {
 	RegistChannel("http", newHttpChannel)
 }
 
+// hstate is a pipe for readloop goroutine to communicate with request goroutine
 type hstate struct {
+	// request goroutine will read data from it
 	reader net.Conn
+	// readloop goroutine will write data to it
 	writer net.Conn
 }
 
@@ -60,20 +63,30 @@ type httpChannel struct {
 	pool *connPool
 }
 
-// newHttpChannel is create rpc.Channel by ChannelConfig
+// newHttpChannel is used to create rpc.Channel according to ChannelConfig
 func newHttpChannel(config ChannelConfig) (rpc.Channel, error) {
 	hc := &httpChannel{}
 	hc.pool = newConnPool(
 		config.Size,
+		// dialFunc
 		func() (net.Conn, error) {
 			local, remote := net.Pipe()
 			localTcpConn := &fakeTcpConn{c: local}
 			remoteTcpConn := &fakeTcpConn{c: remote}
 			if err := acceptFunc(remoteTcpConn, config.Listener); err != nil {
 				return nil, err
 			}
+			// the goroutine model is:
+			// request goroutine --->  localTcpConn ---> mosn
+			//		^										|
+			//		|										|
+			//		|										|
+			// 		hstate <-- readloop goroutine 	<------
 			return localTcpConn, nil
-		}, func() interface{} {
+		},
+		// stateFunc
+		func() interface{} {
+			// hstate is a pipe for readloop goroutine to communicate with request goroutine
 			s := &hstate{}
 			s.reader, s.writer = net.Pipe()
 			return s
@@ -84,25 +97,29 @@ func newHttpChannel(config ChannelConfig) (rpc.Channel, error) {
 	return hc, nil
 }
 
-// Do is handle RPCRequest to RPCResponse
+// Do is used to handle RPCRequest and return RPCResponse
 func (h *httpChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
+	// 1. context.WithTimeout
 	timeout := time.Duration(req.Timeout) * time.Millisecond
 	ctx, cancel := context.WithTimeout(req.Ctx, timeout)
 	defer cancel()
 
+	// 2. get a fake connection with mosn
+	// The pool will start a readloop gorountine,
+	// which aims to read data from mosn and then write data to the hstate.writer
 	conn, err := h.pool.Get(ctx)
 	if err != nil {
 		return nil, err
 	}
-
+	// 3. set deadline before write data to this connection
 	hstate := conn.state.(*hstate)
 	deadline, _ := ctx.Deadline()
 	if err = conn.SetWriteDeadline(deadline); err != nil {
 		hstate.close()
 		h.pool.Put(conn, true)
 		return nil, common.Error(common.UnavailebleCode, err.Error())
 	}
-
+	// 4. write data to this fake connection
 	httpReq := h.constructReq(req)
 	defer fasthttp.ReleaseRequest(httpReq)
 
@@ -112,6 +129,7 @@ func (h *httpChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
 		return nil, common.Error(common.UnavailebleCode, err.Error())
 	}
 
+	// 5. read response data and parse it into fasthttp.Response
 	httpResp := &fasthttp.Response{}
 	hstate.reader.SetReadDeadline(deadline)
 
@@ -121,6 +139,8 @@ func (h *httpChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
 		return nil, common.Error(common.UnavailebleCode, err.Error())
 	}
 	h.pool.Put(conn, false)
+
+	// 6. convert result to rpc.RPCResponse,which is the response of rpc invoker
 	body := httpResp.Body()
 	if httpResp.StatusCode() != http.StatusOK {
 		return nil, common.Errorf(common.UnavailebleCode, "http response code %d, body: %s", httpResp.StatusCode(), string(body))

components/rpc/invoker/mosn/channel/xchannel.go

@@ -51,6 +51,7 @@ func newXChannel(config ChannelConfig) (rpc.Channel, error) {
 	m := &xChannel{proto: proto}
 	m.pool = newConnPool(
 		config.Size,
+		// dialFunc
 		func() (net.Conn, error) {
 			local, remote := net.Pipe()
 			localTcpConn := &fakeTcpConn{c: local}
@@ -60,6 +61,7 @@ func newXChannel(config ChannelConfig) (rpc.Channel, error) {
 			}
 			return localTcpConn, nil
 		},
+		// stateFunc
 		func() interface{} {
 			return &xstate{calls: map[uint32]chan call{}}
 		},
@@ -90,18 +92,20 @@ type xChannel struct {
 
 // Do is handle RPCRequest to RPCResponse
 func (m *xChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
+	// 1. context.WithTimeout
 	timeout := time.Duration(req.Timeout) * time.Millisecond
 	ctx, cancel := context.WithTimeout(req.Ctx, timeout)
 	defer cancel()
 
+	// 2. get fake connection with mosn
 	conn, err := m.pool.Get(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	xstate := conn.state.(*xstate)
 
-	// encode request
+	// 3. encode request
 	frame := m.proto.ToFrame(req)
 	id := atomic.AddUint32(&xstate.reqid, 1)
 	frame.SetRequestId(uint64(id))
@@ -131,6 +135,7 @@ func (m *xChannel) Do(req *rpc.RPCRequest) (*rpc.RPCResponse, error) {
 	}
 	m.pool.Put(conn, false)
 
+	// read response and decode it
 	select {
 	case res := <-callChan:
 		if res.err != nil {

components/rpc/invoker/mosn/mosninvoker.go

@@ -89,24 +89,26 @@ func (m *mosnInvoker) Invoke(ctx context.Context, req *rpc.RPCRequest) (resp *rp
 		}
 	}()
 
+	// 1. validate request
 	if req.Timeout == 0 {
 		req.Timeout = 3000
 	}
 	req.Ctx = ctx
 	log.DefaultLogger.Debugf("[runtime][rpc]request %+v", req)
+	// 2. beforeInvoke callback
 	req, err = m.cb.BeforeInvoke(req)
 	if err != nil {
 		log.DefaultLogger.Errorf("[runtime][rpc]before filter error %s", err.Error())
 		return nil, err
 	}
-
+	// 3. do invocation
 	resp, err = m.channel.Do(req)
 	if err != nil {
 		log.DefaultLogger.Errorf("[runtime][rpc]error %s", err.Error())
 		return nil, err
 	}
-
 	resp.Ctx = req.Ctx
+	// 4. afterInvoke callback
 	resp, err = m.cb.AfterInvoke(resp)
 	if err != nil {
 		log.DefaultLogger.Errorf("[runtime][rpc]after filter error %s", err.Error())

components/rpc/types.go

@@ -53,7 +53,7 @@ func (r RPCHeader) Get(key string) string {
 // RPCRequest is request info
 type RPCRequest struct {
 	// context
-	Ctx         context.Context
+	Ctx context.Context
 	// request id
 	Id          string
 	Timeout     int32
@@ -88,9 +88,9 @@ type Callback interface {
 	// AddAfterInvoke is add AfterInvoke func
 	AddAfterInvoke(CallbackFunc)
 
-	// BeforeInvoke is get BeforeInvoke by RPCRequest
+	// BeforeInvoke is used to invoke beforeInvoke callbacks
 	BeforeInvoke(*RPCRequest) (*RPCRequest, error)
-	// AfterInvoke is get AfterInvoke by RPCRequest
+	// AfterInvoke is used to invoke afterInvoke callbacks
 	AfterInvoke(*RPCResponse) (*RPCResponse, error)
 }
 

pkg/actuator/info/endpoint.go

@@ -61,5 +61,8 @@ func AddInfoContributor(name string, c Contributor) {
 
 // AddInfoContributorFunc register info.Contributor.It's not concurrent-safe,so please invoke it ONLY in init method.
 func AddInfoContributorFunc(name string, f func() (interface{}, error)) {
+	if f == nil {
+		return
+	}
 	AddInfoContributor(name, ContributorAdapter(f))
 }

pkg/actuator/info/endpoint_test.go

@@ -38,6 +38,11 @@ func TestEndpoint_Handle(t *testing.T) {
 	assert.True(t, err == nil)
 	assert.True(t, len(handle) == 0)
 
+	AddInfoContributorFunc("test", nil)
+	handle, err = ep.Handle(context.Background(), nil)
+	assert.True(t, err == nil)
+	assert.True(t, len(handle) == 0)
+
 	AddInfoContributor("test", nil)
 	handle, err = ep.Handle(context.Background(), nil)
 	assert.True(t, err == nil)