adding cache consumers

Selector.CLI/Options.cs

@@ -71,7 +71,7 @@ namespace Selector.CLI
 
     enum Consumers
     {
-        AudioFeatures
+        AudioFeatures, CacheWriter, Publisher
     }
 
     class DatabaseOptions {

Selector.CLI/Program.cs

@@ -39,7 +39,7 @@ namespace Selector.CLI
                     Console.WriteLine("> Adding Services...");
                     // SERVICES
                     services.AddSingleton<IWatcherFactory, WatcherFactory>();
-                    services.AddSingleton<IConsumerFactory, AudioFeatureInjectorFactory>();
+                    services.AddSingleton<IAudioFeatureInjectorFactory, AudioFeatureInjectorFactory>();
                     services.AddSingleton<IWatcherCollectionFactory, WatcherCollectionFactory>();
                     // For generating spotify clients
                     //services.AddSingleton<IRefreshTokenFactoryProvider, RefreshTokenFactoryProvider>();
@@ -68,6 +68,7 @@ namespace Selector.CLI
                         var connMulti = ConnectionMultiplexer.Connect(config.RedisOptions.ConnectionString);
                         services.AddSingleton(connMulti);
                         services.AddTransient<IDatabaseAsync>(services => services.GetService<ConnectionMultiplexer>().GetDatabase());
+                        services.AddTransient<ISubscriber>(services => services.GetService<ConnectionMultiplexer>().GetSubscriber());
                     }
 
                     // EQUAL

Selector.CLI/WatcherService.cs

@@ -10,6 +10,9 @@ using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
+using Selector.Cache;
+using StackExchange.Redis;
+
 namespace Selector.CLI
 {
     class WatcherService : IHostedService
@@ -23,6 +26,9 @@ namespace Selector.CLI
         private readonly IWatcherCollectionFactory WatcherCollectionFactory;
         private readonly IRefreshTokenFactoryProvider SpotifyFactory;
         
+        private readonly IDatabaseAsync Cache; 
+        private readonly ISubscriber Subscriber; 
+
         private Dictionary<string, IWatcherCollection> Watchers { get; set; } = new();
 
         public WatcherService(
@@ -30,7 +36,9 @@ namespace Selector.CLI
             IWatcherCollectionFactory watcherCollectionFactory,
             IRefreshTokenFactoryProvider spotifyFactory,
             ILoggerFactory loggerFactory,
-            IOptions<RootOptions> config
+            IOptions<RootOptions> config,
+            IDatabaseAsync cache = null,
+            ISubscriber subscriber = null
         ) {
             Logger = loggerFactory.CreateLogger<WatcherService>();
             LoggerFactory = loggerFactory;
@@ -38,6 +46,8 @@ namespace Selector.CLI
             WatcherFactory = watcherFactory;
             WatcherCollectionFactory = watcherCollectionFactory;
             SpotifyFactory = spotifyFactory;
+            Cache = cache;
+            Subscriber = subscriber;
 
             SpotifyFactory.Initialise(Config.ClientId, Config.ClientSecret);
         }
@@ -100,8 +110,18 @@ namespace Selector.CLI
                     switch(consumer)
                     {
                         case Consumers.AudioFeatures:
-                            var factory = new AudioFeatureInjectorFactory(LoggerFactory);
+                            var featureInjector = new AudioFeatureInjectorFactory(LoggerFactory);
-                            consumers.Add(await factory.Get(spotifyFactory));
+                            consumers.Add(await featureInjector.Get(spotifyFactory));
+                            break;
+
+                        case Consumers.CacheWriter:
+                            var cacheWriter = new CacheWriterFactory(Cache, LoggerFactory);
+                            consumers.Add(await cacheWriter.Get());
+                            break;
+
+                        case Consumers.Publisher:
+                            var pub = new PublisherFactory(Subscriber, LoggerFactory);
+                            consumers.Add(await pub.Get());
                             break;
                     }
                 }

Selector.CLI/appsettings.json

@@ -9,12 +9,12 @@
                     "name": "Player Watcher",
                     "type": "player",
                     "pollperiod": 2000,
-                    "consumers": [ "audiofeatures" ]
+                    "consumers": [ "audiofeatures", "cachewriter" ]
                 }
             ]
         },
         "Database": {
-            "enabled": true
+            "enabled": false
         },
         "Redis": {
             "enabled": true

Selector.Cache/CacheWriterConsumer.cs

@@ -1,28 +1,32 @@
 using System;
 using System.Collections.Generic;
-using System.Text;
+using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using SpotifyAPI.Web;
+using StackExchange.Redis;
 
 namespace Selector.Cache
 {
-    public class CacheUpdater : IConsumer
+    public class CacheWriter : IConsumer
     {
         private readonly IPlayerWatcher Watcher;
-        private readonly ILogger<CacheUpdater> Logger;
+        private readonly IDatabaseAsync Db;
+        private readonly ILogger<CacheWriter> Logger;
 
         public CancellationToken CancelToken { get; set; }
 
-        public CacheUpdater(
+        public CacheWriter(
             IPlayerWatcher watcher,
-            ILogger<CacheUpdater> logger = null,
+            IDatabaseAsync db,
+            ILogger<CacheWriter> logger = null,
             CancellationToken token = default
         ){
             Watcher = watcher;
-            Logger = logger ?? NullLogger<CacheUpdater>.Instance;
+            Db = db;
+            Logger = logger ?? NullLogger<CacheWriter>.Instance;
             CancelToken = token;
         }
 
@@ -35,18 +39,8 @@ namespace Selector.Cache
 
         public async Task AsyncCallback(ListeningChangeEventArgs e)
         {
-            if (e.Current.Item is FullTrack track)
+            var payload = JsonSerializer.Serialize(e);
-            {
+            await Db.StringSetAsync(Key.CurrentlyPlaying(e.Username), payload);
-                
-            }
-            else if (e.Current.Item is FullEpisode episode)
-            {
-                
-            }
-            else
-            {
-                Logger.LogError($"Unknown item pulled from API [{e.Current.Item}]");
-            }
         }
 
         public void Subscribe(IWatcher watch = null)

Selector.Cache/Factory/CacheWriterFactory.cs

@@ -0,0 +1,37 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+
+using StackExchange.Redis;
+
+namespace Selector.Cache
+{
+    public interface ICacheWriterFactory {
+        public Task<IConsumer> Get(IPlayerWatcher watcher = null);
+    }
+
+    public class CacheWriterFactory: ICacheWriterFactory {
+
+        private readonly ILoggerFactory LoggerFactory;
+        private readonly IDatabaseAsync Cache;
+
+        public CacheWriterFactory(
+            IDatabaseAsync cache, 
+            ILoggerFactory loggerFactory
+        ) {
+            Cache = cache;
+            LoggerFactory = loggerFactory;
+        }
+
+        public async Task<IConsumer> Get(IPlayerWatcher watcher = null)
+        {
+            return new CacheWriter(
+                watcher,
+                Cache,
+                LoggerFactory.CreateLogger<CacheWriter>()
+            );
+        }
+    }
+}

Selector.Cache/Factory/PublisherFactory.cs

@@ -0,0 +1,37 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+
+using StackExchange.Redis;
+
+namespace Selector.Cache
+{
+    public interface IPublisherFactory {
+        public Task<IConsumer> Get(IPlayerWatcher watcher = null);
+    }
+
+    public class PublisherFactory: IPublisherFactory {
+
+        private readonly ILoggerFactory LoggerFactory;
+        private readonly ISubscriber Subscriber;
+
+        public PublisherFactory(
+            ISubscriber subscriber, 
+            ILoggerFactory loggerFactory
+        ) {
+            Subscriber = subscriber;
+            LoggerFactory = loggerFactory;
+        }
+
+        public async Task<IConsumer> Get(IPlayerWatcher watcher = null)
+        {
+            return new Publisher(
+                watcher,
+                Subscriber,
+                LoggerFactory.CreateLogger<Publisher>()
+            );
+        }
+    }
+}

Selector.Cache/PublisherConsumer.cs

@@ -0,0 +1,74 @@
+using System;
+using System.Collections.Generic;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using SpotifyAPI.Web;
+using StackExchange.Redis;
+
+namespace Selector.Cache
+{
+    public class Publisher : IConsumer
+    {
+        private readonly IPlayerWatcher Watcher;
+        private readonly ISubscriber Subscriber;
+        private readonly ILogger<Publisher> Logger;
+
+        public CancellationToken CancelToken { get; set; }
+
+        public Publisher(
+            IPlayerWatcher watcher,
+            ISubscriber subscriber,
+            ILogger<Publisher> logger = null,
+            CancellationToken token = default
+        ){
+            Watcher = watcher;
+            Subscriber = subscriber;
+            Logger = logger ?? NullLogger<Publisher>.Instance;
+            CancelToken = token;
+        }
+
+        public void Callback(object sender, ListeningChangeEventArgs e)
+        {
+            if (e.Current is null) return;
+
+            Task.Run(() => { return AsyncCallback(e); }, CancelToken);
+        }
+
+        public async Task AsyncCallback(ListeningChangeEventArgs e)
+        {
+            var payload = JsonSerializer.Serialize(e);
+            await Subscriber.PublishAsync(Key.CurrentlyPlaying(e.Username), payload);
+        }
+
+        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.Web/Startup.cs

@@ -110,6 +110,7 @@ namespace Selector.Web
                 var connMulti = ConnectionMultiplexer.Connect(config.RedisOptions.ConnectionString);
                 services.AddSingleton(connMulti);
                 services.AddTransient<IDatabaseAsync>(services => services.GetService<ConnectionMultiplexer>().GetDatabase());
+                services.AddTransient<ISubscriber>(services => services.GetService<ConnectionMultiplexer>().GetSubscriber());
             }
         }
 

Selector/Consumers/AudioFeatureInjectorFactory.cs

@@ -8,7 +8,12 @@ using SpotifyAPI.Web;
 
 namespace Selector
 {
-    public class AudioFeatureInjectorFactory: IConsumerFactory {
+    public interface IAudioFeatureInjectorFactory
+    {
+        public Task<IConsumer> Get(ISpotifyConfigFactory spotifyFactory, IPlayerWatcher watcher);
+    }
+    
+    public class AudioFeatureInjectorFactory: IAudioFeatureInjectorFactory {
 
         private readonly ILoggerFactory LoggerFactory;
 

Selector/Consumers/IConsumerFactory.cs

@@ -1,10 +0,0 @@
-using System;
-using System.Threading.Tasks;
-
-namespace Selector
-{
-    public interface IConsumerFactory
-    {
-        public Task<IConsumer> Get(ISpotifyConfigFactory spotifyFactory, IPlayerWatcher watcher);
-    }
-}

Selector/Watcher/Interfaces/Events.cs

@@ -6,13 +6,15 @@ namespace Selector
     public class ListeningChangeEventArgs: EventArgs {
         public CurrentlyPlayingContext Previous;
         public CurrentlyPlayingContext Current;
+        public string Username;
 
-        public static ListeningChangeEventArgs From(CurrentlyPlayingContext previous, CurrentlyPlayingContext current)
+        public static ListeningChangeEventArgs From(CurrentlyPlayingContext previous, CurrentlyPlayingContext current, string username = null)
         {
             return new ListeningChangeEventArgs()
             {
                 Previous = previous,
-                Current = current
+                Current = current,
+                Username = username
             };
         }
     }

Selector/Watcher/PlayerWatcher.cs

@@ -74,14 +74,14 @@ namespace Selector
                         && (Live.Item is FullTrack || Live.Item is FullEpisode))
                     {
                         Logger.LogDebug($"Playback started: {Live.DisplayString()}");
-                        OnPlayingChange(ListeningChangeEventArgs.From(previous, Live));
+                        OnPlayingChange(ListeningChangeEventArgs.From(previous, Live, Username));
                     }
                     // STOPPED PLAYBACK
                     else if((previous.Item is FullTrack || previous.Item is FullEpisode) 
                         && Live is null)
                     {
                         Logger.LogDebug($"Playback stopped: {previous.DisplayString()}");
-                        OnPlayingChange(ListeningChangeEventArgs.From(previous, Live));
+                        OnPlayingChange(ListeningChangeEventArgs.From(previous, Live, Username));
                     }
                     // CONTINUING PLAYBACK
                     else {
@@ -92,17 +92,17 @@ namespace Selector
                         {
                             if(!eq.IsEqual(previousTrack, currentTrack)) {
                                 Logger.LogDebug($"Track changed: {previousTrack.DisplayString()} -> {currentTrack.DisplayString()}");
-                                OnItemChange(ListeningChangeEventArgs.From(previous, Live));
+                                OnItemChange(ListeningChangeEventArgs.From(previous, Live, Username));
                             }
 
                             if(!eq.IsEqual(previousTrack.Album, currentTrack.Album)) {
                                 Logger.LogDebug($"Album changed: {previousTrack.Album.DisplayString()} -> {currentTrack.Album.DisplayString()}");
-                                OnAlbumChange(ListeningChangeEventArgs.From(previous, Live));
+                                OnAlbumChange(ListeningChangeEventArgs.From(previous, Live, Username));
                             }
 
                             if(!eq.IsEqual(previousTrack.Artists[0], currentTrack.Artists[0])) {
                                 Logger.LogDebug($"Artist changed: {previousTrack.Artists.DisplayString()} -> {currentTrack.Artists.DisplayString()}");
-                                OnArtistChange(ListeningChangeEventArgs.From(previous, Live));
+                                OnArtistChange(ListeningChangeEventArgs.From(previous, Live, Username));
                             }
                         }
                         // CHANGED CONTENT
@@ -110,8 +110,8 @@ namespace Selector
                             || (previous.Item is FullEpisode && Live.Item is FullTrack))
                         {
                             Logger.LogDebug($"Media type changed: {previous.Item}, {previous.Item}");
-                            OnContentChange(ListeningChangeEventArgs.From(previous, Live));
+                            OnContentChange(ListeningChangeEventArgs.From(previous, Live, Username));
-                            OnItemChange(ListeningChangeEventArgs.From(previous, Live));
+                            OnItemChange(ListeningChangeEventArgs.From(previous, Live, Username));
                         }
                         // PODCASTS
                         else if(previous.Item is FullEpisode previousEp 
@@ -119,7 +119,7 @@ namespace Selector
                         {
                             if(!eq.IsEqual(previousEp, currentEp)) {
                                 Logger.LogDebug($"Podcast changed: {previousEp.DisplayString()} -> {currentEp.DisplayString()}");
-                                OnItemChange(ListeningChangeEventArgs.From(previous, Live));
+                                OnItemChange(ListeningChangeEventArgs.From(previous, Live, Username));
                             }
                         }
                         else {
@@ -129,25 +129,25 @@ namespace Selector
                         // CONTEXT
                         if(!eq.IsEqual(previous.Context, Live.Context)) {
                             Logger.LogDebug($"Context changed: {previous.Context.DisplayString()} -> {Live.Context.DisplayString()}");
-                            OnContextChange(ListeningChangeEventArgs.From(previous, Live));
+                            OnContextChange(ListeningChangeEventArgs.From(previous, Live, Username));
                         }
 
                         // DEVICE
                         if(!eq.IsEqual(previous?.Device, Live?.Device)) {
                             Logger.LogDebug($"Device changed: {previous?.Device.DisplayString()} -> {Live?.Device.DisplayString()}");
-                            OnDeviceChange(ListeningChangeEventArgs.From(previous, Live));
+                            OnDeviceChange(ListeningChangeEventArgs.From(previous, Live, Username));
                         }
 
                         // IS PLAYING
                         if(previous.IsPlaying != Live.IsPlaying) {
                             Logger.LogDebug($"Playing state changed: {previous.IsPlaying} -> {Live.IsPlaying}");
-                            OnPlayingChange(ListeningChangeEventArgs.From(previous, Live));
+                            OnPlayingChange(ListeningChangeEventArgs.From(previous, Live, Username));
                         }
 
                         // VOLUME
                         if(previous.Device.VolumePercent != Live.Device.VolumePercent) {
                             Logger.LogDebug($"Volume changed: {previous.Device.VolumePercent}% -> {Live.Device.VolumePercent}%");
-                            OnVolumeChange(ListeningChangeEventArgs.From(previous, Live));
+                            OnVolumeChange(ListeningChangeEventArgs.From(previous, Live, Username));
                         }
                     }
                 }