From 179b5169891b465eb144dee25a77894a7490e9e9 Mon Sep 17 00:00:00 2001 From: Anthony Pensiero Date: Wed, 4 Oct 2023 14:05:27 -0700 Subject: [PATCH 1/2] Fix ProcessScheduler memleak --- src/JsonRpc/ProcessScheduler.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/JsonRpc/ProcessScheduler.cs b/src/JsonRpc/ProcessScheduler.cs index ae0da9171..9a3bd09af 100644 --- a/src/JsonRpc/ProcessScheduler.cs +++ b/src/JsonRpc/ProcessScheduler.cs @@ -34,7 +34,7 @@ IScheduler scheduler var observableQueue = new BehaviorSubject<(RequestProcessType type, ReplaySubject> observer, Subject? contentModifiedSource)>( - ( RequestProcessType.Serial, new ReplaySubject>(int.MaxValue, Scheduler.Immediate), supportContentModified ? new Subject() : null ) + ( RequestProcessType.Serial, new ReplaySubject>(10, Scheduler.Immediate), supportContentModified ? new Subject() : null ) ); cd.Add( @@ -52,7 +52,7 @@ IScheduler scheduler logger.LogDebug("Completing existing request process type {Type}", observableQueue.Value.type); observableQueue.Value.observer.OnCompleted(); - observableQueue.OnNext(( item.type, new ReplaySubject>(int.MaxValue, Scheduler.Immediate), supportContentModified ? new Subject() : null )); + observableQueue.OnNext(( item.type, new ReplaySubject>(10, Scheduler.Immediate), supportContentModified ? new Subject() : null )); } logger.LogDebug("Queueing {Type}:{Name} request for processing", item.type, item.name); From 2502eb3cf62c0105260d8f99e24fb528ce34a06b Mon Sep 17 00:00:00 2001 From: Chris Cheetham Date: Thu, 5 Oct 2023 15:51:49 -0400 Subject: [PATCH 2/2] feat: use time span for replay buffer --- src/JsonRpc/ProcessScheduler.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/JsonRpc/ProcessScheduler.cs b/src/JsonRpc/ProcessScheduler.cs index 9a3bd09af..b810dcc77 100644 --- a/src/JsonRpc/ProcessScheduler.cs +++ b/src/JsonRpc/ProcessScheduler.cs @@ -34,7 +34,7 @@ IScheduler scheduler var observableQueue = new BehaviorSubject<(RequestProcessType type, ReplaySubject> observer, Subject? contentModifiedSource)>( - ( RequestProcessType.Serial, new ReplaySubject>(10, Scheduler.Immediate), supportContentModified ? new Subject() : null ) + ( RequestProcessType.Serial, new ReplaySubject>(TimeSpan.FromSeconds(6), Scheduler.Immediate), supportContentModified ? new Subject() : null ) ); cd.Add( @@ -52,7 +52,7 @@ IScheduler scheduler logger.LogDebug("Completing existing request process type {Type}", observableQueue.Value.type); observableQueue.Value.observer.OnCompleted(); - observableQueue.OnNext(( item.type, new ReplaySubject>(10, Scheduler.Immediate), supportContentModified ? new Subject() : null )); + observableQueue.OnNext(( item.type, new ReplaySubject>(TimeSpan.FromSeconds(6), Scheduler.Immediate), supportContentModified ? new Subject() : null )); } logger.LogDebug("Queueing {Type}:{Name} request for processing", item.type, item.name);