feat: Support SubscribeTopicEvents API

Signed-off-by: “huazhongming” <crazyhzm@apache.org>

components/pluggable/grpc_mock.go

@@ -76,7 +76,6 @@ func TestServerFor[TServer any, TClient any](registersvc func(grpc.ServiceRegist
 //	 	client, cleanup, err := serverFactory(&your_service{})
 //		require.NoError(t, err)
 //		defer cleanup()
-//
 func TestSocketServerFor[TServer any, TClient any](registersvc func(grpc.ServiceRegistrar, TServer), clientFactory func(GRPCConnectionDialer) TClient) func(svc TServer) (client TClient, cleanup func(), err error) {
 	return func(srv TServer) (client TClient, cleanup func(), err error) {
 		const fakeSocketFolder = "/tmp"

components/sequencer/snowflake/snowflake.go

@@ -1,4 +1,3 @@
-//
 // 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.
@@ -242,9 +241,9 @@ func NewMysqlClient(meta *SnowflakeMysqlMetadata) (int64, error) {
 	return workId, err
 }
 
-//get id from mysql
-//host_name = "ip"
-//port = "timestamp-random number"
+// get id from mysql
+// host_name = "ip"
+// port = "timestamp-random number"
 func NewWorkId(meta SnowflakeMysqlMetadata) (int64, error) {
 	var workId int64
 	ip, err := getIP()

components/sequencer/snowflake/snowflake_sequencer.go

@@ -1,4 +1,3 @@
-//
 // 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.

components/sequencer/snowflake/snowflake_sequencer_test.go

@@ -1,4 +1,3 @@
-//
 // 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.

components/sequencer/snowflake/snowflake_test.go

@@ -1,4 +1,3 @@
-//
 // 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.

demo/pubsub/README.md

@@ -2,7 +2,8 @@ You can run server/client demo with different component names.
 It is worth noting that both server and client demo should set the same store name by param `-s`.  
 For example:   
 ```shell
-cd ${project_path}/demo/pubsub/server/
+#################### Run pubsub demo with appcallback ####################
+cd ${project_path}/demo/pubsub/appcallback/
 # 1. start subscriber
 go build -o subscriber
 /.subscriber -s pub_subs_demo
@@ -16,5 +17,22 @@ go build -o layotto
  cd ${project_path}/demo/pubsub/client/
 go build -o publisher
  ./publisher -s pub_subs_demo
+
+ #################### Run pubsub demo with SubscribeTopicEvents ####################
+# 1. start layotto
+cd ${project_path}/cmd/layotto
+go build -o layotto
+./layotto start -c ../../configs/config_standalone.json
+
+cd ${project_path}/demo/pubsub/dynamic/
+# 2. start subscriber
+go build -o subscriber
+/.subscriber -s pub_subs_demo
+
+# 3. start publisher
+ cd ${project_path}/demo/pubsub/client/
+go build -o publisher
+ ./publisher -s pub_subs_demo
+
 
 ```

demo/pubsub/server/dynamic/subscribe_server.go

@@ -0,0 +1,96 @@
+/*
+ * 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 main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"log"
+	"time"
+
+	"mosn.io/layotto/sdk/go-sdk/client"
+
+	runtimev1pb "mosn.io/layotto/spec/proto/runtime/v1"
+)
+
+var storeName string
+
+func init() {
+	flag.StringVar(&storeName, "s", "", "set `storeName`")
+}
+
+func main() {
+	flag.Parse()
+	if storeName == "" {
+		panic("storeName is empty.")
+	}
+	testDynamicSub()
+}
+
+func testDynamicSub() {
+	cli, err := client.NewClient()
+
+	// Use SubscribeWithHandler API to subscribe to a topic.
+	stop, err := cli.SubscribeWithHandler(context.Background(), client.SubscriptionRequest{
+		PubsubName: storeName,
+		Topic:      "hello",
+		Metadata:   nil,
+	}, eventHandler)
+
+	//Use Subscribe API to subscribe to a topic.
+	sub, err := cli.Subscribe(context.Background(), client.SubscriptionRequest{
+		PubsubName: storeName,
+		Topic:      "topic1",
+		Metadata:   nil,
+	})
+
+	if err != nil {
+		log.Fatalf("failed to subscribe to topic: %v", err)
+	}
+
+	msg, err := sub.Receive()
+	if err != nil {
+		log.Fatalf("Error receiving message: %v", err)
+	}
+	log.Printf(">>[Subscribe API]Received message\n")
+	log.Printf("event - PubsubName: %s, Topic: %s, ID: %s, Data: %s\n", msg.PubsubName, msg.Topic, msg.Id, msg.Data)
+
+	// Use _MUST_ always signal the result of processing the message, else the
+	// message will not be considered as processed and will be redelivered or
+	// dead lettered.
+	if err := msg.Success(); err != nil {
+		log.Fatalf("error sending message success: %v", err)
+	}
+
+	time.Sleep(time.Second * 10)
+
+	if err := errors.Join(stop(), sub.Close()); err != nil {
+		log.Fatal(err)
+	}
+
+	if err != nil {
+		panic(err)
+	}
+	cli.Close()
+}
+
+func eventHandler(request *runtimev1pb.TopicEventRequest) client.SubscriptionResponseStatus {
+	log.Printf(">>[SubscribeWithHandler API] Received message\n")
+	log.Printf("event - PubsubName: %s, Topic: %s, ID: %s, Data: %s\n", request.PubsubName, request.Topic, request.Id, request.Data)
+	return client.SubscriptionResponseStatusSuccess
+}

pkg/grpc/default_api/api.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"sync/atomic"
 
 	"github.com/dapr/components-contrib/secretstores"
 
@@ -86,8 +87,12 @@ type api struct {
 	// app callback
 	AppCallbackConn   *grpc.ClientConn
 	topicPerComponent map[string]TopicSubscriptions
+	streamer          *streamer
 	// json
-	json jsoniter.API
+	json    jsoniter.API
+	closed  atomic.Bool
+	closeCh chan struct{}
+	wg      sync.WaitGroup
 }
 
 func (a *api) Init(conn *grpc.ClientConn) error {
@@ -147,7 +152,9 @@ func NewAPI(
 		sendToOutputBindingFn:    sendToOutputBindingFn,
 		secretStores:             secretStores,
 		json:                     jsoniter.ConfigFastest,
+		//closeCh:                  make(chan struct{}),
 	}
+
 }
 
 func (a *api) SayHello(ctx context.Context, in *runtimev1pb.SayHelloRequest) (*runtimev1pb.SayHelloResponse, error) {

pkg/grpc/default_api/api_pubsub.go

@@ -151,56 +151,14 @@ func (a *api) getInterestedTopics() (map[string]TopicSubscriptions, error) {
 }
 
 func (a *api) publishMessageGRPC(ctx context.Context, msg *pubsub.NewMessage) error {
-	// 1. Unmarshal to cloudEvent model
-	var cloudEvent map[string]interface{}
-	err := a.json.Unmarshal(msg.Data, &cloudEvent)
-	if err != nil {
-		log.DefaultLogger.Debugf("[runtime]error deserializing cloud events proto: %s", err)
-		return err
-	}
-
-	// 2. Drop msg if the current cloud event has expired
-	if pubsub.HasExpired(cloudEvent) {
-		log.DefaultLogger.Warnf("[runtime]dropping expired pub/sub event %v as of %v", cloudEvent[pubsub.IDField].(string), cloudEvent[pubsub.ExpirationField].(string))
-		return nil
-	}
 
-	// 3. Convert to proto domain struct
-	envelope := &runtimev1pb.TopicEventRequest{
-		Id:              cloudEvent[pubsub.IDField].(string),
-		Source:          cloudEvent[pubsub.SourceField].(string),
-		DataContentType: cloudEvent[pubsub.DataContentTypeField].(string),
-		Type:            cloudEvent[pubsub.TypeField].(string),
-		SpecVersion:     cloudEvent[pubsub.SpecVersionField].(string),
-		Topic:           msg.Topic,
-		PubsubName:      msg.Metadata[Metadata_key_pubsubName],
-	}
-
-	// set data field
-	if data, ok := cloudEvent[pubsub.DataBase64Field]; ok && data != nil {
-		decoded, decodeErr := base64.StdEncoding.DecodeString(data.(string))
-		if decodeErr != nil {
-			log.DefaultLogger.Debugf("unable to base64 decode cloudEvent field data_base64: %s", decodeErr)
-			return err
-		}
-
-		envelope.Data = decoded
-	} else if data, ok := cloudEvent[pubsub.DataField]; ok && data != nil {
-		envelope.Data = nil
-
-		if contenttype.IsStringContentType(envelope.DataContentType) {
-			envelope.Data = []byte(data.(string))
-		} else if contenttype.IsJSONContentType(envelope.DataContentType) {
-			envelope.Data, _ = a.json.Marshal(data)
-		}
-	}
 	// TODO tracing
-
-	// 4. Call appcallback
+	envelope, cloudEvent, err := a.envelopeFromSubscriptionMessage(ctx, msg)
+	// Call appcallback
 	clientV1 := runtimev1pb.NewAppCallbackClient(a.AppCallbackConn)
 	res, err := clientV1.OnTopicEvent(ctx, envelope)
 
-	// 5. Check result
+	// Check result
 	return retryStrategy(err, res, cloudEvent)
 }
 
@@ -246,3 +204,50 @@ func listTopicSubscriptions(client runtimev1pb.AppCallbackClient, log log.ErrorL
 	}
 	return make([]*runtimev1pb.TopicSubscription, 0)
 }
+
+func (a *api) envelopeFromSubscriptionMessage(ctx context.Context, msg *pubsub.NewMessage) (*runtimev1pb.TopicEventRequest, map[string]interface{}, error) {
+	// 1. Unmarshal to cloudEvent model
+	var cloudEvent map[string]interface{}
+	err := a.json.Unmarshal(msg.Data, &cloudEvent)
+	if err != nil {
+		log.DefaultLogger.Debugf("[runtime]error deserializing cloud events proto: %s", err)
+		return nil, cloudEvent, err
+	}
+
+	// 2. Drop msg if the current cloud event has expired
+	if pubsub.HasExpired(cloudEvent) {
+		log.DefaultLogger.Warnf("[runtime]dropping expired pub/sub event %v as of %v", cloudEvent[pubsub.IDField].(string), cloudEvent[pubsub.ExpirationField].(string))
+		return nil, cloudEvent, nil
+	}
+
+	// 3. Convert to proto domain struct
+	envelope := &runtimev1pb.TopicEventRequest{
+		Id:              cloudEvent[pubsub.IDField].(string),
+		Source:          cloudEvent[pubsub.SourceField].(string),
+		DataContentType: cloudEvent[pubsub.DataContentTypeField].(string),
+		Type:            cloudEvent[pubsub.TypeField].(string),
+		SpecVersion:     cloudEvent[pubsub.SpecVersionField].(string),
+		Topic:           msg.Topic,
+		PubsubName:      msg.Metadata[Metadata_key_pubsubName],
+	}
+
+	// set data field
+	if data, ok := cloudEvent[pubsub.DataBase64Field]; ok && data != nil {
+		decoded, decodeErr := base64.StdEncoding.DecodeString(data.(string))
+		if decodeErr != nil {
+			log.DefaultLogger.Debugf("unable to base64 decode cloudEvent field data_base64: %s", decodeErr)
+			return nil, cloudEvent, err
+		}
+
+		envelope.Data = decoded
+	} else if data, ok := cloudEvent[pubsub.DataField]; ok && data != nil {
+		envelope.Data = nil
+
+		if contenttype.IsStringContentType(envelope.DataContentType) {
+			envelope.Data = []byte(data.(string))
+		} else if contenttype.IsJSONContentType(envelope.DataContentType) {
+			envelope.Data, _ = a.json.Marshal(data)
+		}
+	}
+	return envelope, cloudEvent, nil
+}