remove subscribeTopicEvents API demo

Signed-off-by: “huazhongming” <[email protected]>
mosn · Nov 19, 2024 · 1a503bb · 1a503bb
1 parent 530e88d
commit 1a503bb
Show file tree

Hide file tree

Showing 6 changed files with 12 additions and 118 deletions.
diff --git a/demo/pubsub/README.md b/demo/pubsub/README.md
@@ -17,22 +17,5 @@ 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
-
 
 ```
diff --git a/demo/pubsub/server/dynamic/subscribe_server.go b/demo/pubsub/server/dynamic/subscribe_server.go
diff --git a/pkg/grpc/default_api/api.go b/pkg/grpc/default_api/api.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"sync"
-	"sync/atomic"
 
 	"github.com/dapr/components-contrib/secretstores"
 
@@ -90,7 +89,6 @@ type api struct {
 	streamer          *streamer
 	// json
 	json    jsoniter.API
-	closed  atomic.Bool
 	closeCh chan struct{}
 	wg      sync.WaitGroup
 }

diff --git a/pkg/grpc/default_api/api_pubsub.go b/pkg/grpc/default_api/api_pubsub.go
@@ -154,6 +154,15 @@ func (a *api) publishMessageGRPC(ctx context.Context, msg *pubsub.NewMessage) er
 
 	// TODO tracing
 	envelope, cloudEvent, err := a.envelopeFromSubscriptionMessage(ctx, msg)
+
+	if err != nil {
+		return err
+	}
+
+	if envelope == nil {
+		return nil
+	}
+
 	// Call appcallback
 	clientV1 := runtimev1pb.NewAppCallbackClient(a.AppCallbackConn)
 	res, err := clientV1.OnTopicEvent(ctx, envelope)

diff --git a/pkg/grpc/default_api/api_subscribe.go b/pkg/grpc/default_api/api_subscribe.go
@@ -210,13 +210,13 @@ func (a *api) publishMessageForStream(ctx context.Context, msg *pubsub.NewMessag
 		return fmt.Errorf("no streamer subscribed to pubsub %q topic %q", pubsubName, msg.Topic)
 	}
 
-	envelope, cloudEvent, err := a.envelopeFromSubscriptionMessage(ctx, msg)
+	envelope, cloudEvent, _ := a.envelopeFromSubscriptionMessage(ctx, msg)
 
 	ch, defFn := conn.registerPublishResponse(envelope.GetId())
 	defer defFn()
 
 	conn.streamLock.Lock()
-	err = conn.stream.Send(&runtimev1pb.SubscribeTopicEventsResponse{
+	err := conn.stream.Send(&runtimev1pb.SubscribeTopicEventsResponse{
 		SubscribeTopicEventsResponseType: &runtimev1pb.SubscribeTopicEventsResponse_EventMessage{
 			EventMessage: envelope,
 		},

diff --git a/pkg/grpc/default_api/api_subscribe_test.go b/pkg/grpc/default_api/api_subscribe_test.go
@@ -119,7 +119,7 @@ func TestPublishMessageForStream(t *testing.T) {
 
 		go func() {
 			time.Sleep(1 * time.Second)
-			ch, _ := apiForTest.streamer.subscribers["___test||layotto"].publishResponses["1"]
+			ch := apiForTest.streamer.subscribers["___test||layotto"].publishResponses["1"]
 			ch <- &runtimev1pb.SubscribeTopicEventsRequestProcessed{
 				Id: "1",
 				Status: &runtimev1pb.TopicEventResponse{