async output

Microsoft.DotNet.Interactive.Jupyter.Tests/CompleteRequestHandlerTests.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 
 using System;
+using System.Reactive.Concurrency;
 using System.Threading.Tasks;
 using FluentAssertions;
 using Microsoft.DotNet.Interactive.Jupyter.Protocol;

Microsoft.DotNet.Interactive.Jupyter.Tests/ExecuteRequestHandlerTests.cs

@@ -7,6 +7,10 @@
 using Microsoft.DotNet.Interactive.Jupyter.Protocol;
 using WorkspaceServer.Kernel;
 using Xunit;
+using System.Reactive.Linq;
+using Microsoft.DotNet.Interactive.Events;
+using FluentAssertions.Extensions;
+using System.Reactive.Concurrency;
 
 namespace Microsoft.DotNet.Interactive.Jupyter.Tests
 {

Microsoft.DotNet.Interactive.Jupyter/CompleteRequestHandler.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Text.RegularExpressions;
 using System.Threading.Tasks;
 using Microsoft.DotNet.Interactive.Commands;
@@ -12,14 +13,15 @@
 
 namespace Microsoft.DotNet.Interactive.Jupyter
 {
-    public class CompleteRequestHandler: RequestHandlerBase<CompleteRequest>
+    public class CompleteRequestHandler : RequestHandlerBase<CompleteRequest>
     {
         private static readonly Regex _lastToken = new Regex(@"(?<lastToken>\S+)$", RegexOptions.Compiled | RegexOptions.IgnoreCase | RegexOptions.CultureInvariant | RegexOptions.Multiline);
 
 
-        public CompleteRequestHandler(IKernel kernel) : base(kernel)
+        public CompleteRequestHandler(IKernel kernel, IScheduler scheduler = null)
+            : base(kernel, scheduler ?? CurrentThreadScheduler.Instance)
         {
-            
+
         }
 
         public async Task Handle(JupyterRequestContext context)
@@ -34,22 +36,18 @@ public async Task Handle(JupyterRequestContext context)
 
             InFlightRequests[command] = openRequest;
 
-            var kernelResult = await Kernel.SendAsync(command);
-            openRequest.AddDisposable(kernelResult.KernelEvents.Subscribe(OnKernelResultEvent));
-
+            await Kernel.SendAsync(command);
         }
 
-        void OnKernelResultEvent(IKernelEvent value)
+        protected override void OnKernelEvent(IKernelEvent @event)
         {
-            switch (value)
+            switch (@event)
             {
                 case CompletionRequestCompleted completionRequestCompleted:
                     OnCompletionRequestCompleted(completionRequestCompleted, InFlightRequests);
                     break;
                 case CompletionRequestReceived _:
                     break;
-                default:
-                    throw new NotSupportedException();
             }
         }
 
@@ -68,7 +66,7 @@ private static void OnCompletionRequestCompleted(CompletionRequestCompleted comp
             openRequest.Context.ServerChannel.Send(completeReply);
             openRequest.Context.RequestHandlerStatus.SetAsIdle();
             openRequest.Dispose();
-            
+
         }
 
         private static int ComputeReplacementStartPosition(string code, int cursorPosition)

Microsoft.DotNet.Interactive.Jupyter/ExecuteRequestHandler.cs

@@ -5,6 +5,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.DotNet.Interactive.Commands;
@@ -17,7 +18,8 @@ public class ExecuteRequestHandler : RequestHandlerBase<ExecuteRequest>
     {
         private int _executionCount;
 
-        public ExecuteRequestHandler(IKernel kernel) : base(kernel)
+        public ExecuteRequestHandler(IKernel kernel, IScheduler scheduler = null)
+            : base(kernel, scheduler ?? CurrentThreadScheduler.Instance)
         {
         }
 
@@ -36,8 +38,7 @@ public async Task Handle(JupyterRequestContext context)
 
             try
             {
-                var kernelResult = await Kernel.SendAsync(command);
-                openRequest.AddDisposable(kernelResult.KernelEvents.Subscribe(OnKernelResultEvent));
+                await Kernel.SendAsync(command);
             }
             catch (Exception e)
             {
@@ -84,9 +85,9 @@ public async Task Handle(JupyterRequestContext context)
             return transient;
         }
 
-        void OnKernelResultEvent(IKernelEvent value)
+        protected override void OnKernelEvent(IKernelEvent @event)
         {
-            switch (value)
+            switch (@event)
             {
                 case ValueProduced valueProduced:
                     OnValueProduced(valueProduced, InFlightRequests);
@@ -101,8 +102,6 @@ void OnKernelResultEvent(IKernelEvent value)
                 case IncompleteCodeSubmissionReceived _:
                 case CompleteCodeSubmissionReceived _:
                     break;
-                default:
-                    throw new NotSupportedException();
             }
         }
 

Microsoft.DotNet.Interactive.Jupyter/JupyterRequestContextHandler.cs

@@ -2,7 +2,9 @@
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 
 using System;
+using System.Reactive.Concurrency;
 using System.Text.RegularExpressions;
+using System.Threading;
 using System.Threading.Tasks;
 using Clockwise;
 using Microsoft.DotNet.Interactive.Jupyter.Protocol;
@@ -24,8 +26,16 @@ public class JupyterRequestContextHandler : ICommandHandler<JupyterRequestContex
             PackageRegistry packageRegistry,
             IKernel kernel)
         {
-            _executeHandler = new ExecuteRequestHandler(kernel);
-            _completeHandler = new CompleteRequestHandler(kernel);
+            var scheduler = new EventLoopScheduler(t =>
+            {
+                var thread = new Thread(t);
+                thread.IsBackground = true;
+                thread.Name = "MessagePump";
+                return thread;
+            });
+
+            _executeHandler = new ExecuteRequestHandler(kernel, scheduler);
+            _completeHandler = new CompleteRequestHandler(kernel, scheduler);
 
             if (packageRegistry == null)
             {

Microsoft.DotNet.Interactive.Jupyter/RequestHandlerBase.cs

@@ -3,23 +3,32 @@
 
 using System;
 using System.Collections.Concurrent;
-using System.Collections.Generic;
+using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
+using System.Reactive.Linq;
 using Microsoft.DotNet.Interactive.Commands;
+using Microsoft.DotNet.Interactive.Events;
 using Microsoft.DotNet.Interactive.Jupyter.Protocol;
 
 namespace Microsoft.DotNet.Interactive.Jupyter
 {
     public abstract class RequestHandlerBase<T> : IDisposable
         where T : JupyterMessageContent
     {
+
         private readonly CompositeDisposable _disposables = new CompositeDisposable();
+        protected IObservable<IKernelEvent> _kernelEventSource { get; }
 
-        protected RequestHandlerBase(IKernel kernel)
+        protected RequestHandlerBase(IKernel kernel, IScheduler scheduler)
         {
             Kernel = kernel ?? throw new ArgumentNullException(nameof(kernel));
+
+            _kernelEventSource = Kernel.KernelEvents.ObserveOn(scheduler ?? throw new ArgumentNullException(nameof(scheduler)));
+            _disposables.Add(_kernelEventSource.Subscribe(OnKernelEvent));
         }
 
+        protected abstract void OnKernelEvent(IKernelEvent @event);
+
         protected static T GetJupyterRequest(JupyterRequestContext context)
         {
             var request = context.GetRequestContent<T>() ??

Microsoft.DotNet.Interactive.Jupyter/Shell.cs

@@ -81,7 +81,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
                     activity.Info("Received: {message}", message.ToJson());
 
                     var status = new RequestHandlerStatus(message.Header, new MessageSender(_ioPubSocket, _signatureValidator));
-                    
+
 
                     switch (message.Header.MessageType)
                     {
@@ -107,8 +107,6 @@ public async Task StartAsync(CancellationToken cancellationToken)
 
                             break;
                     }
-
-
                 }
             }
         }

Microsoft.DotNet.Interactive/Events/ValueProduced.cs

@@ -9,9 +9,9 @@ namespace Microsoft.DotNet.Interactive.Events
     public class ValueProduced : KernelEventBase
     {
         public ValueProduced(object value,
-            SubmitCode submitCode,
+            IKernelCommand command,
             bool isLastValue = false,
-            IReadOnlyCollection<FormattedValue> formattedValues = null) : base(submitCode)
+            IReadOnlyCollection<FormattedValue> formattedValues = null) : base(command)
         {
             Value = value;
             IsLastValue = isLastValue;

WorkspaceServer.Tests/Kernel/CSharpKernelTestBase.cs

@@ -8,6 +8,8 @@
 using Pocket;
 using WorkspaceServer.Kernel;
 using Xunit.Abstractions;
+using System.Reactive.Linq;
+using System.Reactive;
 
 namespace WorkspaceServer.Tests.Kernel
 {
@@ -24,14 +26,14 @@ protected CSharpKernel CreateKernel()
                 .LogEventsToPocketLogger();
 
             DisposeAfterTest(
-                kernel.KernelEvents.Subscribe(KernelEvents.Add));
+                kernel.KernelEvents.Timestamp().Subscribe(KernelEvents.Add));
 
             return kernel;
         }
 
         private readonly CompositeDisposable _disposables = new CompositeDisposable();
 
-        protected IList<IKernelEvent> KernelEvents { get; } = new List<IKernelEvent>();
+        protected IList<Timestamped<IKernelEvent>> KernelEvents { get; } = new List<Timestamped<IKernelEvent>>();
 
         protected void DisposeAfterTest(IDisposable disposable)
         {