client: Add support for streams.

Author: freezy

VisualPinball.Engine.Mpf.Test/Program.cs

@@ -12,14 +12,15 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Linq.Expressions;
 using System.Threading.Tasks;
 using VisualPinball.Engine.Mpf;
 
 namespace MpfTest
 {
 	public static class Program
 	{
-		public static void Main(string[] args)
+		public static async Task Main(string[] args)
 		{
 			// Console.WriteLine("Starting...");
 			// var client = new MpfClient();
@@ -42,6 +43,21 @@ public static void Main(string[] args)
 
 			var descr = mpfApi.GetMachineDescription();
 			Console.WriteLine($"Description: {descr} in {s.ElapsedMilliseconds}ms");
+
+			ConsoleKeyInfo key;
+			do {
+				key = Console.ReadKey();
+				switch (key.Key) {
+					case ConsoleKey.A:
+						await mpfApi.Switch("s_sling", true);
+						break;
+					case ConsoleKey.S:
+						await mpfApi.Switch("s_sling", false);
+						break;
+				}
+			} while (key.Key != ConsoleKey.Escape);
+
+			mpfApi.Dispose();
 		}
 	}
 }

VisualPinball.Engine.Mpf/MpfApi.cs

@@ -55,6 +55,11 @@ public MachineDescription GetMachineDescription()
 			return _client.GetMachineDescription();
 		}
 
+		public async Task Switch(string swName, bool swValue)
+		{
+			await _client.Switch(swName, swValue);
+		}
+
 		public void Dispose()
 		{
 			_client?.Shutdown();

VisualPinball.Engine.Mpf/MpfClient.cs

@@ -12,8 +12,8 @@
 using Grpc.Core;
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
-using Grpc.Core.Utils;
 using Mpf.Vpe;
 using NLog;
 
@@ -35,7 +35,9 @@ public class MpfClient
 		private MpfHardwareService.MpfHardwareServiceClient _client;
 
 		private static readonly Logger Logger = LogManager.GetCurrentClassLogger();
-
+		private Thread _commandsThread;
+		private AsyncServerStreamingCall<Commands> _commandStream;
+		private AsyncClientStreamingCall<SwitchChanges, EmptyResponse> _switchStream;
 
 		public void Connect(string serverIpPort = "127.0.0.1:50051")
 		{
@@ -51,42 +53,54 @@ public void StartGame(Dictionary<string, bool> initialSwitches)
 				ms.InitialSwitchStates.Add(sw, initialSwitches[sw]);
 			}
 
-			Logger.Info("Starting client...");
-			using (var call = _client.Start(ms)) {
+			_commandsThread =  new Thread(() => ReceiveCommands(ms)) { IsBackground = true };
+			_commandsThread.Start();
 
-				Logger.Info("Client started, retrieving commands...");
-				var count = 0;
-				call.ResponseStream.ForEachAsync(commands => {
+			_switchStream = _client.SendSwitchChanges();
+		}
+
+		public async Task Switch(string swName, bool swValue)
+		{
+			await _switchStream.RequestStream.WriteAsync(new SwitchChanges
+				{SwitchNumber = swName, SwitchState = swValue});
+		}
+
+		private async void ReceiveCommands(MachineState ms)
+		{
+			Logger.Info("Starting client...");
+			_commandStream = _client.Start(ms);
 
-					Logger.Info($"New command: {commands.CommandCase}");
-					count++;
-					switch (commands.CommandCase) {
-						case Commands.CommandOneofCase.None:
-							break;
-						case Commands.CommandOneofCase.FadeLight:
-							OnFadeLight?.Invoke(this, commands.FadeLight);
-							break;
-						case Commands.CommandOneofCase.PulseCoil:
-							OnPulseCoil?.Invoke(this, commands.PulseCoil);
-							break;
-						case Commands.CommandOneofCase.EnableCoil:
-							OnEnableCoil?.Invoke(this, commands.EnableCoil);
-							break;
-						case Commands.CommandOneofCase.DisableCoil:
-							OnDisableCoil?.Invoke(this, commands.DisableCoil);
-							break;
-						case Commands.CommandOneofCase.ConfigureHardwareRule:
-							OnConfigureHardwareRule?.Invoke(this, commands.ConfigureHardwareRule);
-							break;
-						case Commands.CommandOneofCase.RemoveHardwareRule:
-							OnRemoveHardwareRule?.Invoke(this, commands.RemoveHardwareRule);
-							break;
-						default:
-							throw new ArgumentOutOfRangeException();
-					}
-					return Task.CompletedTask;
-				}).Wait();
+			Logger.Info("Client started, retrieving commands...");
+			var count = 0;
+			while (await _commandStream.ResponseStream.MoveNext()) {
 
+				var commands = _commandStream.ResponseStream.Current;
+				Logger.Info($"New command: {commands.CommandCase}");
+				count++;
+				switch (commands.CommandCase) {
+					case Commands.CommandOneofCase.None:
+						break;
+					case Commands.CommandOneofCase.FadeLight:
+						OnFadeLight?.Invoke(this, commands.FadeLight);
+						break;
+					case Commands.CommandOneofCase.PulseCoil:
+						OnPulseCoil?.Invoke(this, commands.PulseCoil);
+						break;
+					case Commands.CommandOneofCase.EnableCoil:
+						OnEnableCoil?.Invoke(this, commands.EnableCoil);
+						break;
+					case Commands.CommandOneofCase.DisableCoil:
+						OnDisableCoil?.Invoke(this, commands.DisableCoil);
+						break;
+					case Commands.CommandOneofCase.ConfigureHardwareRule:
+						OnConfigureHardwareRule?.Invoke(this, commands.ConfigureHardwareRule);
+						break;
+					case Commands.CommandOneofCase.RemoveHardwareRule:
+						OnRemoveHardwareRule?.Invoke(this, commands.RemoveHardwareRule);
+						break;
+					default:
+						throw new ArgumentOutOfRangeException();
+				}
 				Logger.Info($"{count} commands dispatched.");
 			}
 		}
@@ -98,6 +112,7 @@ public MachineDescription GetMachineDescription()
 		}
 
 		public void Shutdown() {
+			_commandStream.Dispose();
 			_channel.ShutdownAsync().Wait();
 		}
 	}