UserEventFirer, web hook tests

Selector.CLI/DbWatcherService.cs

@@ -30,6 +30,8 @@ namespace Selector.CLI
         private readonly IAudioFeatureInjectorFactory AudioFeatureInjectorFactory;
         private readonly IPlayCounterFactory PlayCounterFactory;
 
+        private readonly IUserEventFirerFactory UserEventFirerFactory;
+
         private readonly IPublisherFactory PublisherFactory;
         private readonly ICacheWriterFactory CacheWriterFactory;
         private ConcurrentDictionary<string, IWatcherCollection> Watchers { get; set; } = new();
@@ -48,7 +50,9 @@ namespace Selector.CLI
             IServiceProvider serviceProvider,
 
             IPublisherFactory publisherFactory = null,
-            ICacheWriterFactory cacheWriterFactory = null
+            ICacheWriterFactory cacheWriterFactory = null,
+
+            IUserEventFirerFactory userEventFirerFactory = null
         )
         {
             Logger = logger;
@@ -62,6 +66,8 @@ namespace Selector.CLI
             AudioFeatureInjectorFactory = audioFeatureInjectorFactory;
             PlayCounterFactory = playCounterFactory;
 
+            UserEventFirerFactory = userEventFirerFactory;
+
             PublisherFactory = publisherFactory;
             CacheWriterFactory = cacheWriterFactory;
         }
@@ -123,6 +129,8 @@ namespace Selector.CLI
                     if (CacheWriterFactory is not null) consumers.Add(await CacheWriterFactory.Get());
                     if (PublisherFactory is not null) consumers.Add(await PublisherFactory.Get());
 
+                    if (UserEventFirerFactory is not null) consumers.Add(await UserEventFirerFactory.Get());
+
                     if (!string.IsNullOrWhiteSpace(dbWatcher.User.LastFmUsername))
                     {
                         consumers.Add(await PlayCounterFactory.Get(creds: new() { Username = dbWatcher.User.LastFmUsername }));

Selector.CLI/Program.cs

@@ -125,8 +125,8 @@ namespace Selector.CLI
                 services.AddRedisServices(config.RedisOptions.ConnectionString);
 
                 Console.WriteLine("> Adding cache event maps...");
-                services.AddTransient<IEventMapping, SpotifyLinkFromCacheMapping>();
+                services.AddTransient<IEventMapping, FromPubSub.SpotifyLink>();
-                services.AddTransient<IEventMapping, LastfmFromCacheMapping>();
+                services.AddTransient<IEventMapping, FromPubSub.Lastfm>();
 
                 Console.WriteLine("> Adding caching Spotify consumers...");
                 services.AddCachingSpotify();

Selector.Event/CacheMappings/LastfmMapping.cs

@@ -14,13 +14,15 @@ namespace Selector.Events
         public string NewUsername { get; set; }
     }
 
-    public class LastfmFromCacheMapping : IEventMapping
+    public partial class FromPubSub
     {
-        private readonly ILogger<LastfmFromCacheMapping> Logger;
+        public class Lastfm : IEventMapping
+        {
+            private readonly ILogger<Lastfm> Logger;
             private readonly ISubscriber Subscriber;
             private readonly UserEventBus UserEvent;
 
-        public LastfmFromCacheMapping(ILogger<LastfmFromCacheMapping> logger, 
+            public Lastfm(ILogger<Lastfm> logger,
                 ISubscriber subscriber,
                 UserEventBus userEvent)
             {
@@ -35,7 +37,8 @@ namespace Selector.Events
 
                 (await Subscriber.SubscribeAsync(Key.AllUserLastfm)).OnMessage(message => {
 
-                try{
+                    try
+                    {
                         var userId = Key.Param(message.Channel);
 
                         var deserialised = JsonSerializer.Deserialize<LastfmChange>(message.Message);
@@ -48,21 +51,24 @@ namespace Selector.Events
 
                         UserEvent.OnLastfmCredChange(this, deserialised);
                     }
-                catch(Exception e)
+                    catch (Exception e)
                     {
                         Logger.LogError(e, "Error parsing Last.fm username event");
                     }
                 });
             }
         }
+    }
 
-    public class LastfmToCacheMapping : IEventMapping
+    public partial class ToPubSub
     {
-        private readonly ILogger<LastfmToCacheMapping> Logger;
+        public class Lastfm : IEventMapping
+        {
+            private readonly ILogger<Lastfm> Logger;
             private readonly ISubscriber Subscriber;
             private readonly UserEventBus UserEvent;
 
-        public LastfmToCacheMapping(ILogger<LastfmToCacheMapping> logger,
+            public Lastfm(ILogger<Lastfm> logger,
                 ISubscriber subscriber,
                 UserEventBus userEvent)
             {
@@ -84,4 +90,5 @@ namespace Selector.Events
                 return Task.CompletedTask;
             }
         }
+    }
 }

Selector.Event/CacheMappings/NowPlayingMapping.cs

@@ -7,13 +7,15 @@ using Selector.Cache;
 
 namespace Selector.Events
 {
-    public class NowPlayingFromCacheMapping : IEventMapping
+    public partial class FromPubSub
     {
-        private readonly ILogger<NowPlayingFromCacheMapping> Logger;
+        public class NowPlaying : IEventMapping
+        {
+            private readonly ILogger<NowPlaying> Logger;
             private readonly ISubscriber Subscriber;
             private readonly UserEventBus UserEvent;
 
-        public NowPlayingFromCacheMapping(ILogger<NowPlayingFromCacheMapping> logger, 
+            public NowPlaying(ILogger<NowPlaying> logger,
                 ISubscriber subscriber,
                 UserEventBus userEvent)
             {
@@ -28,7 +30,8 @@ namespace Selector.Events
 
                 (await Subscriber.SubscribeAsync(Key.AllCurrentlyPlaying)).OnMessage(message => {
 
-                try{
+                    try
+                    {
                         var userId = Key.Param(message.Channel);
 
                         var deserialised = JsonSerializer.Deserialize<CurrentlyPlayingDTO>(message.Message);
@@ -36,11 +39,46 @@ namespace Selector.Events
 
                         UserEvent.OnCurrentlyPlayingChange(this, userId, deserialised);
                     }
-                catch(Exception e)
+                    catch (Exception e)
                     {
                         Logger.LogError(e, $"Error parsing new currently playing [{message}]");
                     }
                 });
             }
         }
+    }
+
+    public partial class ToPubSub
+    {
+        public class NowPlaying : IEventMapping
+        {
+            private readonly ILogger<NowPlaying> Logger;
+            private readonly ISubscriber Subscriber;
+            private readonly UserEventBus UserEvent;
+
+            public NowPlaying(ILogger<NowPlaying> logger,
+                ISubscriber subscriber,
+                UserEventBus userEvent)
+            {
+                Logger = logger;
+                Subscriber = subscriber;
+                UserEvent = userEvent;
+            }
+
+            public Task ConstructMapping()
+            {
+                Logger.LogDebug("Forming now playing event mapping TO cache FROM event bus");
+
+                UserEvent.CurrentlyPlaying += async (o, e) =>
+                {
+                    (string id, CurrentlyPlayingDTO args) = e;
+
+                    var payload = JsonSerializer.Serialize(e);
+                    await Subscriber.PublishAsync(Key.CurrentlyPlaying(id), payload);
+                };
+
+                return Task.CompletedTask;
+            }
+        }
+    }
 }

Selector.Event/CacheMappings/SpotifyMapping.cs

@@ -14,13 +14,15 @@ namespace Selector.Events
         public bool NewLinkState { get; set; }
     }
 
-    public class SpotifyLinkFromCacheMapping : IEventMapping
+    public partial class FromPubSub
     {
-        private readonly ILogger<SpotifyLinkFromCacheMapping> Logger;
+        public class SpotifyLink : IEventMapping
+        {
+            private readonly ILogger<SpotifyLink> Logger;
             private readonly ISubscriber Subscriber;
             private readonly UserEventBus UserEvent;
 
-        public SpotifyLinkFromCacheMapping(ILogger<SpotifyLinkFromCacheMapping> logger, 
+            public SpotifyLink(ILogger<SpotifyLink> logger,
                 ISubscriber subscriber,
                 UserEventBus userEvent)
             {
@@ -35,7 +37,8 @@ namespace Selector.Events
 
                 (await Subscriber.SubscribeAsync(Key.AllUserSpotify)).OnMessage(message => {
 
-                try{
+                    try
+                    {
                         var userId = Key.Param(message.Channel);
 
                         var deserialised = JsonSerializer.Deserialize<SpotifyLinkChange>(message.Message);
@@ -48,25 +51,28 @@ namespace Selector.Events
 
                         UserEvent.OnSpotifyLinkChange(this, deserialised);
                     }
-                catch(TaskCanceledException)
+                    catch (TaskCanceledException)
                     {
                         Logger.LogDebug("Task Cancelled");
                     }
-                catch(Exception e)
+                    catch (Exception e)
                     {
                         Logger.LogError(e, "Error parsing new Spotify link event");
                     }
                 });
             }
         }
+    }
 
-    public class SpotifyLinkToCacheMapping : IEventMapping
+    public partial class ToPubSub
     {
-        private readonly ILogger<SpotifyLinkToCacheMapping> Logger;
+        public class SpotifyLink : IEventMapping
+        {
+            private readonly ILogger<SpotifyLink> Logger;
             private readonly ISubscriber Subscriber;
             private readonly UserEventBus UserEvent;
 
-        public SpotifyLinkToCacheMapping(ILogger<SpotifyLinkToCacheMapping> logger,
+            public SpotifyLink(ILogger<SpotifyLink> logger,
                 ISubscriber subscriber,
                 UserEventBus userEvent)
             {
@@ -88,4 +94,5 @@ namespace Selector.Events
                 return Task.CompletedTask;
             }
         }
+    }
 }

Selector.Event/Consumers/UserEventFirer.cs

@@ -0,0 +1,83 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
+using Selector.Events;
+
+namespace Selector
+{
+    public class UserEventFirer : IConsumer
+    {
+        protected readonly IPlayerWatcher Watcher;
+        protected readonly ILogger<UserEventFirer> Logger;
+
+        protected readonly UserEventBus UserEvent;
+
+        public CancellationToken CancelToken { get; set; }
+
+        public UserEventFirer(
+            IPlayerWatcher watcher,
+            UserEventBus userEvent,
+            ILogger<UserEventFirer> logger = null,
+            CancellationToken token = default
+        )
+        {
+            Watcher = watcher;
+            UserEvent = userEvent;
+            Logger = logger ?? NullLogger<UserEventFirer>.Instance;
+            CancelToken = token;
+        }
+
+        public void Callback(object sender, ListeningChangeEventArgs e)
+        {
+            if (e.Current is null) return;
+            
+            Task.Run(async () => {
+                try
+                {
+                    await AsyncCallback(e);
+                }
+                catch (Exception e)
+                {
+                    Logger.LogError(e, "Error occured during callback");
+                }
+            }, CancelToken);
+        }
+
+        public Task AsyncCallback(ListeningChangeEventArgs e)
+        {
+            Logger.LogDebug("Firing now playing event on user bus [{username}/{userId}]", e.SpotifyUsername, e.Id);
+
+            UserEvent.OnCurrentlyPlayingChange(this, e.Id, (CurrentlyPlayingDTO) e);
+
+            return Task.CompletedTask;
+        }
+
+        public void Subscribe(IWatcher watch = null)
+        {
+            var watcher = watch ?? Watcher ?? throw new ArgumentNullException("No watcher provided");
+
+            if (watcher is IPlayerWatcher watcherCast)
+            {
+                watcherCast.ItemChange += Callback;
+            }
+            else
+            {
+                throw new ArgumentException("Provided watcher is not a PlayerWatcher");
+            }
+        }
+
+        public void Unsubscribe(IWatcher watch = null)
+        {
+            var watcher = watch ?? Watcher ?? throw new ArgumentNullException("No watcher provided");
+
+            if (watcher is IPlayerWatcher watcherCast)
+            {
+                watcherCast.ItemChange -= Callback;
+            }
+            else
+            {
+                throw new ArgumentException("Provided watcher is not a PlayerWatcher");
+            }
+        }
+    }
+}

Selector.Event/Consumers/UserEventFirerFactory.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Selector.Events;
+
+namespace Selector
+{
+    public interface IUserEventFirerFactory
+    {
+        public Task<UserEventFirer> Get(IPlayerWatcher watcher = null);
+    }
+    
+    public class UserEventFirerFactory: IUserEventFirerFactory
+    {
+        private readonly ILoggerFactory LoggerFactory;
+        private readonly UserEventBus UserEvent;
+
+        public UserEventFirerFactory(ILoggerFactory loggerFactory, UserEventBus userEvent)
+        {
+            LoggerFactory = loggerFactory;
+            UserEvent = userEvent;
+        }
+
+        public Task<UserEventFirer> Get(IPlayerWatcher watcher = null)
+        {
+            return Task.FromResult(new UserEventFirer(
+                watcher,
+                UserEvent,
+                LoggerFactory.CreateLogger<UserEventFirer>()
+            ));
+        }
+    }
+}

Selector.Event/Extensions/ServiceExtensions.cs

@@ -8,6 +8,9 @@ namespace Selector.Events
         {
             services.AddEventBus();
             services.AddEventMappingAgent();
+
+            services.AddTransient<IUserEventFirerFactory, UserEventFirerFactory>();
+            services.AddTransient<UserEventFirerFactory>();
         }
 
         public static void AddEventBus(this IServiceCollection services)

Selector.Event/UserEventBus.cs

@@ -1,9 +1,4 @@
-using System;
+using Microsoft.Extensions.Logging;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using Microsoft.Extensions.Logging;
 
 using Selector.Model;
 

Selector.Tests/Consumer/WebHook.cs

@@ -0,0 +1,106 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Net.Http;
+
+using Xunit;
+using Moq;
+using Moq.Protected;
+using FluentAssertions;
+using System.Net;
+
+using SpotifyAPI.Web;
+
+namespace Selector.Tests
+{
+    public class WebHookTest
+    {
+        [Fact(Skip = "Not working atm")]
+        public async Task TestHttpClientUsed()
+        {
+            var msg = new HttpResponseMessage(HttpStatusCode.OK);
+
+            var httpHandlerMock = new Mock<HttpMessageHandler>();
+            httpHandlerMock.Protected()
+                .Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
+                .ReturnsAsync(msg);
+
+            var watcherMock = new Mock<IPlayerWatcher>();
+            watcherMock.SetupAdd(w => w.ItemChange += It.IsAny<EventHandler<ListeningChangeEventArgs>>());
+            watcherMock.SetupRemove(w => w.ItemChange -= It.IsAny<EventHandler<ListeningChangeEventArgs>>());
+
+            var link = "https://link";
+            var content = new StringContent("");
+            var config = new WebHookConfig()
+            {
+                Url = link,
+                Content = content,
+            };
+
+            var http = new HttpClient(httpHandlerMock.Object);
+
+            var webHook = new WebHook(watcherMock.Object, http, config);
+
+            webHook.Subscribe();
+            watcherMock.Raise(w => w.ItemChange += null, this, new ListeningChangeEventArgs());
+
+            await Task.Delay(100);
+
+            httpHandlerMock.Protected().Verify<Task<HttpResponseMessage>>("SendAsync", Times.Once(), ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>());
+        }
+
+        [Theory]
+        [InlineData(200, true, true)]
+        [InlineData(404, true, false)]
+        [InlineData(500, true, false)]
+        public async Task TestEventFiring(int code, bool predicate, bool successful)
+        {
+            var msg = new HttpResponseMessage(Enum.Parse<HttpStatusCode>(code.ToString()));
+
+            var httpHandlerMock = new Mock<HttpMessageHandler>();
+            httpHandlerMock.Protected()
+                .Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
+                .ReturnsAsync(msg);
+
+            var watcherMock = new Mock<IPlayerWatcher>();
+
+            var link = "https://link";
+            var content = new StringContent("");
+            var config = new WebHookConfig()
+            {
+                Url = link,
+                Content = content,
+            };
+
+            var http = new HttpClient(httpHandlerMock.Object);
+
+            bool predicateEvent = false, successfulEvent = false, failedEvent = false;
+
+            var webHook = new WebHook(watcherMock.Object, http, config);
+
+            webHook.PredicatePass += (o, e) =>
+            {
+                predicateEvent = predicate;
+            };
+
+            webHook.SuccessfulRequest += (o, e) =>
+            {
+                successfulEvent = successful;
+            };
+
+            webHook.FailedRequest += (o, e) =>
+            {
+                failedEvent = !successful;
+            };
+
+            await webHook.AsyncCallback(ListeningChangeEventArgs.From(new (), new (), new()));
+
+            predicateEvent.Should().Be(predicate);
+            successfulEvent.Should().Be(successful);
+            failedEvent.Should().Be(!successful);
+        }
+    }
+}

Selector.Web/Areas/Identity/Pages/Account/Manage/Lastfm.cshtml

@@ -5,7 +5,7 @@
     ViewData["ActivePage"] = ManageNavPages.LastFm;
 }
 
-<h4>@ViewData["Title"] <a href="https://last.fm"><img src="/last-fm.png" class="lastfm-logo central" /></a></h4>
+<h4>@ViewData["Title"] <a href="https://last.fm" target="_blank"><img src="/last-fm.png" class="lastfm-logo central" /></a></h4>
 <partial name="_StatusMessage" model="Model.StatusMessage" />
 <div class="row">
     <div class="col-md-6">

Selector.Web/Areas/Identity/Pages/Account/Manage/Spotify.cshtml

@@ -5,7 +5,7 @@
     ViewData["ActivePage"] = ManageNavPages.Spotify;
 }
 
-<h4>@ViewData["Title"] <a href="https://spotify.com"><img src="/Spotify_Icon_RGB_White.png" class="spotify-logo central" /></a></h4>
+<h4>@ViewData["Title"] <a href="https://spotify.com" target="_blank"><img src="/Spotify_Icon_RGB_White.png" class="spotify-logo central" /></a></h4>
 <partial name="_StatusMessage" model="Model.StatusMessage" />
 
 <div class="row">

Selector.Web/Startup.cs

@@ -107,9 +107,9 @@ namespace Selector.Web
 
                 Console.WriteLine("> Adding cache event maps...");
 
-                services.AddTransient<IEventMapping, SpotifyLinkToCacheMapping>();
+                services.AddTransient<IEventMapping, ToPubSub.SpotifyLink>();
-                services.AddTransient<IEventMapping, LastfmToCacheMapping>();
+                services.AddTransient<IEventMapping, ToPubSub.Lastfm>();
-                services.AddTransient<IEventMapping, NowPlayingFromCacheMapping>();
+                services.AddTransient<IEventMapping, FromPubSub.NowPlaying>();
 
                 services.AddCacheHubProxy();
 

Selector/Consumers/WebHook.cs

@@ -38,9 +38,9 @@ namespace Selector
 
         protected readonly WebHookConfig Config;
 
-        protected event EventHandler PredicatePass;
+        public event EventHandler PredicatePass;
-        protected event EventHandler SuccessfulRequest;
+        public event EventHandler SuccessfulRequest;
-        protected event EventHandler FailedRequest;
+        public event EventHandler FailedRequest;
 
         public CancellationToken CancelToken { get; set; }
 
@@ -80,12 +80,14 @@ namespace Selector
         public async Task AsyncCallback(ListeningChangeEventArgs e)
         {
             if(Config.ShouldRequest(e))
+            {
+                try
                 {
                     Logger.LogDebug("[{name}] predicate passed, making request to [{url}]", Config.Name, Config.Url);
-                var response = await HttpClient.PostAsync(Config.Url, Config.Content, CancelToken);
-
                     OnPredicatePass(new EventArgs());
 
+                    var response = await HttpClient.PostAsync(Config.Url, Config.Content, CancelToken);
+
                     if (response.IsSuccessStatusCode)
                     {
                         Logger.LogDebug("[{name}] request success", Config.Name);
@@ -97,6 +99,15 @@ namespace Selector
                         OnFailedRequest(new EventArgs());
                     }
                 }
+                catch(HttpRequestException ex)
+                {
+                    Logger.LogError(ex, "Exception occured during request");
+                }
+                catch (TaskCanceledException)
+                {
+                    Logger.LogDebug("Task cancelled");
+                }
+            }
             else
             {
                 Logger.LogTrace("[{name}] predicate failed, skipping", Config.Name);