Moving requesting push message delivery to background. Resolves #4

tpeczek · Mar 21, 2018 · 5005dc1 · 5005dc1
1 parent d1c5444
commit 5005dc1
Show file tree

Hide file tree

Showing 11 changed files with 152 additions and 15 deletions.
diff --git a/Demo.AspNetCore.PushNotifications.Services.Abstractions/IPushNotificationService.cs b/Demo.AspNetCore.PushNotifications.Services.Abstractions/IPushNotificationService.cs
@@ -1,4 +1,5 @@
-using System.Threading.Tasks;
+using System.Threading;
+using System.Threading.Tasks;
 using Lib.Net.Http.WebPush;
 
 namespace Demo.AspNetCore.PushNotifications.Services.Abstractions
@@ -8,5 +9,7 @@ public interface IPushNotificationService
         string PublicKey { get; }
 
         Task SendNotificationAsync(PushSubscription subscription, PushMessage message);
+
+        Task SendNotificationAsync(PushSubscription subscription, PushMessage message, CancellationToken cancellationToken);
     }
 }
diff --git a/Demo.AspNetCore.PushNotifications.Services.Abstractions/IPushNotificationsQueue.cs b/Demo.AspNetCore.PushNotifications.Services.Abstractions/IPushNotificationsQueue.cs
@@ -0,0 +1,13 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Lib.Net.Http.WebPush;
+
+namespace Demo.AspNetCore.PushNotifications.Services.Abstractions
+{
+    public interface IPushNotificationsQueue
+    {
+        void Enqueue(PushMessage message);
+
+        Task<PushMessage> DequeueAsync(CancellationToken cancellationToken);
+    }
+}
diff --git a/Demo.AspNetCore.PushNotifications.Services.Abstractions/IPushSubscriptionStore.cs b/Demo.AspNetCore.PushNotifications.Services.Abstractions/IPushSubscriptionStore.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Lib.Net.Http.WebPush;
 
@@ -11,5 +12,7 @@ public interface IPushSubscriptionStore
         Task DiscardSubscriptionAsync(string endpoint);
 
         Task ForEachSubscriptionAsync(Action<PushSubscription> action);
+
+        Task ForEachSubscriptionAsync(Action<PushSubscription> action, CancellationToken cancellationToken);
     }
 }
diff --git a/Demo.AspNetCore.PushNotifications.Services.PushService/PushServicePushNotificationService.cs b/Demo.AspNetCore.PushNotifications.Services.PushService/PushServicePushNotificationService.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Options;
 using Microsoft.Extensions.Logging;
@@ -33,11 +34,16 @@ public PushServicePushNotificationService(IOptions<PushNotificationServiceOption
             _logger = logger;
         }
 
-        public async Task SendNotificationAsync(PushSubscription subscription, PushMessage message)
+        public Task SendNotificationAsync(PushSubscription subscription, PushMessage message)
+        {
+            return SendNotificationAsync(subscription, message, CancellationToken.None);
+        }
+
+        public async Task SendNotificationAsync(PushSubscription subscription, PushMessage message, CancellationToken cancellationToken)
         {
             try
             {
-                await _pushClient.RequestPushMessageDeliveryAsync(subscription, message);
+                await _pushClient.RequestPushMessageDeliveryAsync(subscription, message, cancellationToken);
             }
             catch (Exception ex)
             {

diff --git a/Demo.AspNetCore.PushNotifications.Services.Sqlite/SqlitePushSubscriptionStore.cs b/Demo.AspNetCore.PushNotifications.Services.Sqlite/SqlitePushSubscriptionStore.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.EntityFrameworkCore;
 using Lib.Net.Http.WebPush;
@@ -33,7 +34,12 @@ public async Task DiscardSubscriptionAsync(string endpoint)
 
         public Task ForEachSubscriptionAsync(Action<PushSubscription> action)
         {
-            return _context.Subscriptions.AsNoTracking().ForEachAsync(action);
+            return ForEachSubscriptionAsync(action, CancellationToken.None);
+        }
+
+        public Task ForEachSubscriptionAsync(Action<PushSubscription> action, CancellationToken cancellationToken)
+        {
+            return _context.Subscriptions.AsNoTracking().ForEachAsync(action, cancellationToken);
         }
     }
 }
diff --git a/Demo.AspNetCore.PushNotifications.Services/Demo.AspNetCore.PushNotifications.Services.csproj b/Demo.AspNetCore.PushNotifications.Services/Demo.AspNetCore.PushNotifications.Services.csproj
@@ -5,6 +5,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.AspNetCore.Hosting.Abstractions" Version="2.0.1" />
     <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.1" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="2.0.0" />
   </ItemGroup>

diff --git a/Demo.AspNetCore.PushNotifications.Services/PushNotificationsDequeuer.cs b/Demo.AspNetCore.PushNotifications.Services/PushNotificationsDequeuer.cs
@@ -0,0 +1,65 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.DependencyInjection;
+using Lib.Net.Http.WebPush;
+using Demo.AspNetCore.PushNotifications.Services.Abstractions;
+
+namespace Demo.AspNetCore.PushNotifications.Services
+{
+    internal class PushNotificationsDequeuer : IHostedService
+    {
+        private readonly IServiceProvider _serviceProvider;
+        private readonly IPushNotificationsQueue _messagesQueue;
+        private readonly IPushNotificationService _notificationService;
+        private readonly CancellationTokenSource _stopTokenSource = new CancellationTokenSource();
+
+        private Task _dequeueMessagesTask;
+
+        public PushNotificationsDequeuer(IServiceProvider serviceProvider, IPushNotificationsQueue messagesQueue, IPushNotificationService notificationService)
+        {
+            _serviceProvider = serviceProvider;
+            _messagesQueue = messagesQueue;
+            _notificationService = notificationService;
+        }
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            _dequeueMessagesTask = Task.Run(DequeueMessagesAsync);
+
+            return Task.CompletedTask;
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            _stopTokenSource.Cancel();
+
+            return Task.WhenAny(_dequeueMessagesTask, Task.Delay(Timeout.Infinite, cancellationToken));
+        }
+
+        private async Task DequeueMessagesAsync()
+        {
+            while (!_stopTokenSource.IsCancellationRequested)
+            {
+                PushMessage message = await _messagesQueue.DequeueAsync(_stopTokenSource.Token);
+
+                if (!_stopTokenSource.IsCancellationRequested)
+                {
+                    using (IServiceScope serviceScope = _serviceProvider.CreateScope())
+                    {
+                        IPushSubscriptionStore subscriptionStore = serviceScope.ServiceProvider.GetRequiredService<IPushSubscriptionStore>();
+
+                        await subscriptionStore.ForEachSubscriptionAsync((PushSubscription subscription) =>
+                        {
+                            // Fire-and-forget 
+                            _notificationService.SendNotificationAsync(subscription, message, _stopTokenSource.Token);
+                        }, _stopTokenSource.Token);
+                    }
+
+                }
+            }
+
+        }
+    }
+}
diff --git a/Demo.AspNetCore.PushNotifications.Services/PushNotificationsQueue.cs b/Demo.AspNetCore.PushNotifications.Services/PushNotificationsQueue.cs
@@ -0,0 +1,36 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Collections.Concurrent;
+using Lib.Net.Http.WebPush;
+using Demo.AspNetCore.PushNotifications.Services.Abstractions;
+
+namespace Demo.AspNetCore.PushNotifications.Services
+{
+    internal class PushNotificationsQueue : IPushNotificationsQueue
+    {
+        private readonly ConcurrentQueue<PushMessage> _messages = new ConcurrentQueue<PushMessage>();
+        private readonly SemaphoreSlim _messageEnqueuedSignal = new SemaphoreSlim(0);
+
+        public void Enqueue(PushMessage message)
+        {
+            if (message == null)
+            {
+                throw new ArgumentNullException(nameof(message));
+            }
+
+            _messages.Enqueue(message);
+
+            _messageEnqueuedSignal.Release();
+        }
+
+        public async Task<PushMessage> DequeueAsync(CancellationToken cancellationToken)
+        {
+            await _messageEnqueuedSignal.WaitAsync(cancellationToken);
+
+            _messages.TryDequeue(out PushMessage message);
+
+            return message;
+        }
+    }
+}
diff --git a/Demo.AspNetCore.PushNotifications.Services/ServiceCollectionExtensions.cs b/Demo.AspNetCore.PushNotifications.Services/ServiceCollectionExtensions.cs
@@ -1,4 +1,4 @@
-using System;
+using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Demo.AspNetCore.PushNotifications.Services.Abstractions;
@@ -27,5 +27,13 @@ public static IServiceCollection AddPushNotificationService(this IServiceCollect
 
             return services;
         }
+
+        public static IServiceCollection AddPushNotificationsQueue(this IServiceCollection services)
+        {
+            services.AddSingleton<IPushNotificationsQueue, PushNotificationsQueue>();
+            services.AddSingleton<IHostedService, PushNotificationsDequeuer>();
+
+            return services;
+        }
     }
 }
diff --git a/Demo.AspNetCore.PushNotifications/Controllers/PushNotificationsApiController.cs b/Demo.AspNetCore.PushNotifications/Controllers/PushNotificationsApiController.cs
@@ -11,11 +11,13 @@ public class PushNotificationsApiController : Controller
     {
         private readonly IPushSubscriptionStore _subscriptionStore;
         private readonly IPushNotificationService _notificationService;
+        private readonly IPushNotificationsQueue _pushNotificationsQueue;
 
-        public PushNotificationsApiController(IPushSubscriptionStore subscriptionStore, IPushNotificationService notificationService)
+        public PushNotificationsApiController(IPushSubscriptionStore subscriptionStore, IPushNotificationService notificationService, IPushNotificationsQueue pushNotificationsQueue)
         {
             _subscriptionStore = subscriptionStore;
             _notificationService = notificationService;
+            _pushNotificationsQueue = pushNotificationsQueue;
         }
 
         // GET push-notifications-api/public-key
@@ -45,19 +47,12 @@ public async Task<IActionResult> DiscardSubscription(string endpoint)
 
         // POST push-notifications-api/notifications
         [HttpPost("notifications")]
-        public async Task<IActionResult> SendNotification([FromBody]PushMessageViewModel message)
+        public IActionResult SendNotification([FromBody]PushMessageViewModel message)
         {
-            PushMessage pushMessage = new PushMessage(message.Notification)
+            _pushNotificationsQueue.Enqueue(new PushMessage(message.Notification)
             {
                 Topic = message.Topic,
                 Urgency = message.Urgency
-            };
-
-            // TODO: This should be scheduled in background
-            await _subscriptionStore.ForEachSubscriptionAsync((PushSubscription subscription) =>
-            {
-                // Fire-and-forget 
-                _notificationService.SendNotificationAsync(subscription, pushMessage);
             });
 
             return NoContent();

diff --git a/Demo.AspNetCore.PushNotifications/Startup.cs b/Demo.AspNetCore.PushNotifications/Startup.cs
@@ -22,6 +22,7 @@ public void ConfigureServices(IServiceCollection services)
         {
             services.AddPushSubscriptionStore(Configuration)
                 .AddPushNotificationService(Configuration)
+                .AddPushNotificationsQueue()
                 .AddMvc(options =>
                 {
                     options.InputFormatters.Add(new TextPlainInputFormatter());