Updated to eliminate the double call for 'Register.SubscribeAsync' as…

… the GrpcClient creates the PublishSubscribeReceiver, then calls SubscribeAsync internally. Renamed Register to `SubscribeAsync` and updated return type + example

Signed-off-by: Whit Waldo <whit.waldo@innovian.net>

examples/Client/PublishSubscribe/StreamingSubscriptionExample/Program.cs

@@ -19,14 +19,11 @@ async Task<TopicResponseAction> HandleMessage(TopicMessage message, Cancellation
     }
 }
 
-//Create a dynamic streaming subscription
-var subscription = daprMessagingClient.Register("pubsub", "myTopic",
-    new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(15), TopicResponseAction.Retry)),
-    HandleMessage, CancellationToken.None);
-
-//Subscribe to messages on it with a timeout of 30 seconds
-var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
-await subscription.SubscribeAsync(cancellationTokenSource.Token);
+//Create a dynamic streaming subscription and subscribe with a timeout of 20 seconds
+var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(20));
+var subscription = await daprMessagingClient.SubscribeAsync("pubsub", "myTopic",
+    new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(10), TopicResponseAction.Retry)),
+    HandleMessage, cancellationTokenSource.Token);
 
 await Task.Delay(TimeSpan.FromMinutes(1));
 

src/Dapr.Messaging/PublishSubscribe/DaprPublishSubscribeClient.cs

@@ -27,5 +27,5 @@ public abstract class DaprPublishSubscribeClient
     /// <param name="messageHandler">The delegate reflecting the action to take upon messages received by the subscription.</param>
     /// <param name="cancellationToken">Cancellation token.</param>
     /// <returns></returns>
-    public abstract PublishSubscribeReceiver Register(string pubSubName, string topicName, DaprSubscriptionOptions options, TopicMessageHandler messageHandler, CancellationToken cancellationToken);
+    public abstract Task<IAsyncDisposable> SubscribeAsync(string pubSubName, string topicName, DaprSubscriptionOptions options, TopicMessageHandler messageHandler, CancellationToken cancellationToken);
 }

src/Dapr.Messaging/PublishSubscribe/DaprPublishSubscribeGrpcClient.cs

@@ -39,5 +39,10 @@ public DaprPublishSubscribeGrpcClient(P.DaprClient client)
     /// <param name="messageHandler">The delegate reflecting the action to take upon messages received by the subscription.</param>
     /// <param name="cancellationToken">Cancellation token.</param>
     /// <returns></returns>
-    public override PublishSubscribeReceiver Register(string pubSubName, string topicName, DaprSubscriptionOptions options, TopicMessageHandler messageHandler, CancellationToken cancellationToken) => new(pubSubName, topicName, options, messageHandler, daprClient);
+    public override async Task<IAsyncDisposable> SubscribeAsync(string pubSubName, string topicName, DaprSubscriptionOptions options, TopicMessageHandler messageHandler, CancellationToken cancellationToken)
+    {
+        var receiver = new PublishSubscribeReceiver(pubSubName, topicName, options, messageHandler, daprClient);
+        await receiver.SubscribeAsync(cancellationToken);
+        return receiver;
+    }
 }

src/Dapr.Messaging/PublishSubscribe/PublishSubscribeReceiver.cs

@@ -91,7 +91,7 @@ internal PublishSubscribeReceiver(string pubSubName, string topicName, DaprSubsc
     /// </summary>
     /// <param name="cancellationToken">Cancellation token.</param>
     /// <returns>An <see cref="IAsyncEnumerable{TopicMessage}"/> containing messages provided by the sidecar.</returns>
-    public async Task SubscribeAsync(CancellationToken cancellationToken = default)
+    internal async Task SubscribeAsync(CancellationToken cancellationToken = default)
     {
         //Prevents the receiver from performing the subscribe operation more than once (as the multiple initialization messages would cancel the stream).
         if (hasInitialized)