Я новичок в мире реактивных расширений и все еще пытаюсь учиться.
Я занимаюсь разработкой приложения с сеткой данных, которая отображает некоторые запущенные процессы Windows и их использование памяти. Использование памяти каждым процессом должно часто обновляться, т. Е. Каждые 200 мс.
Требования
- Когда флажок установлен
- сетка данных должна быть заполнена процессами, а использование памяти обновляется с использованием таймера с интервалом 200 мс.
- монитор (все должно быть сделано в фоновом потоке)
- если процесс завершился, его следует удалить из источника.
- если процесс запускается, его следует добавить к источнику
- файл для изменений
- Когда флажок снят
- вся деятельность монитора должна быть остановлена
- сетка данных очищена
Любая помощь будет принята с благодарностью!
Примечания:
- В прошлом я пробовал несколько подходов, таких как использование ObservableConcurrentDictionary в качестве ресурса и таймера для периодического обновления ресурса, но я столкнулся с проблемами (параллелизм, блокировка и т. Д.), Поэтому я хотел бы есть решение на основе Rx / ReactiveUI
- Из-за технических ограничений я могу использовать только .NET Framework 4.0, Reactive-core.Net40
Обновление
ViewModel
private ReactiveList<IProcessModel> _processes = new ReactiveList<IProcessModel>() { ChangeTrackingEnabled = true };
public ReactiveList<IProcessModel> Processes { get { return _processes; } }
public MainViewModel(IMonitorService monitorService)
{
this.WhenAnyValue(vm => vm.ShowProcessesIsChecked).Subscribe((b) => DoShowProcesses(b));
}
private void DoShowProcesses(bool checkboxChecked)
{
IDisposable timer;
Processes.Clear();
if (checkboxChecked)
{
//checkbox checked
lock (Processes)
Processes.AddRange(_monitorService.GetProcesses());
timer = Observable.Timer(TimeSpan.FromMilliseconds(200.0))
.Select(x =>
{
lock (Processes)
{
foreach (var process in Processes) //throws the 'Collection was modified; enumeration operation may not execute.'
process.UpdateMemory();
return Processes.Where(p => p.ProcessObject.HasExited).ToList();
}
}).
ObserveOnDispatcher()
.Subscribe(processesExited =>
{
if (processesExited.Count() > 0)
{
lock (Processes)
Processes.RemoveAll(processesExited); //remove all processes that have exited
}
});
}
else
{
if (timer != null)
timer.Dispose();
}
}
Я создал новую тему
Оригинал
ViewModel
public class MainViewModel : ReactiveObject
{
public ReactiveList<IProcessModel> Processes { get; private set; }
IMonitorService _monitorService;
public MainViewModel(IMonitorService monitorService)
{
_monitorService = monitorService;
Processes = new ReactiveList<IProcessModel>() { ChangeTrackingEnabled = true };
this.WhenAnyValue(vm => vm.ShowProcessesIsChecked)
.Where(value => value == true) //checkbox checked
.ObserveOn(Scheduler.Default) //raise notifications on thread-pool thread to keep UI responsive
.Select((isChecked) =>
{
return monitorService.GetProcesses();
})
.ObserveOn(SynchronizationContext.Current)
.Subscribe(processes => {
Processes.AddRange(processes); }
);
//start the MonitorService with MonitorService.Start(Processes)
//start a timer with an interval of 200ms --> at interval
//- do UpdateMemory() foreach IProcessModel in Processes
//- if ProcessObject.HasExited --> remove it from the collection source
;
this.WhenAnyValue(vm => vm.ShowProcessesIsChecked)
.Where(value => value == false) //checkbox unchecked
.Subscribe((isChecked) =>
{
monitorService.Stop(); //this stops monitoring for starting processes and clears the Processes
});
}
private bool _showProcessesIsChecked;
public bool ShowProcessesIsChecked
{
get { return _showProcessesIsChecked; }
set { this.RaiseAndSetIfChanged(ref _showProcessesIsChecked, value); }
}
}
Модель
public class ProcessModel : ProcessModelBase, IProcessModel
{
public ProcessModel(Process process)
{
ProcessObject = process;
}
public void UpdateMemory()
{
try
{
if (!ProcessObject.HasExited)
{
long mem = ProcessObject.PagedMemorySize64;
ProcessObject.Refresh();
if (mem != ProcessObject.PagedMemorySize64)
OnPropertyChanged(nameof(ProcessObject));
}
}
catch (Exception)
{
//log it
}
}
}
Услуги
public class MonitorService : IMonitorService
{
ManagementEventWatcher managementEventWatcher;
ReactiveList<IProcessModel> _processes;
public List<IProcessModel> GetProcesses()
{
List<IProcessModel> processes = new List<IProcessModel>();
foreach (var process in Process.GetProcesses().Where(p => p.ProcessName.Contains("chrome")))
processes.Add(new ProcessModel(process));
return processes;
}
/// <summary>
/// Starts the manager. Monitor a starting process and changes in log file
/// </summary>
/// <param name="processes"></param>
public void Start(ReactiveList<IProcessModel> processes)
{
_processes = processes;
var qStart = "SELECT * FROM Win32_ProcessStartTrace WHERE ProcessName like 'chrome'";
managementEventWatcher = new ManagementEventWatcher(new WqlEventQuery(qStart));
managementEventWatcher.EventArrived += new EventArrivedEventHandler(OnProcessStarted);
try
{
managementEventWatcher.Start();
}
catch (Exception)
{
//log it
}
Task.Factory.StartNew(() => MonitorLogFile());
}
public void Stop()
{
if (managementEventWatcher != null)
managementEventWatcher.Stop();
if (_processes != null)
_processes.Clear();
}
private void MonitorLogFile()
{
//this code monitors a log file for changes. It is possible that the IsChecked property of a ProcessModel object is set in the Processes collection
}
private void OnProcessStarted(object sender, EventArrivedEventArgs e)
{
try
{
Process process = Process.GetProcessById(Convert.ToInt32(e.NewEvent.Properties["ProcessID"].Value));
_processes.Add(new ProcessModel(process));
}
catch (ArgumentException)
{
//log it
}
catch (InvalidOperationException)
{
//log it
}
}
}
XAML
<CheckBox Content='Show Processes' IsChecked='{Binding ShowProcessesIsChecked}' />
<DataGrid ItemsSource="{Binding Processes}">
<DataGrid.Resources>
<DataGridTemplateColumn Header='Process'
x:Key='dgProcessName'
IsReadOnly='True'
x:Shared='False'>
<DataGridTemplateColumn.CellTemplate>
<DataTemplate>
<StackPanel Orientation='Horizontal' VerticalAlignment='Center'>
<CheckBox IsChecked='{Binding IsChecked, Mode=TwoWay, UpdateSourceTrigger=PropertyChanged}' HorizontalAlignment='Stretch' VerticalAlignment='Stretch'> </CheckBox>
<TextBlock Text='{Binding ProcessObject.ProcessName}' />
</StackPanel>
</DataTemplate>
</DataGridTemplateColumn.CellTemplate>
</DataGridTemplateColumn>
<DataGridTextColumn Header="PID"
Binding="{Binding ProcessObject.Id}"
IsReadOnly='True'
x:Key='dgPID'
x:Shared='False' />
<DataGridTextColumn Header="Commit Size"
Binding='{Binding ProcessObject.PagedMemorySize64}'
IsReadOnly='True'
x:Key='dgCommitSize'
x:Shared='False' />
</DataGrid.Resources>
</DataGrid>