feat: Add startup hooks

cmd/layotto/main.go

@@ -197,6 +197,7 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 	}
 	// 2. new instance
 	rt := runtime.NewMosnRuntime(cfg)
+	rt.AppendInitRuntimeStage(runtime.DefaultInitRuntimeStage)
 	// 3. run
 	server, err := rt.Run(
 		runtime.WithGrpcOptions(opts...),

cmd/layotto_multiple_api/helloworld/grpc_api.go

@@ -38,9 +38,9 @@ func (s *server) Init(conn *rawGRPC.ClientConn) error {
 	return nil
 }
 
-func (s *server) Register(grpcServer *rawGRPC.Server, registeredServer mgrpc.RegisteredServer) (mgrpc.RegisteredServer, error) {
+func (s *server) Register(rawGrpcServer *rawGRPC.Server) error {
 	pb.RegisterGreeterServer(grpcServer, s)
-	return registeredServer, nil
+	return nil
 }
 
 // SayHello implements helloworld.GreeterServer

cmd/layotto_multiple_api/main.go

@@ -188,6 +188,7 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 	}
 	// 2. new instance
 	rt := runtime.NewMosnRuntime(cfg)
+	rt.AppendInitRuntimeStage(runtime.DefaultInitRuntimeStage)
 	// 3. run
 	server, err := rt.Run(
 		runtime.WithGrpcOptions(opts...),

components/custom/component.go

@@ -0,0 +1,20 @@
+//
+// 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 custom
+
+import "context"
+
+type Component interface {
+	Initialize(ctx context.Context, config Config) error
+}

components/custom/config.go

@@ -0,0 +1,19 @@
+//
+// 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 custom
+
+type Config struct {
+	Version  string            `json:"version"`
+	Metadata map[string]string `json:"metadata"`
+}

components/custom/registry.go

@@ -0,0 +1,76 @@
+//
+// 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 custom
+
+import (
+	"fmt"
+	"mosn.io/layotto/components/pkg/info"
+)
+
+type Registry interface {
+	Register(componentType string, factorys ...*ComponentFactory)
+	Create(componentType, name string) (Component, error)
+}
+
+type ComponentFactory struct {
+	Name          string
+	FactoryMethod func() Component
+}
+
+func NewComponentFactory(name string, f func() Component) *ComponentFactory {
+	return &ComponentFactory{
+		Name:          name,
+		FactoryMethod: f,
+	}
+}
+
+type componentRegistry struct {
+	stores map[string]map[string]func() Component
+	info   *info.RuntimeInfo
+}
+
+func NewRegistry(info *info.RuntimeInfo) Registry {
+	return &componentRegistry{
+		stores: make(map[string]map[string]func() Component),
+		info:   info,
+	}
+}
+
+func (r *componentRegistry) Register(componentType string, fs ...*ComponentFactory) {
+	if len(fs) == 0 {
+		return
+	}
+	r.info.AddService(componentType)
+	// lazy init
+	if _, ok := r.stores[componentType]; !ok {
+		r.stores[componentType] = make(map[string]func() Component)
+	}
+	// register FactoryMethod
+	for _, f := range fs {
+		r.stores[componentType][f.Name] = f.FactoryMethod
+		r.info.RegisterComponent(componentType, f.Name)
+	}
+}
+
+func (r *componentRegistry) Create(componentType, name string) (Component, error) {
+	store, ok := r.stores[componentType]
+	if !ok {
+		return nil, fmt.Errorf("custom component type %s is not regsitered", componentType)
+	}
+	if f, ok := store[name]; ok {
+		r.info.LoadComponent(componentType, name)
+		return f(), nil
+	}
+	return nil, fmt.Errorf("custom component %s is not regsitered", name)
+}

components/custom/registry_test.go

@@ -0,0 +1,41 @@
+/*
+ * 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 custom
+
+import (
+	"mosn.io/layotto/components/pkg/info"
+	"strings"
+	"testing"
+)
+
+func TestNewRegistry(t *testing.T) {
+	r := NewRegistry(info.NewRuntimeInfo())
+	compType := "my_component"
+	compName := "etcd"
+	r.Register(compType,
+		NewComponentFactory(compName, func() Component {
+			return nil
+		}),
+	)
+	_, err := r.Create(compType, compName)
+	if err != nil {
+		t.Fatalf("create mock store failed: %v", err)
+	}
+	if _, err := r.Create(compType, "not exists"); !strings.Contains(err.Error(), "not regsitered") {
+		t.Fatalf("create mock store failed: %v", err)
+	}
+}

components/pkg/mock/custom_component_mock.go

@@ -0,0 +1,47 @@
+//
+// 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 mock
+
+import (
+	"context"
+	"mosn.io/layotto/components/custom"
+)
+
+type CustomComponentMock struct {
+	ctx       context.Context
+	config    *custom.Config
+	initTimes int
+}
+
+func NewCustomComponentMock() custom.Component {
+	return &CustomComponentMock{}
+}
+
+func (c *CustomComponentMock) InitTimes() int {
+	return c.initTimes
+}
+
+func (c *CustomComponentMock) Initialize(ctx context.Context, config custom.Config) error {
+	c.ctx = ctx
+	c.config = &config
+	c.initTimes++
+	return nil
+}
+
+func (c *CustomComponentMock) GetReceivedConfig() *custom.Config {
+	return c.config
+}
+func (c *CustomComponentMock) GetReceivedCtx() context.Context {
+	return c.ctx
+}

pkg/grpc/dapr/dapr_api.go

@@ -42,7 +42,6 @@ import (
 	"mosn.io/layotto/pkg/grpc/dapr/proto/runtime/v1"
 	dapr_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/runtime/v1"
 	"mosn.io/layotto/pkg/messages"
-	mgrpc "mosn.io/mosn/pkg/filter/network/grpc"
 	"mosn.io/pkg/log"
 	"strings"
 )
@@ -82,9 +81,9 @@ func (d *daprGrpcAPI) startSubscribing() error {
 	return nil
 }
 
-func (d *daprGrpcAPI) Register(s *grpc.Server, registeredServer mgrpc.RegisteredServer) (mgrpc.RegisteredServer, error) {
-	dapr_v1pb.RegisterDaprServer(s, d)
-	return registeredServer, nil
+func (d *daprGrpcAPI) Register(rawGrpcServer *grpc.Server) error {
+	dapr_v1pb.RegisterDaprServer(rawGrpcServer, d)
+	return nil
 }
 
 func (d *daprGrpcAPI) InvokeService(ctx context.Context, in *runtime.InvokeServiceRequest) (*dapr_common_v1pb.InvokeResponse, error) {

pkg/grpc/dapr/dapr_api_secret_test.go

@@ -116,9 +116,7 @@ func TestNewDaprAPI_GetSecretStores(t *testing.T) {
 	}
 	// Setup Dapr API server
 	grpcAPI := NewDaprAPI_Alpha(&grpc_api.ApplicationContext{
-		"", nil, nil, nil, nil,
-		nil, nil, nil, nil,
-		nil, fakeStores})
+		SecretStores: fakeStores})
 	err := grpcAPI.Init(nil)
 	if err != nil {
 		t.Errorf("grpcAPI.Init error")
@@ -190,9 +188,7 @@ func TestGetBulkSecret(t *testing.T) {
 	// Setup Dapr API server
 	// Setup Dapr API server
 	grpcAPI := NewDaprAPI_Alpha(&grpc_api.ApplicationContext{
-		"", nil, nil, nil, nil,
-		nil, nil, nil, nil,
-		nil, fakeStores})
+		SecretStores: fakeStores})
 	// Run test server
 	err := grpcAPI.Init(nil)
 	if err != nil {

pkg/grpc/dapr/dapr_api_test.go

@@ -65,14 +65,13 @@ func TestNewDaprAPI_Alpha(t *testing.T) {
 	}
 	// construct API
 	grpcAPI := NewDaprAPI_Alpha(&grpc_api.ApplicationContext{
-		"", nil, nil, nil, nil,
-		map[string]state.Store{"mock": store}, nil, nil, nil,
-		func(name string, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
+		StateStores: map[string]state.Store{"mock": store},
+		SendToOutputBindingFn: func(name string, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
 			if name == "error-binding" {
 				return nil, errors.New("error when invoke binding")
 			}
 			return &bindings.InvokeResponse{Data: []byte("ok")}, nil
-		}, nil})
+		}})
 	err := grpcAPI.Init(nil)
 	if err != nil {
 		t.Errorf("grpcAPI.Init error")
@@ -108,7 +107,7 @@ func startDaprServerForTest(port int, srv DaprGrpcAPI) *grpc.Server {
 
 	server := grpc.NewServer()
 	go func() {
-		srv.Register(server, server)
+		srv.Register(server)
 		if err := server.Serve(lis); err != nil {
 			panic(err)
 		}

pkg/grpc/default_api/api.go

@@ -23,18 +23,16 @@ import (
 	"sync"
 
 	"github.com/dapr/components-contrib/bindings"
-	grpc_api "mosn.io/layotto/pkg/grpc"
-	"mosn.io/layotto/pkg/grpc/dapr"
-	dapr_common_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/common/v1"
-	dapr_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/runtime/v1"
-	mgrpc "mosn.io/mosn/pkg/filter/network/grpc"
-
 	"github.com/dapr/components-contrib/pubsub"
 	"github.com/dapr/components-contrib/state"
 	jsoniter "github.com/json-iterator/go"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"mosn.io/layotto/components/file"
+	grpc_api "mosn.io/layotto/pkg/grpc"
+	"mosn.io/layotto/pkg/grpc/dapr"
+	dapr_common_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/common/v1"
+	dapr_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/runtime/v1"
 
 	"mosn.io/layotto/components/configstores"
 	"mosn.io/layotto/components/hello"
@@ -78,6 +76,7 @@ type API interface {
 	PublishEvent(context.Context, *runtimev1pb.PublishEventRequest) (*emptypb.Empty, error)
 	// State
 	GetState(ctx context.Context, in *runtimev1pb.GetStateRequest) (*runtimev1pb.GetStateResponse, error)
+	// Get a batch of state data
 	GetBulkState(ctx context.Context, in *runtimev1pb.GetBulkStateRequest) (*runtimev1pb.GetBulkStateResponse, error)
 	SaveState(ctx context.Context, in *runtimev1pb.SaveStateRequest) (*emptypb.Empty, error)
 	DeleteState(ctx context.Context, in *runtimev1pb.DeleteStateRequest) (*emptypb.Empty, error)
@@ -136,10 +135,10 @@ func (a *api) Init(conn *grpc.ClientConn) error {
 	return a.startSubscribing()
 }
 
-func (a *api) Register(s *grpc.Server, registeredServer mgrpc.RegisteredServer) (mgrpc.RegisteredServer, error) {
+func (a *api) Register(rawGrpcServer *grpc.Server) error {
 	LayottoAPISingleton = a
-	runtimev1pb.RegisterRuntimeServer(s, a)
-	return registeredServer, nil
+	runtimev1pb.RegisterRuntimeServer(rawGrpcServer, a)
+	return nil
 }
 
 func NewGrpcAPI(ac *grpc_api.ApplicationContext) grpc_api.GrpcAPI {

pkg/grpc/default_api/api_pubsub.go

@@ -37,6 +37,7 @@ import (
 	"mosn.io/pkg/log"
 )
 
+// Publishes events to the specific topic.
 func (a *api) PublishEvent(ctx context.Context, in *runtimev1pb.PublishEventRequest) (*emptypb.Empty, error) {
 	result, err := a.doPublishEvent(ctx, in.PubsubName, in.Topic, in.Data, in.DataContentType, in.Metadata)
 	if err != nil {

pkg/grpc/default_api/api_state.go

@@ -62,6 +62,7 @@ func (a *api) SaveState(ctx context.Context, in *runtimev1pb.SaveStateRequest) (
 	return a.daprAPI.SaveState(ctx, daprReq)
 }
 
+// GetBulkState gets a batch of state data
 func (a *api) GetBulkState(ctx context.Context, in *runtimev1pb.GetBulkStateRequest) (*runtimev1pb.GetBulkStateResponse, error) {
 	if in == nil {
 		return &runtimev1pb.GetBulkStateResponse{}, status.Error(codes.InvalidArgument, "GetBulkStateRequest is nil")

pkg/grpc/default_api/api_test.go

@@ -168,10 +168,21 @@ func TestInvokeService(t *testing.T) {
 			},
 		}
 
-		a := NewAPI("", nil, nil,
+		a := NewAPI(
+			"",
+			nil,
+			nil,
 			map[string]rpc.Invoker{
 				mosninvoker.Name: mockInvoker,
-			}, nil, nil, nil, nil, nil, nil, nil)
+			},
+			nil,
+			nil,
+			nil,
+			nil,
+			nil,
+			nil,
+			nil,
+		)
 
 		_, err := a.InvokeService(context.Background(), in)
 		assert.Nil(t, err)