隨著Mediatr 10的發布，現在有一個范式允許開發人員創建由IAsyncEnumerable. 我正在利用這種范例來創建多個不同的檔案系統觀察程式來監視多個檔案夾。為了監控檔案夾，我使用了兩種不同的方法：輪詢和FileSystemWatcher. 作為我管道的一部分，所有不同的檔案夾監視器都聚合到一個IEnumerable<IAsyncEnumerable<FileRecord>. 在每種型別的觀察者中，都有一個內部回圈運行，直到通過CancellationToken.
這是投票觀察者：
public class PolledFileStreamHandler :
IStreamRequestHandler<PolledFileStream, FileRecord>
{
private readonly ISeenFileStore _seenFileStore;
private readonly IPublisher _publisher;
private readonly ILogger<PolledFileStreamHandler> _logger;
public PolledFileStreamHandler(
ISeenFileStore seenFileStore,
IPublisher publisher,
ILogger<PolledFileStreamHandler> logger)
{
_seenFileStore = seenFileStore;
_publisher = publisher;
_logger = logger;
}
public async IAsyncEnumerable<FileRecord> Handle(
PolledFileStream request,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
var queue = new ConcurrentQueue<FileRecord>();
while (!cancellationToken.IsCancellationRequested)
{
var files = Directory.EnumerateFiles(request.Folder)
.Where(f => !_seenFileStore.Contains(f));
await Parallel.ForEachAsync(files, CancellationToken.None, async (f,t) =>
{
var info = new FileRecord(f);
_seenFileStore.Add(f);
await _publisher.Publish(new FileSeenNotification { FileInfo = info }, t);
queue.Enqueue(info);
});
// TODO: Try mixing the above parallel task with the serving task... Might be chaos...
while (!queue.IsEmpty)
{
if (queue.TryDequeue(out var result))
yield return result;
}
_logger.LogInformation("PolledFileStreamHandler watching {Directory} at: {Time}", request.Folder, DateTimeOffset.Now);
await Task.Delay(request.Interval, cancellationToken)
.ContinueWith(_ => {}, CancellationToken.None);
}
}
}
和 FileSystemWatcher
public class FileSystemStreamHandler :
IStreamRequestHandler<FileSystemStream, FileRecord>
{
private readonly ISeenFileStore _seenFileStore;
private readonly ILogger<FileSystemStreamHandler> _logger;
private readonly IPublisher _publisher;
private readonly ConcurrentQueue<FileRecord> _queue;
private Action<object, FileSystemEventArgs>? _tearDown;
public FileSystemStreamHandler(
ISeenFileStore seenFileStore,
ILogger<FileSystemStreamHandler> logger,
IPublisher publisher)
{
_seenFileStore = seenFileStore;
_logger = logger;
_publisher = publisher;
_queue = new ConcurrentQueue<FileRecord>();
}
public async IAsyncEnumerable<FileRecord> Handle(
FileSystemStream request,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
var watcher = SetupWatcher(request.Folder, cancellationToken);
while (!cancellationToken.IsCancellationRequested)
{
if (_queue.TryDequeue(out var record))
yield return record;
await Task.Delay(100, cancellationToken)
.ContinueWith(_ => {}, CancellationToken.None);
}
TearDownWatcher(watcher);
}
private FileSystemWatcher SetupWatcher(string folder, CancellationToken cancellation)
{
var watcher = new FileSystemWatcher(folder);
watcher.NotifyFilter = NotifyFilters.Attributes
| NotifyFilters.CreationTime
| NotifyFilters.DirectoryName
| NotifyFilters.FileName
| NotifyFilters.LastAccess
| NotifyFilters.LastWrite
| NotifyFilters.Security
| NotifyFilters.Size;
watcher.EnableRaisingEvents = true;
_tearDown = (_, args) => OnWatcherOnChanged(args, cancellation);
watcher.Created = _tearDown.Invoke;
return watcher;
}
private async void OnWatcherOnChanged(FileSystemEventArgs args, CancellationToken cancellationToken)
{
var path = args.FullPath;
if (_seenFileStore.Contains(path)) return;
_seenFileStore.Add(path);
try
{
if ((File.GetAttributes(path) & FileAttributes.Directory) != 0) return;
}
catch (FileNotFoundException)
{
_logger.LogWarning("File {File} was not found. During a routine check. Will not be broadcast", path);
return;
}
var record = new FileRecord(path);
_queue.Enqueue(record);
await _publisher.Publish(new FileSeenNotification { FileInfo = record }, cancellationToken);
}
private void TearDownWatcher(FileSystemWatcher watcher)
{
if (_tearDown != null)
watcher.Created -= _tearDown.Invoke;
}
}
Finally, here's the class that ties everything together and attempts to monitor the streams (in the StartAsync method). You'll notice the presence of a Merge operator coming from System.Interactive.Async, this does not currently operate as desired.
public class StreamedFolderWatcher : IDisposable
{
private readonly ConcurrentBag<Func<IAsyncEnumerable<FileRecord>>> _streams;
private CancellationTokenSource? _cancellationTokenSource;
private readonly IMediator _mediator;
private readonly ILogger<StreamedFolderWatcher> _logger;
public StreamedFolderWatcher(
IMediator mediator,
IEnumerable<IFileStream> fileStreams,
ILogger<StreamedFolderWatcher> logger)
{
_mediator = mediator;
_logger = logger;
_streams = new ConcurrentBag<Func<IAsyncEnumerable<FileRecord>>>();
_cancellationTokenSource = new CancellationTokenSource();
fileStreams.ToList()
.ForEach(f => AddStream(f, _cancellationTokenSource.Token));
}
private void AddStream<T>(
T request,
CancellationToken cancellationToken)
where T : IStreamRequest<FileRecord>
{
_streams.Add(() => _mediator.CreateStream(request, cancellationToken));
}
public async Task StartAsync(CancellationToken cancellationToken)
{
_cancellationTokenSource = CancellationTokenSource
.CreateLinkedTokenSource(cancellationToken);
var streams = _streams.Select(s => s()).ToList();
while (!cancellationToken.IsCancellationRequested)
{
await foreach (var file in streams.Merge().WithCancellation(cancellationToken))
{
_logger.LogInformation("Incoming file {File}", file);
}
await Task.Delay(1000, cancellationToken)
.ContinueWith(_ => {}, CancellationToken.None);
}
}
public Task StopAsync()
{
_cancellationTokenSource?.Cancel();
return Task.CompletedTask;
}
public void Dispose()
{
_cancellationTokenSource?.Dispose();
GC.SuppressFinalize(this);
}
}
My expectation for the Merge behavior is that if I have 3 IAsyncEnumerables, each item should be emitted as soon as it's yielded. Instead, unless I place yield break somewhere within the loops, the first IStreamRequestHandler fetched will simply execute ad infinitum until the cancellation token forces a stop.
How can I merge multiple input IAsyncEnumerables into a single long-lived output stream, that emits each time a result is yielded?
Minimum Reproducible Sample
static async IAsyncEnumerable<(Guid Id, int Value)> CreateSequence(
[EnumeratorCancellation] CancellationToken cancellationToken)
{
var random = new Random();
var id = Guid.NewGuid();
while (!cancellationToken.IsCancellationRequested)
{
await Task.Delay(TimeSpan.FromMilliseconds(random.Next(100, 1000)));
yield return (id, random.Next(0, 10));
}
}
var token = new CancellationTokenSource();
var sequences = Enumerable.Range(0, 10)
.Select(_ => CreateSequence(token.Token));
var merged = sequences.Merge();
await foreach (var (id, value) in merged)
{
Console.WriteLine($"[{DateTime.Now.ToShortTimeString()}] Value {value} Emitted from {id}");
}
uj5u.com熱心網友回復：
似乎 Rx 團隊搞砸了Merge操作員，并創建了具有不同行為的多載。此多載支持并發：
public static IAsyncEnumerable<TSource> Merge<TSource>(
params IAsyncEnumerable<TSource>[] sources);
此多載不支持并發：
public static IAsyncEnumerable<TSource> Merge<TSource>(
this IEnumerable<IAsyncEnumerable<TSource>> sources);
從源代碼中的注釋：
// REVIEW:
// This implementation does not exploit concurrency. We should not introduce such
// behavior in order to avoid breaking changes, but we could introduce a parallel
// ConcurrentMerge implementation. It is unfortunate though that the Merge
// overload accepting an array has always been concurrent, so we can't change that
// either (in order to have consistency where Merge is non-concurrent, and
// ConcurrentMerge is).
所以你要做的就是.ToArray()在Merge().
uj5u.com熱心網友回復：
我設法提出了一個可行的，但可能效率低下且可能存在錯誤的解決方案。通過將每個都IAsyncEnumerable放入自己的后臺任務中，我能夠將每個發送到執行緒安全佇列中，當每個可用時，它們都會在其中提供服務。
public static async IAsyncEnumerable<TSource> MergeAsyncEnumerable<TSource>(
this IEnumerable<IAsyncEnumerable<TSource>> sources,
TimeSpan? debounceTime = default,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
var queue = new ConcurrentQueue<TSource>();
var tasks = SetupCollections(sources, queue, cancellationToken);
while (!Task.WhenAll(tasks).IsCompleted)
{
while (!queue.IsEmpty)
if (queue.TryDequeue(out var record))
yield return record;
// Small debounce to prevent an infinite loop from just spinning.
await WaitIfDebounce(debounceTime, cancellationToken);
}
await Task.CompletedTask;
}
private static Task WaitIfDebounce(
TimeSpan? debounceTime,
CancellationToken cancellationToken)
{
return debounceTime.HasValue
? Task.Delay(debounceTime.Value, cancellationToken)
.ContinueWith(_ => { }, CancellationToken.None)
: Task.CompletedTask;
}
private static IList<Task> SetupCollections<TSource>(
IEnumerable<IAsyncEnumerable<TSource>> sources,
ConcurrentQueue<TSource> queue,
CancellationToken cancellationToken)
{
return sources
.Select(s => Task.Run(async () =>
{
await foreach (var file in s.WithCancellation(cancellationToken))
queue.Enqueue(file);
}, cancellationToken))
.ToList();
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/410855.html
標籤：