Added fixture based tests and re-enabled disabled tests

src/Client/LanguageClientOptions.cs

@@ -61,6 +61,9 @@ ILanguageClientRegistry IJsonRpcHandlerRegistry<ILanguageClientRegistry>.AddHand
         ILanguageClientRegistry IJsonRpcHandlerRegistry<ILanguageClientRegistry>.AddHandler(string method, Type type, JsonRpcHandlerOptions options) =>
             AddHandler(method, type, options);
 
+        ILanguageClientRegistry IJsonRpcHandlerRegistry<ILanguageClientRegistry>.AddHandlerLink(string sourceMethod, string destinationMethod) =>
+            AddHandlerLink(sourceMethod, destinationMethod);
+
         ILanguageClientRegistry IJsonRpcHandlerRegistry<ILanguageClientRegistry>.OnJsonRequest(string method, Func<JToken, Task<JToken>> handler, JsonRpcHandlerOptions options) =>
             OnJsonRequest(method, handler, options);
 

src/Dap.Client/DebugAdapterClientOptions.cs

@@ -49,6 +49,9 @@ IDebugAdapterClientRegistry IJsonRpcHandlerRegistry<IDebugAdapterClientRegistry>
         IDebugAdapterClientRegistry IJsonRpcHandlerRegistry<IDebugAdapterClientRegistry>.AddHandler(string method, Type type, JsonRpcHandlerOptions options) =>
             AddHandler(method, type, options);
 
+        IDebugAdapterClientRegistry IJsonRpcHandlerRegistry<IDebugAdapterClientRegistry>.AddHandlerLink(string sourceMethod, string destinationMethod) =>
+            AddHandlerLink(sourceMethod, destinationMethod);
+
         IDebugAdapterClientRegistry IJsonRpcHandlerRegistry<IDebugAdapterClientRegistry>.OnJsonRequest(
             string method, Func<JToken, Task<JToken>> handler, JsonRpcHandlerOptions options
         ) => OnJsonRequest(method, handler, options);

src/Dap.Protocol/IProgressObservable.cs

@@ -1,9 +1,11 @@
 using System;
 using OmniSharp.Extensions.DebugAdapter.Protocol.Events;
+using OmniSharp.Extensions.DebugAdapter.Protocol.Models;
 
 namespace OmniSharp.Extensions.DebugAdapter.Protocol
 {
     public interface IProgressObservable : IObservable<ProgressEvent>
     {
+        ProgressToken ProgressToken { get; }
     }
 }

src/Dap.Protocol/Models/ProgressToken.cs

@@ -69,7 +69,6 @@ public bool Equals(ProgressToken other) =>
             String == other.String;
 
         public bool Equals(long other) => IsLong && Long == other;
-
         public bool Equals(string other) => IsString && String == other;
 
         private string DebuggerDisplay => IsString ? String : IsLong ? Long.ToString() : "";

src/Dap.Server/DebugAdapterServerOptions.cs

@@ -37,6 +37,9 @@ IDebugAdapterServerRegistry IJsonRpcHandlerRegistry<IDebugAdapterServerRegistry>
         IDebugAdapterServerRegistry IJsonRpcHandlerRegistry<IDebugAdapterServerRegistry>.AddHandler(string method, Type type, JsonRpcHandlerOptions options) =>
             AddHandler(method, type, options);
 
+        IDebugAdapterServerRegistry IJsonRpcHandlerRegistry<IDebugAdapterServerRegistry>.AddHandlerLink(string sourceMethod, string destinationMethod) =>
+            AddHandlerLink(sourceMethod, destinationMethod);
+
         IDebugAdapterServerRegistry IJsonRpcHandlerRegistry<IDebugAdapterServerRegistry>.OnJsonRequest(
             string method, Func<JToken, Task<JToken>> handler, JsonRpcHandlerOptions options
         ) => OnJsonRequest(method, handler, options);

src/JsonRpc.Testing/AggregateSettler.cs

@@ -13,7 +13,7 @@ public class AggregateSettler : ISettler
 
         public AggregateSettler(params ISettler[] settlers) => _settlers = settlers;
 
-        public Task SettleNext() => Settle().Take(1).IgnoreElements().LastOrDefaultAsync().ToTask();
+        public Task SettleNext() => Task.WhenAll(_settlers.Select(z => z.SettleNext()));
 
         public IObservable<Unit> Settle() =>
             _settlers

src/JsonRpc.Testing/JsonRpcIntegrationServerTestBase.cs

@@ -19,10 +19,10 @@ public JsonRpcIntegrationServerTestBase(JsonRpcTestOptions testOptions)
             _cancellationTokenSource = new CancellationTokenSource();
             if (!Debugger.IsAttached)
             {
-                _cancellationTokenSource.CancelAfter(testOptions.TestTimeout);
+                _cancellationTokenSource.CancelAfter(testOptions.CancellationTimeout);
             }
 
-            Events = ClientEvents = new Settler(TestOptions.SettleTimeSpan, TestOptions.SettleTimeout, CancellationToken);
+            Events = ClientEvents = new Settler(TestOptions, CancellationToken);
         }
 
         protected CompositeDisposable Disposable { get; }

src/JsonRpc.Testing/JsonRpcTestBase.cs

@@ -19,22 +19,22 @@ public JsonRpcTestBase(JsonRpcTestOptions testOptions)
             _cancellationTokenSource = new CancellationTokenSource();
             if (!Debugger.IsAttached)
             {
-                _cancellationTokenSource.CancelAfter(testOptions.TestTimeout);
+                _cancellationTokenSource.CancelAfter(testOptions.CancellationTimeout);
             }
 
-            ClientEvents = new Settler(TestOptions.SettleTimeSpan, TestOptions.SettleTimeout, CancellationToken);
-            ServerEvents = new Settler(TestOptions.SettleTimeSpan, TestOptions.SettleTimeout, CancellationToken);
+            ClientEvents = new Settler(TestOptions, CancellationToken);
+            ServerEvents = new Settler(TestOptions, CancellationToken);
             Events = new AggregateSettler(ClientEvents, ServerEvents);
         }
 
         protected CompositeDisposable Disposable { get; }
-        protected ISettler ClientEvents { get; }
-        protected ISettler ServerEvents { get; }
-        protected ISettler Events { get; }
-        protected JsonRpcTestOptions TestOptions { get; }
-        protected internal CancellationToken CancellationToken => _cancellationTokenSource.Token;
-        protected Task SettleNext() => Events.SettleNext();
-        protected IObservable<Unit> Settle() => Events.Settle();
+        public ISettler ClientEvents { get; }
+        public  ISettler ServerEvents { get; }
+        public  ISettler Events { get; }
+        public  JsonRpcTestOptions TestOptions { get; }
+        public  CancellationToken CancellationToken => _cancellationTokenSource.Token;
+        public  Task SettleNext() => Events.SettleNext();
+        public  IObservable<Unit> Settle() => Events.Settle();
 
         public void Dispose()
         {

src/JsonRpc.Testing/JsonRpcTestOptions.cs

@@ -21,9 +21,9 @@ public JsonRpcTestOptions(ILoggerFactory clientLoggerFactory, ILoggerFactory ser
 
         public ILoggerFactory ClientLoggerFactory { get; internal set; } = NullLoggerFactory.Instance;
         public ILoggerFactory ServerLoggerFactory { get; internal set; } = NullLoggerFactory.Instance;
-        public TimeSpan SettleTimeSpan { get; internal set; } = TimeSpan.FromMilliseconds(100);
-        public TimeSpan SettleTimeout { get; internal set; } = TimeSpan.FromMilliseconds(500);
-        public TimeSpan TestTimeout { get; internal set; } = TimeSpan.FromMinutes(5);
+        public TimeSpan WaitTime { get; internal set; } = TimeSpan.FromMilliseconds(50);
+        public TimeSpan Timeout { get; internal set; } = TimeSpan.FromMilliseconds(500);
+        public TimeSpan CancellationTimeout { get; internal set; } = TimeSpan.FromMinutes(5);
         public PipeOptions DefaultPipeOptions { get; internal set; } = new PipeOptions();
     }
 }

src/JsonRpc.Testing/JsonRpcTestOptionsExtensions.cs

@@ -18,21 +18,21 @@ public static JsonRpcTestOptions WithClientLoggerFactory(this JsonRpcTestOptions
             return options;
         }
 
-        public static JsonRpcTestOptions WithSettleTimeSpan(this JsonRpcTestOptions options, TimeSpan settleTimeSpan)
+        public static JsonRpcTestOptions WithWaitTime(this JsonRpcTestOptions options, TimeSpan waitTime)
         {
-            options.SettleTimeSpan = settleTimeSpan;
+            options.WaitTime = waitTime;
             return options;
         }
 
-        public static JsonRpcTestOptions WithSettleTimeout(this JsonRpcTestOptions options, TimeSpan timeout)
+        public static JsonRpcTestOptions WithTimeout(this JsonRpcTestOptions options, TimeSpan timeout)
         {
-            options.SettleTimeout = timeout;
+            options.Timeout = timeout;
             return options;
         }
 
-        public static JsonRpcTestOptions WithTestTimeout(this JsonRpcTestOptions options, TimeSpan testTimeout)
+        public static JsonRpcTestOptions WithCancellationTimeout(this JsonRpcTestOptions options, TimeSpan cancellationTimeout)
         {
-            options.TestTimeout = testTimeout;
+            options.CancellationTimeout = cancellationTimeout;
             return options;
         }
 

src/JsonRpc.Testing/Settler.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Reactive;
 using System.Reactive.Concurrency;
+using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Reactive.Threading.Tasks;
 using System.Threading;
@@ -11,21 +12,21 @@ namespace OmniSharp.Extensions.JsonRpc.Testing
 {
     public class Settler : ISettler, IRequestSettler, IDisposable
     {
-        private readonly TimeSpan _timeout;
+        private readonly JsonRpcTestOptions _options;
         private readonly CancellationToken _cancellationToken;
         private readonly IScheduler _scheduler;
         private readonly IObservable<Unit> _settle;
         private readonly IObserver<int> _requester;
         private readonly IDisposable _connectable;
-        private readonly IObservable<Unit> _defaultValue;
+        private readonly IObservable<Unit> _timeoutValue;
 
-        public Settler(TimeSpan waitTime, TimeSpan timeout, CancellationToken cancellationToken, IScheduler scheduler = null)
+        public Settler(JsonRpcTestOptions options, CancellationToken cancellationToken, IScheduler scheduler = null)
         {
-            _timeout = timeout;
+            _options = options;
             _cancellationToken = cancellationToken;
             scheduler ??= Scheduler.Immediate;
             _scheduler = scheduler;
-            _defaultValue = Return(Unit.Default, _scheduler);
+            _timeoutValue = Return(Unit.Default, _scheduler);
             var subject = new Subject<int>();
             var data = subject;
 
@@ -42,24 +43,34 @@ public Settler(TimeSpan waitTime, TimeSpan timeout, CancellationToken cancellati
                                   z => {
                                       if (z > 0)
                                       {
-                                          return Timer(_timeout, _scheduler)
-                                             .Select(z => Unit.Default);
+                                          return Never<Unit>();
+//                                          return Timer(_options.Timeout, _scheduler)
+//                                             .Select(z => Unit.Default);
                                       }
 
-                                      return Amb(Timer(waitTime, _scheduler), Timer(_timeout, _scheduler))
+                                      return Timer(_options.WaitTime, _scheduler)
                                          .Select(z => Unit.Default);
                                   }
                               )
                              .Replay(1, _scheduler);
             _connectable = connectable.Connect();
             _settle = connectable
-               .Switch();
+                     .Select(o => o.Timeout(_options.Timeout, _scheduler))
+                     .Switch();
             _requester = subject.AsObserver();
         }
 
-        public Task SettleNext() => _settle.Take(1).IgnoreElements().LastOrDefaultAsync().ToTask(_cancellationToken);
+        public Task SettleNext() => SettleNextInternal().ToTask(_cancellationToken);
 
-        public IObservable<Unit> Settle() => _settle.Timeout(_timeout, _scheduler).Catch<Unit, Exception>(_ => _defaultValue);
+        public IObservable<Unit> SettleNextInternal() => _settle
+                                                        .Catch<Unit, Exception>(_ => _timeoutValue)
+                                                        .Take(1)
+                                                        .IgnoreElements()
+                                                        .LastOrDefaultAsync();
+
+        public IObservable<Unit> Settle() =>  _settle
+                                            .Timeout(_options.Timeout, _scheduler)
+                                             .Catch<Unit, Exception>(_ => _timeoutValue);
 
         void IRequestSettler.OnStartRequest() => _requester.OnNext(1);
 

src/JsonRpc/CompositeHandlersManager.cs

@@ -27,15 +27,40 @@ public IDisposable Add(string method, IJsonRpcHandler handler, JsonRpcHandlerOpt
             return result;
         }
 
-        public IDisposable Add(JsonRpcHandlerFactory factory, JsonRpcHandlerOptions options) => _parent.Add(factory, options);
+        public IDisposable Add(JsonRpcHandlerFactory factory, JsonRpcHandlerOptions options)
+        {
+            var result = _parent.Add(factory, options);
+            _compositeDisposable.Add(result);
+            return result;
+        }
 
-        public IDisposable Add(string method, JsonRpcHandlerFactory factory, JsonRpcHandlerOptions options) => _parent.Add(method, factory, options);
+        public IDisposable Add(string method, JsonRpcHandlerFactory factory, JsonRpcHandlerOptions options)
+        {
+            var result = _parent.Add(factory, options);
+            _compositeDisposable.Add(result);
+            return result;
+        }
 
-        public IDisposable Add(Type handlerType, JsonRpcHandlerOptions options) => _parent.Add(handlerType, options);
+        public IDisposable Add(Type handlerType, JsonRpcHandlerOptions options)
+        {
+            var result = _parent.Add(handlerType, options);
+            _compositeDisposable.Add(result);
+            return result;
+        }
 
-        public IDisposable Add(string method, Type handlerType, JsonRpcHandlerOptions options) => _parent.Add(method, handlerType, options);
+        public IDisposable Add(string method, Type handlerType, JsonRpcHandlerOptions options)
+        {
+            var result = _parent.Add(method, handlerType, options);
+            _compositeDisposable.Add(result);
+            return result;
+        }
 
-        public IDisposable AddLink(string sourceMethod, string destinationMethod) => _parent.AddLink(sourceMethod, destinationMethod);
+        public IDisposable AddLink(string sourceMethod, string destinationMethod)
+        {
+            var result = _parent.AddLink(sourceMethod,destinationMethod);
+            _compositeDisposable.Add(result);
+            return result;
+        }
 
         public CompositeDisposable GetDisposable() => _compositeDisposable;
     }

src/JsonRpc/IJsonRpcHandlerRegistry.cs

@@ -25,6 +25,7 @@ public interface IJsonRpcHandlerRegistry<out T> : IJsonRpcHandlerRegistry where
         T AddHandler<TTHandler>(string method, JsonRpcHandlerOptions options = null) where TTHandler : IJsonRpcHandler;
         T AddHandler(Type type, JsonRpcHandlerOptions options = null);
         T AddHandler(string method, Type type, JsonRpcHandlerOptions options = null);
+        T AddHandlerLink(string sourceMethod, string destinationMethod);
 
         T OnJsonRequest(string method, Func<JToken, Task<JToken>> handler, JsonRpcHandlerOptions options = null);
         T OnJsonRequest(string method, Func<JToken, CancellationToken, Task<JToken>> handler, JsonRpcHandlerOptions options = null);

src/JsonRpc/InterimJsonRpcServerRegistry.cs

@@ -57,5 +57,11 @@ public sealed override T AddHandler(string method, Type type, JsonRpcHandlerOpti
             _handlersManager.Add(method, type, options);
             return (T) (object) this;
         }
+
+        public sealed override T AddHandlerLink(string sourceMethod, string destinationMethod)
+        {
+            _handlersManager.AddLink(sourceMethod, destinationMethod);
+            return (T) (object) this;
+        }
     }
 }

src/JsonRpc/JsonRpcCommonMethodsBase.cs

@@ -112,6 +112,7 @@ public T OnNotification(string method, Func<CancellationToken, Task> handler, Js
         public abstract T AddHandler<THandler>(string method, JsonRpcHandlerOptions options = null) where THandler : IJsonRpcHandler;
         public abstract T AddHandler(Type type, JsonRpcHandlerOptions options = null);
         public abstract T AddHandler(string method, Type type, JsonRpcHandlerOptions options = null);
+        public abstract T AddHandlerLink(string sourceMethod, string destinationMethod);
 
         #endregion
     }

src/JsonRpc/JsonRpcOptionsRegistryBase.cs

@@ -69,6 +69,12 @@ public sealed override T AddHandler(string method, Type type, JsonRpcHandlerOpti
             return (T) (object) this;
         }
 
+        public sealed override T AddHandlerLink(string sourceMethod, string destinationMethod)
+        {
+            Handlers.Add(JsonRpcHandlerDescription.Link(sourceMethod, destinationMethod));
+            return (T) (object) this;
+        }
+
         #endregion
     }
 }

src/JsonRpc/OutputHandler.cs

@@ -55,7 +55,8 @@ ILogger<OutputHandler> logger
             pipeWriter,
             serializer,
             receiver,
-            new EventLoopScheduler(_ => new Thread(_) { IsBackground = true, Name = "OutputHandler" }),
+            TaskPoolScheduler.Default,
+            //new EventLoopScheduler(_ => new Thread(_) { IsBackground = true, Name = "OutputHandler" }),
             logger
         )
         {

src/Protocol/Client/WorkDone/LanguageClientWorkDoneManager.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Reactive.Disposables;
+using System.Reactive.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MediatR;
@@ -96,6 +97,19 @@ public void Dispose()
             _progressObservable.Dispose();
         }
 
-        public IDisposable Subscribe(IObserver<WorkDoneProgress> observer) => _progressObservable.Subscribe(observer);
+        public IDisposable Subscribe(IObserver<WorkDoneProgress> observer)
+        {
+            return _progressObservable.Subscribe(
+                _ => {
+                    observer.OnNext(_);
+                    if (_ is WorkDoneProgressEnd)
+                    {
+                        observer.OnCompleted();
+                    }
+                },
+                observer.OnError,
+                observer.OnCompleted
+            );
+        }
     }
 }

src/Server/LanguageServerOptions.cs

@@ -43,6 +43,9 @@ ILanguageServerRegistry IJsonRpcHandlerRegistry<ILanguageServerRegistry>.AddHand
         ILanguageServerRegistry IJsonRpcHandlerRegistry<ILanguageServerRegistry>.AddHandler(string method, Type type, JsonRpcHandlerOptions options) =>
             AddHandler(method, type, options);
 
+        ILanguageServerRegistry IJsonRpcHandlerRegistry<ILanguageServerRegistry>.AddHandlerLink(string sourceMethod, string destinationMethod) =>
+            AddHandlerLink(sourceMethod, destinationMethod);
+
         ILanguageServerRegistry IJsonRpcHandlerRegistry<ILanguageServerRegistry>.OnJsonRequest(string method, Func<JToken, Task<JToken>> handler, JsonRpcHandlerOptions options) =>
             OnJsonRequest(method, handler, options);
 

test/Dap.Tests/Integration/ConnectionAndDisconnectionTests.cs

@@ -15,9 +15,7 @@ namespace Dap.Tests.Integration
     public class ConnectionAndDisconnectionTests : DebugAdapterProtocolTestBase
     {
         public ConnectionAndDisconnectionTests(ITestOutputHelper outputHelper) : base(
-            new JsonRpcTestOptions()
-               .ConfigureForXUnit(outputHelper)
-               .WithTestTimeout(TimeSpan.FromSeconds(20))
+            new JsonRpcTestOptions().ConfigureForXUnit(outputHelper)
         )
         {
         }