Add UpdateResults Functionality

Author: taooceros

Flow.Launcher.Core/Plugin/JsonRPCPluginV2.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 using Flow.Launcher.Core.Plugin.JsonRPCV2Models;
@@ -10,16 +11,18 @@
 
 namespace Flow.Launcher.Core.Plugin
 {
-    internal abstract class JsonRPCPluginV2 : JsonRPCPluginBase, IDisposable
+    internal abstract class JsonRPCPluginV2 : JsonRPCPluginBase, IAsyncDisposable, IAsyncReloadable, IResultUpdated
     {
         public abstract string SupportedLanguage { get; set; }
 
         public const string JsonRpc = "JsonRPC";
 
-        protected abstract JsonRpc RPC { get; set; }
+        protected abstract IDuplexPipe ClientPipe { get; set; }
 
         protected StreamReader ErrorStream { get; set; }
 
+        private JsonRpc RPC { get; set; }
+
 
         protected override async Task<bool> ExecuteResultAsync(JsonRPCResult result)
         {
@@ -59,6 +62,8 @@ public override async Task InitAsync(PluginInitContext context)
         {
             await base.InitAsync(context);
 
+            SetupJsonRPC();
+
             _ = ReadErrorAsync();
 
             await RPC.InvokeAsync("initialize", context);
@@ -74,10 +79,40 @@ async Task ReadErrorAsync()
             }
         }
 
-        public void Dispose()
+        public event ResultUpdatedEventHandler ResultsUpdated;
+
+
+        private void SetupJsonRPC()
+        {
+            var formatter = new JsonMessageFormatter();
+            var handler = new NewLineDelimitedMessageHandler(ClientPipe,
+                formatter);
+
+            RPC = new JsonRpc(handler, new JsonRPCPublicAPI(Context.API));
+
+            RPC.AddLocalRpcMethod("UpdateResults", new Action<string, JsonRPCQueryResponseModel>((rawQuery, response) =>
+            {
+                var results = ParseResults(response);
+                ResultsUpdated?.Invoke(this, new ResultUpdatedEventArgs { Query = new Query()
+                {
+                    RawQuery = rawQuery
+                }, Results = results });
+            }));
+            RPC.SynchronizationContext = null;
+            RPC.StartListening();
+        }
+
+        public virtual Task ReloadDataAsync()
+        {
+            SetupJsonRPC();
+            return Task.CompletedTask;
+        }
+
+        public virtual ValueTask DisposeAsync()
         {
             RPC?.Dispose();
             ErrorStream?.Dispose();
+            return ValueTask.CompletedTask;
         }
     }
 }

Flow.Launcher.Core/Plugin/PythonPluginV2.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
+using System.IO.Pipelines;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -10,18 +11,19 @@
 using Flow.Launcher.Infrastructure;
 using Flow.Launcher.Plugin;
 using Microsoft.VisualStudio.Threading;
+using Nerdbank.Streams;
 using StreamJsonRpc;
 
 namespace Flow.Launcher.Core.Plugin
 {
-    internal class PythonPluginV2 : JsonRPCPluginV2, IReloadable, IDisposable
+    internal class PythonPluginV2 : JsonRPCPluginV2
     {
         private readonly ProcessStartInfo _startInfo;
         private Process _process;
 
         public override string SupportedLanguage { get; set; } = AllowedLanguage.Python;
 
-        protected override JsonRpc RPC { get; set; }
+        protected override IDuplexPipe ClientPipe { get; set; }
 
 
         public PythonPluginV2(string filename)
@@ -61,43 +63,38 @@ public override async Task InitAsync(PluginInitContext context)
             _startInfo.WorkingDirectory = context.CurrentPluginMetadata.PluginDirectory;
 
             _process = Process.Start(_startInfo);
-
             ArgumentNullException.ThrowIfNull(_process);
-
-            SetupJsonRPC(_process, context.API);
+            
+            SetupPipe(_process);
 
             await base.InitAsync(context);
         }
 
-        public void Dispose()
+        public override async ValueTask DisposeAsync()
         {
             _process.Kill(true);
+            await _process.WaitForExitAsync();
             _process.Dispose();
-            base.Dispose();
+            await base.DisposeAsync();
         }
 
-        public void ReloadData()
+        private void SetupPipe(Process process)
+        {
+            var (reader, writer) = (PipeReader.Create(process.StandardOutput.BaseStream),
+                PipeWriter.Create(process.StandardInput.BaseStream));
+            ClientPipe = new DuplexPipe(reader, writer);
+        }
+        
+        public override async Task ReloadDataAsync()
         {
             var oldProcess = _process;
             _process = Process.Start(_startInfo);
             ArgumentNullException.ThrowIfNull(_process);
-            SetupJsonRPC(_process, Context.API);
+            SetupPipe(_process);
+            await base.ReloadDataAsync();
             oldProcess.Kill(true);
+            await oldProcess.WaitForExitAsync();
             oldProcess.Dispose();
         }
-
-        private void SetupJsonRPC(Process process, IPublicAPI api)
-        {
-            var formatter = new JsonMessageFormatter();
-            var handler = new NewLineDelimitedMessageHandler(process.StandardInput.BaseStream,
-                process.StandardOutput.BaseStream,
-                formatter);
-
-            ErrorStream = _process.StandardError;
-
-            RPC = new JsonRpc(handler, new JsonRPCPublicAPI(api));
-            RPC.SynchronizationContext = null;
-            RPC.StartListening();
-        }
     }
 }