fix server shutdown requested (#41)

examples/components/comp-nats.yaml

@@ -6,16 +6,17 @@ spec:
   type: pubsub.natsstreaming
   version: v1
   metadata:
-  - name: natsURL
-    value: "nats://kube-worker1.totalsoft.local:31291"
-  - name: natsStreamingClusterID
-    value: "faas-cluster"
-    # below are subscription configuration.
-  - name: subscriptionType
-    value: queue # Required. Allowed values: topic, queue.
-  - name: ackWaitTime
-    value: "" # Optional.
-  - name: maxInFlight
-    value: "1" # Optional.
-  - name: durableSubscriptionName
-    value: "" # Optional.
+    - name: natsURL
+      value: "nats://kube-worker1:31291"
+    - name: natsStreamingClusterID
+      value: faas-cluster
+    - name: subscriptionType
+      value: queue
+    - name: connectWait
+      value: 10s
+    - name: ackWaitTime
+      value: 50s
+    - name: maxInFlight
+      value: '1'
+    - name: durableSubscriptionName
+      value: durable

examples/components/config-node-pipeline.yaml

@@ -5,14 +5,14 @@ metadata:
 spec:
   metric:
     enabled: true
-  tracing:
-    samplingRate: '1'
-    jaeger:
-      useAgent: false
-      collectorEndpointAddress: 'http://kube-worker1.totalsoft.local:31034/api/traces'
-  subscriberPipeline:
-    handlers:
-    - name: pubsub-uppercase
-      type: middleware.pubsub.uppercase
+#  tracing:
+#    samplingRate: '1'
+#    jaeger:
+#      useAgent: false
+#      collectorEndpointAddress: 'http://kube-worker1.totalsoft.local:31034/api/traces'
+#  subscriberPipeline:
+#    handlers:
+#    - name: pubsub-uppercase
+#      type: middleware.pubsub.uppercase
   pubSub:
     name: natsstreaming-pubsub

pkg/api/runtime/grpc/grpc_api.go

@@ -166,7 +166,7 @@ func (srv *rusiServerImpl) Subscribe(stream v1.Rusi_SubscribeServer) error {
 		if exit {
 			klog.V(4).InfoS("closing subscription stream", "topic", request.Topic, "error", err)
 			if errors.Is(err, context.Canceled) {
-				return nil
+				err = status.Error(codes.Aborted, "Cancellation requested")
 			}
 			return err
 		}