Notification sound for channel points WORKING!

Author: csharpfritz

Fritz.StreamTools/Services/TwitchPubSubService.cs

@@ -3,6 +3,7 @@
 using Fritz.Twitch.PubSub;
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Options;
 using System;
@@ -16,13 +17,13 @@ namespace Fritz.StreamTools.Services
 	public class TwitchPubSubService : IHostedService
 	{
 
-		private readonly IHubContext<AttentionHub, IAttentionHubClient> _HubContext;
+		private IServiceProvider _ServiceProvider;
 		private readonly Twitch.PubSub.Proxy _Proxy;
 		private readonly ConfigurationSettings _Configuration;
 
-		public TwitchPubSubService(IHubContext<AttentionHub, IAttentionHubClient> hubContext, Twitch.PubSub.Proxy proxy, IOptions<ConfigurationSettings> settings)
+		public TwitchPubSubService(IServiceProvider serviceProvider, Twitch.PubSub.Proxy proxy, IOptions<ConfigurationSettings> settings)
 		{
-			_HubContext = hubContext;
+			_ServiceProvider = serviceProvider;
 			_Proxy = proxy;
 			_Configuration = settings.Value;
 		}
@@ -35,7 +36,11 @@ public Task StartAsync(CancellationToken cancellationToken)
 
 		private void _Proxy_OnChannelPointsRedeemed(object sender, ChannelRedemption e)
 		{
-			_HubContext.Clients.All.PlaySoundEffect("pointsredeemed.mp3");
+			using (var scope = _ServiceProvider.CreateScope())
+			{
+				var context = scope.ServiceProvider.GetRequiredService<IHubContext<AttentionHub, IAttentionHubClient>>();
+				context.Clients.All.PlaySoundEffect("pointsredeemed.mp3");
+			}
 		}
 
 		public Task StopAsync(CancellationToken cancellationToken)

Fritz.StreamTools/StartupServices/ConfigureServices.cs

@@ -22,13 +22,13 @@ namespace Fritz.StreamTools.StartupServices
 {
   public static class ConfigureServices
 	{
-		private static Dictionary<Type, string[]> _servicesRequiredConfiguration;
-		private static IConfiguration _configuration;
+		private static Dictionary<Type, string[]> _ServicesRequiredConfiguration;
+		private static IConfiguration _Configuration;
 
 		public static void Execute(IServiceCollection services, IConfiguration configuration, Dictionary<Type, string[]> servicesRequiredConfiguration)
 		{
-			_configuration = configuration;
-			_servicesRequiredConfiguration = servicesRequiredConfiguration;
+			_Configuration = configuration;
+			_ServicesRequiredConfiguration = servicesRequiredConfiguration;
 
 			services.AddSingleton<RundownItemRepository>();
 			services.AddSingleton<RundownRepository>();
@@ -56,13 +56,17 @@ public static void Execute(IServiceCollection services, IConfiguration configura
 
 			services.AddLazyCache();
 
+			services.RegisterTwitchPubSub();
+
 			RegisterConfiguredServices(services, configuration);
 			RegisterGitHubServices(services, configuration);
+
+
 		}
 
 		private static void RegisterConfiguredServices(IServiceCollection services, IConfiguration configuration)
 		{
-			foreach (var configuredService in _servicesRequiredConfiguration)
+			foreach (var configuredService in _ServicesRequiredConfiguration)
 			{
 				if (!configuredService.Value.Any(cs => configuration[cs] == null))
 				{
@@ -78,7 +82,7 @@ private static void RegisterGitHubServices(IServiceCollection services, IConfigu
 
 			services.AddTransient(_ => new GitHubClient(new ProductHeaderValue("Fritz.StreamTools"))
 			{
-				Credentials = new Credentials(_configuration["GitHub:User"], _configuration["GitHub:AuthenticationToken"])
+				Credentials = new Credentials(_Configuration["GitHub:User"], _Configuration["GitHub:AuthenticationToken"])
 			});
 
 			services.AddHttpClient("GitHub", c =>
@@ -138,10 +142,6 @@ private static void AddStreamService<TStreamService>(this IServiceCollection ser
 				return;
 			}
 
-			if (typeof(TStreamService) == typeof(TwitchService)) {
-				services.AddSingleton<Twitch.PubSub.Proxy>();
-			}
-
 			// Configure and grab a logger so that we can log information
 			// about the creation of the services
 			var provider = services.BuildServiceProvider();   // Build a 'temporary' instance of the DI container
@@ -153,17 +153,6 @@ private static void AddStreamService<TStreamService>(this IServiceCollection ser
 			services.AddSingleton(service as IStreamService);
 			services.AddSingleton(service);
 
-			if (typeof(TStreamService) == typeof(TwitchService))
-			{
-				var pubSub = new TwitchPubSubService(
-				provider.GetRequiredService<IHubContext<AttentionHub, IAttentionHubClient>>(),
-				provider.GetRequiredService<Twitch.PubSub.Proxy>(),
-				provider.GetRequiredService<IOptions<Twitch.ConfigurationSettings>>());
-				services.AddSingleton(pubSub as IHostedService);
-				services.AddSingleton(pubSub);
-			}
-
-
 			if (service is IChatService chatService)
 			{
 				services.AddSingleton(chatService);
@@ -190,5 +179,24 @@ private static void AddAspNetFeatures(this IServiceCollection services)
 
 		}
 
+		private static void RegisterTwitchPubSub(this IServiceCollection services) {
+
+			services.AddSingleton<Twitch.PubSub.Proxy>();
+			services.AddHostedService<TwitchPubSubService>();
+			//var provider = services.BuildServiceProvider();
+
+			//var pubSub = new TwitchPubSubService(
+			//provider,
+			//provider.GetRequiredService<Twitch.PubSub.Proxy>(),
+			//provider.GetRequiredService<IOptions<Twitch.ConfigurationSettings>>());
+			//services.AddSingleton(pubSub as IHostedService);
+			//services.AddSingleton(pubSub);
+
+		}
+
+		private static bool IsTwitchEnabled {
+			get { return string.IsNullOrEmpty(_Configuration["StreamServices:Twitch:ClientId"]); }
+		}
+
 	}
 }

Fritz.Twitch/PubSub/ChannelRedemption.cs

@@ -3,6 +3,8 @@
 namespace Fritz.Twitch.PubSub
 {
 
+	public class PubSubRedemptionMessage : PubSubMessage<ChannelRedemption> { }
+
 	public class ChannelRedemption : EventArgs, IPubSubData
 	{
 		public string timestamp { get; set; }

Fritz.Twitch/PubSub/Proxy.cs

@@ -21,7 +21,8 @@ public class Proxy : IDisposable
 	{
 
 		private ClientWebSocket _Socket;
-		private readonly System.Timers.Timer _PingTimer;
+		private System.Timers.Timer _PingTimer;
+		private System.Timers.Timer _PongTimer;
 		private ConfigurationSettings _Configuration;
 		private ILogger _Logger;
 		private static bool _Reconnect;
@@ -34,29 +35,52 @@ public Proxy(IOptions<ConfigurationSettings> settings, ILoggerFactory loggerFact
 			_Configuration = settings.Value;
 			_Logger = loggerFactory.CreateLogger("TwitchPubSub");
 
-			_Socket = new ClientWebSocket();
-
-			// Start a timer to manage the connection over the websocket
-			_PingTimer = new System.Timers.Timer(TimeSpan.FromMinutes(4).TotalMilliseconds);
-			_PingTimer.Elapsed += _PingTimer_Elapsed;
-			_PingTimer.Start();
-
 		}
 
 		public async Task StartAsync(IEnumerable<TwitchTopic> topics, CancellationToken token)
 		{
 
 			_Topics = topics;
-			await StartListening(topics);
-			var messageBuffer = new Memory<byte>();
 
+			// Start a timer to manage the connection over the websocket
+			_PingTimer = new System.Timers.Timer(TimeSpan.FromSeconds(30).TotalMilliseconds);
+			_PingTimer.Elapsed += _PingTimer_Elapsed;
+			_PingTimer.Start();
+
+			await StartListening(topics);
 
 			while (!token.IsCancellationRequested)
 			{
 
-				await _Socket.ReceiveAsync(messageBuffer, token);
+				var buffer = new byte[1024];
+				var messageBuffer = new ArraySegment<byte>(buffer);
+				var completeMessage = new StringBuilder();
+
+				var result = await _Socket.ReceiveAsync(messageBuffer, token);
+				completeMessage.Append(Encoding.UTF8.GetString(messageBuffer));
+				while (!result.EndOfMessage)
+				{
+					buffer = new byte[1024];
+					messageBuffer = new ArraySegment<byte>(buffer);
+					result = await _Socket.ReceiveAsync(messageBuffer, token);
+					completeMessage.Append(Encoding.UTF8.GetString(messageBuffer));
+				}
+
+				if (result.MessageType == WebSocketMessageType.Close) {
+					_Reconnect = true;
+					break;
+				}
 
-				HandleMessage(UTF8Encoding.UTF8.GetString(messageBuffer.Span));
+				try
+				{
+					HandleMessage(completeMessage.ToString());
+				} catch (UnhandledPubSubMessageException) {
+					// do nothing
+				} catch (Exception e) {
+					_Logger.LogError(e, "Error while parsing message from Twitch: " + completeMessage.ToString());
+					_Logger.LogError("Reconnecting...");
+					_Reconnect = true;
+				}
 
 				if (_Reconnect) {
 					break;
@@ -75,6 +99,15 @@ public async Task StartAsync(IEnumerable<TwitchTopic> topics, CancellationToken
 		private void HandleMessage(string receivedMessage)
 		{
 
+			var jDoc = JObject.Parse(receivedMessage);
+			var messageType = jDoc["type"].Value<string>();
+			if (messageType == "RESPONSE" && jDoc["error"].Value<string>() != "")
+			{
+				throw new Exception("Unable to connect");
+			} else if (messageType == "RESPONSE") {
+				return;
+			}
+
 			foreach (var strategy in _Strategies)
 			{
 				if (strategy(receivedMessage)) return;
@@ -87,7 +120,7 @@ private void HandleMessage(string receivedMessage)
 		private async Task StartListening(IEnumerable<TwitchTopic> topics)
 		{
 
-			await _Socket.ConnectAsync(new Uri("wss://pubsub-edge.twitch.tv"), CancellationToken.None);
+			_Socket = new ClientWebSocket();
 
 			var message = new PubSubListen
 			{
@@ -98,25 +131,39 @@ private async Task StartListening(IEnumerable<TwitchTopic> topics)
 				}
 			};
 
-			await SendMessageOnSocket(JsonConvert.SerializeObject(message));
+			await _Socket.ConnectAsync(new Uri("wss://pubsub-edge.twitch.tv:443"), CancellationToken.None)
+				.ContinueWith(t => SendMessageOnSocket(JsonConvert.SerializeObject(message)));
 
 		}
 
 		private void _PingTimer_Elapsed(object sender, ElapsedEventArgs e)
 		{
 			var message = @"{ ""type"": ""PING"" }";
 			SendMessageOnSocket(message).GetAwaiter().GetResult();
+			_PongTimer = new System.Timers.Timer(TimeSpan.FromSeconds(10).TotalMilliseconds);
+			_PongTimer.Elapsed += _PongTimer_Elapsed;
+			_PongTimer.Start();
 			_PingAcknowledged = false;
 
 			// TODO: handle the lack of returned PONG message 
 
 		}
 
-		private async Task SendMessageOnSocket(string message)
+		private void _PongTimer_Elapsed(object sender, ElapsedEventArgs e)
 		{
+			if (!_PingAcknowledged) {
+				_Reconnect = true;
+				_PongTimer.Dispose();
+			}
+		}
+
+		private Task SendMessageOnSocket(string message)
+		{
+
+			if (_Socket.State != WebSocketState.Open) return Task.CompletedTask;
 
-			var byteArray = Encoding.UTF8.GetBytes(message);
-			await _Socket.SendAsync(byteArray, WebSocketMessageType.Text, false, CancellationToken.None);
+			var byteArray = Encoding.ASCII.GetBytes(message);
+			return _Socket.SendAsync(byteArray, WebSocketMessageType.Text, true, CancellationToken.None);
 
 		}
 
@@ -140,6 +187,8 @@ private bool HandlePongMessage(string message) {
 
 			if (message.Contains(@"""PONG""")) {
 				_PingAcknowledged = true;
+				_PongTimer.Stop();
+				_PongTimer.Dispose();
 				return true;
 			}
 
@@ -164,11 +213,19 @@ private bool HandleChannelPointsMessage(string message) {
 
 			var jDoc = JObject.Parse(message);
 
-			if (jDoc["type"].Value<string>() == "reward-redeemed") {
+			if (jDoc["type"].Value<string>() == "MESSAGE" && jDoc["data"]["topic"].Value<string>().StartsWith("channel-points-channel-v1") ) {
 
-				var messageObj = JsonConvert.DeserializeObject<PubSubMessage<ChannelRedemption>>(message);
+				var innerMessage = jDoc["data"]["message"].Value<string>();
 
-				OnChannelPointsRedeemed?.BeginInvoke(null, messageObj.data, null, null);
+				PubSubRedemptionMessage messageObj = null;
+				try
+				{
+					messageObj = JsonConvert.DeserializeObject<PubSubRedemptionMessage>(innerMessage);
+				} catch (Exception e) {
+					_Logger.LogError(e, "Error while deserializing the message");
+					_Logger.LogInformation("Message contents: " + innerMessage);
+				}
+				OnChannelPointsRedeemed?.Invoke(null, messageObj?.data);
 				return true;
 
 			}
@@ -191,6 +248,7 @@ protected virtual void Dispose(bool disposing)
 				if (disposing)
 				{
 					_PingTimer.Dispose();
+					_PongTimer.Dispose();
 				}
 
 				_Socket.Dispose();

Fritz.Twitch/PubSub/PubSubListenMessage.cs

@@ -23,4 +23,5 @@ public class PubSubListenData : IPubSubData
 
 	}
 
+
 }

Fritz.Twitch/PubSub/PubSubMessage.cs

@@ -1,11 +1,15 @@
-namespace Fritz.Twitch.PubSub
+using System;
+
+namespace Fritz.Twitch.PubSub
 {
 
 	public abstract class PubSubMessage<DataType> where DataType : IPubSubData
 	{
 
 		public string type { get; protected set; }
 
+		public string nonce { get; set; } = Guid.NewGuid().ToString().Replace("-", "");
+
 		public DataType data { get; set; }
 
 	}