Selector/Selector.Cache/Consumer/PublisherConsumer.cs

92 lines
2.8 KiB
C#
Raw Normal View History

using System;
using System.Collections.Generic;
2021-10-28 23:05:07 +01:00
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
2021-10-28 23:05:07 +01:00
using StackExchange.Redis;
namespace Selector.Cache
{
2022-06-28 07:30:27 +01:00
public class Publisher : IPlayerConsumer
{
private readonly IPlayerWatcher Watcher;
2021-10-28 23:05:07 +01:00
private readonly ISubscriber Subscriber;
private readonly ILogger<Publisher> Logger;
public CancellationToken CancelToken { get; set; }
2021-10-28 23:05:07 +01:00
public Publisher(
IPlayerWatcher watcher,
2021-10-28 23:05:07 +01:00
ISubscriber subscriber,
ILogger<Publisher> logger = null,
CancellationToken token = default
){
Watcher = watcher;
2021-10-28 23:05:07 +01:00
Subscriber = subscriber;
Logger = logger ?? NullLogger<Publisher>.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 async Task AsyncCallback(ListeningChangeEventArgs e)
{
2022-07-01 22:04:17 +01:00
using var scope = Logger.GetListeningEventArgsScope(e);
var payload = JsonSerializer.Serialize((CurrentlyPlayingDTO) e, JsonContext.Default.CurrentlyPlayingDTO);
2022-07-01 22:04:17 +01:00
Logger.LogTrace("Publishing current");
2021-11-05 07:58:48 +00:00
// TODO: currently using spotify username for cache key, use db username
var receivers = await Subscriber.PublishAsync(Key.CurrentlyPlaying(e.Id), payload);
2022-07-01 22:04:17 +01:00
Logger.LogDebug("Published current, {receivers} receivers", receivers);
}
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");
}
}
}
}