Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor request invocation to be extensible. #641

Merged
merged 4 commits into from
Aug 27, 2021
Merged

Refactor request invocation to be extensible. #641

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b4e8140
Refactor request invocation to be extensible.
NTaylorMullen Aug 23, 2021
202a3d0
Make existing request response types fully public
NTaylorMullen Aug 23, 2021
cc5a169
Fix test.
NTaylorMullen Aug 23, 2021
903ca76
Addressed code review comments
NTaylorMullen Aug 24, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 35 additions & 6 deletions src/JsonRpc/Connection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ public class Connection : IDisposable
private readonly InputHandler _inputHandler;
public bool IsOpen { get; private set; }

[Obsolete("Use the other constructor that takes a request invoker")]
public Connection(
PipeReader input,
IOutputHandler outputHandler,
Expand All @@ -25,21 +26,49 @@ public Connection(
int concurrency,
IScheduler scheduler,
CreateResponseExceptionHandler? getException = null
) : this(
input,
outputHandler,
receiver,
requestRouter,
responseRouter,
new DefaultRequestInvoker(
requestRouter,
outputHandler,
requestProcessIdentifier,
new RequestInvokerOptions(
requestTimeout,
supportContentModified,
concurrency),
loggerFactory,
scheduler),
loggerFactory,
onUnhandledException,
getException)
{
}

public Connection(
PipeReader input,
IOutputHandler outputHandler,
IReceiver receiver,
IRequestRouter<IHandlerDescriptor?> requestRouter,
IResponseRouter responseRouter,
RequestInvoker requestInvoker,
ILoggerFactory loggerFactory,
OnUnhandledExceptionHandler onUnhandledException,
CreateResponseExceptionHandler? getException = null
) =>
_inputHandler = new InputHandler(
input,
outputHandler,
receiver,
requestProcessIdentifier,
requestRouter,
responseRouter,
requestInvoker,
loggerFactory,
onUnhandledException,
getException,
requestTimeout,
supportContentModified,
concurrency > 1 ? (int?) concurrency : null,
scheduler
getException
);

public void Open()
Expand Down
6 changes: 3 additions & 3 deletions src/JsonRpc/ContentModified.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
using OmniSharp.Extensions.JsonRpc.Server;
using OmniSharp.Extensions.JsonRpc.Server;
using OmniSharp.Extensions.JsonRpc.Server.Messages;

namespace OmniSharp.Extensions.JsonRpc
{
public class ContentModified : RpcError
{
internal ContentModified(string method) : base(null, method, new ErrorMessage(ErrorCodes.ContentModified, "Content Modified"))
public ContentModified(string method) : base(null, method, new ErrorMessage(ErrorCodes.ContentModified, "Content Modified"))
{
}

internal ContentModified(object id, string method) : base(id, method, new ErrorMessage(ErrorCodes.ContentModified, "Content Modified"))
public ContentModified(object id, string method) : base(id, method, new ErrorMessage(ErrorCodes.ContentModified, "Content Modified"))
{
}
}
Expand Down
181 changes: 181 additions & 0 deletions src/JsonRpc/DefaultRequestInvoker.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,181 @@
using System;
using System.Reactive;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using Microsoft.Extensions.Logging;
using OmniSharp.Extensions.JsonRpc.Server;
using OmniSharp.Extensions.JsonRpc.Server.Messages;
using Notification = OmniSharp.Extensions.JsonRpc.Server.Notification;

namespace OmniSharp.Extensions.JsonRpc
{
public class DefaultRequestInvoker : RequestInvoker
{
private readonly IRequestRouter<IHandlerDescriptor?> _requestRouter;
private readonly IOutputHandler _outputHandler;
private readonly ProcessScheduler _processScheduler;
private readonly IRequestProcessIdentifier _requestProcessIdentifier;
private readonly RequestInvokerOptions _options;
private readonly ILogger<DefaultRequestInvoker> _logger;

public DefaultRequestInvoker(
IRequestRouter<IHandlerDescriptor?> requestRouter,
IOutputHandler outputHandler,
IRequestProcessIdentifier requestProcessIdentifier,
RequestInvokerOptions options,
ILoggerFactory loggerFactory,
IScheduler scheduler)
{
_requestRouter = requestRouter;
_outputHandler = outputHandler;
_requestProcessIdentifier = requestProcessIdentifier;
_options = options;
_processScheduler = new ProcessScheduler(loggerFactory, _options.SupportContentModified, _options.Concurrency, scheduler);
_logger = loggerFactory.CreateLogger<DefaultRequestInvoker>();
}

public override RequestInvocationHandle InvokeRequest(IRequestDescriptor<IHandlerDescriptor?> descriptor, Request request)
{
if (descriptor.Default is null)
{
throw new ArgumentNullException(nameof(descriptor.Default));
}

var handle = new RequestInvocationHandle(request);
var type = _requestProcessIdentifier.Identify(descriptor.Default);

var schedulerDelegate = RouteRequest(descriptor, request, handle);
_processScheduler.Add(type, $"{request.Method}:{request.Id}", schedulerDelegate);

return handle;
}

public override void InvokeNotification(IRequestDescriptor<IHandlerDescriptor?> descriptor, Notification notification)
{
if (descriptor.Default is null)
{
throw new ArgumentNullException(nameof(descriptor.Default));
}

var type = _requestProcessIdentifier.Identify(descriptor.Default);
var schedulerDelegate = RouteNotification(descriptor, notification);
_processScheduler.Add(type, notification.Method, schedulerDelegate);
}

public override void Dispose()
{
_processScheduler.Dispose();
}

private SchedulerDelegate RouteRequest(
IRequestDescriptor<IHandlerDescriptor?> descriptor,
Request request,
RequestInvocationHandle handle)
{
var cts = handle.CancellationTokenSource;
return (contentModifiedToken, scheduler) =>
Observable.Create<ErrorResponse>(
observer => {
// ITS A RACE!
var sub = Observable.Amb(
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All this is verbatim from the original RouteRequest method in InputHandler.

contentModifiedToken.Select(
_ => {
_logger.LogTrace(
"Request {Id} was abandoned due to content be modified", request.Id
);
return new ErrorResponse(
new ContentModified(request.Id, request.Method)
);
}
),
Observable.Timer(_options.RequestTimeout, scheduler).Select(
_ => new ErrorResponse(new RequestCancelled(request.Id, request.Method))
),
Observable.FromAsync(
async ct => {
using var timer = _logger.TimeDebug(
"Processing request {Method} {ResponseId}", request.Method,
request.Id
);
ct.Register(cts.Cancel);
// ObservableToToken(contentModifiedToken).Register(cts.Cancel);
try
{
var result = await _requestRouter.RouteRequest(
descriptor, request, cts.Token
).ConfigureAwait(false);
return result;
}
catch (OperationCanceledException)
{
_logger.LogTrace("Request {Id} was cancelled", request.Id);
return new RequestCancelled(request.Id, request.Method);
}
catch (RpcErrorException e)
{
_logger.LogCritical(
Events.UnhandledRequest, e,
"Failed to handle request {Method} {RequestId}", request.Method,
request.Id
);
return new RpcError(
request.Id, request.Method,
new ErrorMessage(e.Code, e.Message, e.Error)
);
}
catch (Exception e)
{
_logger.LogCritical(
Events.UnhandledRequest, e,
"Failed to handle request {Method} {RequestId}", request.Method,
request.Id
);
return new InternalError(request.Id, request.Method, e.ToString());
}
}
)
)
.Subscribe(observer);
return new CompositeDisposable(sub, handle);
}
)
.Select(
response => {
_outputHandler.Send(response.Value);
return Unit.Default;
}
);
}

private SchedulerDelegate RouteNotification(
IRequestDescriptor<IHandlerDescriptor?> descriptors,
Notification notification) =>
(_, scheduler) =>
// ITS A RACE!
Observable.Amb(
Observable.Timer(_options.RequestTimeout, scheduler)
.Select(_ => Unit.Default)
.Do(
_ => _logger.LogTrace("Notification was cancelled due to timeout")
),
Observable.FromAsync(
async ct => {
using var timer = _logger.TimeDebug("Processing notification {Method}", notification.Method);
try
{
await _requestRouter.RouteNotification(descriptors, notification, ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
_logger.LogTrace("Notification was cancelled");
}
catch (Exception e)
{
_logger.LogCritical(Events.UnhandledRequest, e, "Failed to handle request {Method}", notification.Method);
}
}
)
);
}
}
Loading