diff --git a/Flow.Launcher.Core/Plugin/JsonRPCPluginV2.cs b/Flow.Launcher.Core/Plugin/JsonRPCPluginV2.cs index ee7009f9a..c6ef870c3 100644 --- a/Flow.Launcher.Core/Plugin/JsonRPCPluginV2.cs +++ b/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 @@ using StreamJsonRpc; 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 ExecuteResultAsync(JsonRPCResult result) { @@ -59,6 +62,8 @@ namespace Flow.Launcher.Core.Plugin { await base.InitAsync(context); + SetupJsonRPC(); + _ = ReadErrorAsync(); await RPC.InvokeAsync("initialize", context); @@ -74,10 +79,40 @@ namespace Flow.Launcher.Core.Plugin } } - 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((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; } } } diff --git a/Flow.Launcher.Core/Plugin/PythonPluginV2.cs b/Flow.Launcher.Core/Plugin/PythonPluginV2.cs index 83dce2187..98b55f896 100644 --- a/Flow.Launcher.Core/Plugin/PythonPluginV2.cs +++ b/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.Core.Plugin.JsonRPCV2Models; 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 @@ namespace Flow.Launcher.Core.Plugin _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(); - } } }