prehistoric-systems/Flow.Launcher
1
0
Fork 0
mirror of https://github.com/Flow-Launcher/Flow.Launcher.git synced 2026-03-11 08:54:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Use Channel instead of BufferBlock

Browse source
This commit is contained in:
弘韬 张 2021-02-15 17:58:15 +08:00
parent 7c3aac47cf
commit 8a2a71000e
1 changed files with 23 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
Flow.Launcher/ViewModel/MainViewModel.cs
View file

@ -19,6 +19,7 @@ using Flow.Launcher.Plugin;
using Flow.Launcher.Plugin.SharedCommands;
using Flow.Launcher.Storage;
using Flow.Launcher.Infrastructure.Logger;
using System.Threading.Channels;
namespace Flow.Launcher.ViewModel
{
@ -46,7 +47,7 @@ namespace Flow.Launcher.ViewModel
private readonly Internationalization _translator = InternationalizationManager.Instance;
private BufferBlock<ResultsForUpdate> _resultsUpdateQueue;
private Channel<ResultsForUpdate> _resultsUpdateQueue;
private Task _resultsViewUpdateTask;
#endregion
@ -85,18 +86,21 @@ namespace Flow.Launcher.ViewModel
private void RegisterViewUpdate()
{
_resultsUpdateQueue = new BufferBlock<ResultsForUpdate>();
_resultsUpdateQueue = Channel.CreateUnbounded<ResultsForUpdate>();
_resultsViewUpdateTask =
Task.Run(updateAction).ContinueWith(continueAction, TaskContinuationOptions.OnlyOnFaulted);
async Task updateAction()
{
var queue = new Dictionary<string, ResultsForUpdate>();
while (await _resultsUpdateQueue.OutputAvailableAsync())
var queueReader = _resultsUpdateQueue.Reader;
// it is not supposed to be false because it won't be complete
while (await queueReader.WaitToReadAsync())
{
queue.Clear();
await Task.Delay(20);
while (_resultsUpdateQueue.TryReceive(out var item))
await foreach (var item in queueReader.ReadAllAsync())
{
if (!item.Token.IsCancellationRequested)
queue[item.ID] = item;
@ -104,6 +108,8 @@ namespace Flow.Launcher.ViewModel
UpdateResultView(queue.Values);
}
Log.Error("MainViewModel", "Unexpected ResultViewUpdate ends");
};
void continueAction(Task t)
@ -128,7 +134,10 @@ namespace Flow.Launcher.ViewModel
if (e.Query.RawQuery == QueryText) // TODO: allow cancellation
{
PluginManager.UpdatePluginMetadata(e.Results, pair.Metadata, e.Query);
_resultsUpdateQueue.Post(new ResultsForUpdate(e.Results, pair.Metadata, e.Query, _updateToken));
if (!_resultsUpdateQueue.Writer.TryWrite(new ResultsForUpdate(e.Results, pair.Metadata, e.Query, _updateToken)))
{
Log.Error("MainViewModel", "Unable to add item to Result Update Queue");
};
}
};
}
@ -529,9 +538,15 @@ namespace Flow.Launcher.ViewModel
await Task.Yield();
var results = await PluginManager.QueryForPlugin(plugin, query, currentCancellationToken);
if (!currentCancellationToken.IsCancellationRequested && results != null)
_resultsUpdateQueue.Post(new ResultsForUpdate(results, plugin.Metadata, query,
currentCancellationToken));
if (currentCancellationToken.IsCancellationRequested || results == null)
{
return;
}
if (!_resultsUpdateQueue.Writer.TryWrite(new ResultsForUpdate(results, plugin.Metadata, query, currentCancellationToken)))
{
Log.Error("MainViewModel", "Unable to add item to Result Update Queue");
};
}
}, currentCancellationToken)
.ContinueWith(t => Log.Exception("|MainViewModel|Plugins Query Exceptions", t.Exception),