Selector/Selector.Cache/Consumer/PublisherConsumer.cs

80 lines
2.4 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
{
2021-10-28 23:05:07 +01:00
public class Publisher : IConsumer
{
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(() => { return AsyncCallback(e); }, CancelToken);
}
public async Task AsyncCallback(ListeningChangeEventArgs e)
{
var payload = JsonSerializer.Serialize((CurrentlyPlayingDTO) e);
Logger.LogTrace($"Publishing current for [{e.Username}]");
var receivers = await Subscriber.PublishAsync(Key.CurrentlyPlaying(e.Username), payload);
Logger.LogDebug($"Published current for [{e.Username}], {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");
}
}
}
}