Adds stream cancelled error

Signed-off-by: Elena Kolevska <[email protected]>

Author: elena-kolevska

dapr/aio/clients/grpc/subscription.py

@@ -4,7 +4,8 @@
 
 from dapr.clients.grpc._response import TopicEventResponse
 from dapr.clients.health import DaprHealth
-from dapr.common.pubsub.subscription import StreamInactiveError, SubscriptionMessage
+from dapr.common.pubsub.subscription import StreamInactiveError, SubscriptionMessage, \
+    StreamCancelledError
 from dapr.proto import api_v1, appcallback_v1
 
 
@@ -69,7 +70,9 @@ async def next_message(self):
                     f'Attempting to reconnect...'
                 )
                 await self.reconnect_stream()
-            elif e.code() != StatusCode.CANCELLED:
+            elif e.code() == StatusCode.CANCELLED:
+                raise StreamCancelledError('Stream has been cancelled')
+            else:
                 raise Exception(f'gRPC error while reading from subscription stream: {e} ')
         except Exception as e:
             raise Exception(f'Error while fetching message: {e}')

dapr/clients/grpc/subscription.py

@@ -2,7 +2,8 @@
 
 from dapr.clients.grpc._response import TopicEventResponse
 from dapr.clients.health import DaprHealth
-from dapr.common.pubsub.subscription import StreamInactiveError, SubscriptionMessage
+from dapr.common.pubsub.subscription import StreamInactiveError, SubscriptionMessage, \
+    StreamCancelledError
 from dapr.proto import api_v1, appcallback_v1
 import queue
 import threading
@@ -85,7 +86,9 @@ def next_message(self):
                     f'gRPC error while reading from stream: {e.details()}, Status Code: {e.code()}'
                 )
                 self.reconnect_stream()
-            elif e.code() != StatusCode.CANCELLED:
+            elif e.code() == StatusCode.CANCELLED:
+                raise StreamCancelledError('Stream has been cancelled')
+            else:
                 raise Exception(
                     f'gRPC error while reading from subscription stream: {e.details()} '
                     f'Status Code: {e.code()}'

dapr/common/pubsub/subscription.py

@@ -86,3 +86,6 @@ def _parse_data_content(self):
 
 class StreamInactiveError(Exception):
     pass
+
+class StreamCancelledError(Exception):
+    pass

examples/pubsub-streaming-async/subscriber.py

@@ -3,6 +3,7 @@
 
 from dapr.aio.clients import DaprClient
 from dapr.clients.grpc.subscription import StreamInactiveError
+from dapr.common.pubsub.subscription import StreamCancelledError
 
 parser = argparse.ArgumentParser(description='Publish events to a Dapr pub/sub topic.')
 parser.add_argument('--topic', type=str, required=True, help='The topic name to publish to.')
@@ -33,15 +34,18 @@ async def main():
             while counter < 5:
                 try:
                     message = await subscription.next_message()
+                    if message is None:
+                        print('No message received within timeout period. '
+                              'The stream might have been cancelled.')
+                        continue
 
                 except StreamInactiveError:
                     print('Stream is inactive. Retrying...')
                     await asyncio.sleep(1)
                     continue
-                if message is None:
-                    print('No message received within timeout period.')
-                    continue
-
+                except StreamCancelledError as e:
+                    print('Stream was cancelled')
+                    break
                 # Process the message
                 response_status = process_message(message)
 

examples/pubsub-streaming/subscriber.py

@@ -3,6 +3,7 @@
 
 from dapr.clients import DaprClient
 from dapr.clients.grpc.subscription import StreamInactiveError
+from dapr.common.pubsub.subscription import StreamCancelledError
 
 counter = 0
 
@@ -38,17 +39,22 @@ def main():
             while counter < 5:
                 try:
                     message = subscription.next_message()
+                    if message is None:
+                        print('No message received within timeout period. '
+                              'The stream might have been cancelled.')
+                        continue
 
                 except StreamInactiveError as e:
                     print('Stream is inactive. Retrying...')
                     time.sleep(1)
                     continue
+                except StreamCancelledError as e:
+                    print('Stream was cancelled')
+                    break
                 except Exception as e:
                     print(f'Error occurred: {e}')
                     pass
-                if message is None:
-                    print('No message received within timeout period.')
-                    continue
+
 
                 # Process the message
                 response_status = process_message(message)