Add "Open XEL file" support to profiler in sqltoolsservice (#2091)

* Open XEL file changes

* placeholders for openxel

* add observable xe reader

* md format tweaks

* implement localfile as a new session type

* add ErrorMessage to session stopped notice

* fix flaky test

* handle already running session

* fix stopped session event send on file completion

* fix flaky unit test

* Update XElite and dependent versions

* Fix errors after merge and remove failing tests for now

* Fix main merge mess-up.
Address comments.
Add one more relevant test.

* Remove extra namespace.

* Remove unnecessary import

* Fix build error

* Address comments.

* Remove disabiling JSON002 compiler warning

* Address comments and update json handling

* Fix build error

* Fix integration test (emerged due to Main merge mess up)

* Clean up code (no functional changes)

---------

Co-authored-by: Karl Burtram <karlb@microsoft.com>
Co-authored-by: shueybubbles <david.shiflet@microsoft.com>

Packages.props

@@ -48,6 +48,7 @@
 		<PackageReference Update="coverlet.collector" Version="3.1.2" />
 		<PackageReference Update="coverlet.msbuild" Version="3.1.2" />
 		<PackageReference Update="TextCopy" Version="6.2.1" />
+		<PackageReference Update="Microsoft.SqlServer.XEvent.XELite" Version="2023.1.30.3" />
 	</ItemGroup>
 
 	<!-- When updating version of Dependencies in the below section, please also update the version in the following files:

README.md

@@ -2,15 +2,19 @@
 [![Build Status](https://mssqltools.visualstudio.com/CrossPlatBuildScripts/_apis/build/status/Tools%20Service%20Integration%20tests?branchName=main&label=Integration%20Tests)](https://mssqltools.visualstudio.com/CrossPlatBuildScripts/_build/latest?definitionId=379&branchName=main)
 
 # Microsoft SQL Tools Service 
+
 The SQL Tools Service is an application that provides core functionality for various SQL Server tools.  These features include the following:
+
 * Connection management
 * Language Service support using VS Code protocol
 * Query execution and resultset management
 
 # SQL Tools Service API Documentation
+
 Please see the SQL Tools Service API documentation at https://microsoft.github.io/sqltoolssdk/.
 
 # Setup, Building and Testing the codebase
+
 Please see the SQL Tools Service wiki documentation at https://github.com/Microsoft/sqltoolsservice/wiki
 
 # Contribution Guidelines
@@ -104,10 +108,8 @@ so that your commits provide a good history of the changes you are making.  To b
 
 ### Add Unit Tests for New Code
 
-If you're adding a new feature to the project, please make sure to include adequate [xUnit](http://xunit.github.io/)
-tests with your change.  In this project, we have chosen write out unit tests in a way that uses the
-actual PowerShell environment rather than extensive interface mocking.  This allows us to be sure that
-our features will work in practice.  
+If you're adding a new feature to the project, please make sure to include adequate [nUnit](http://nunit.org/)
+tests with your change.
 
 We do both component-level and scenario-level testing depending on what code is being tested.  We don't
 expect contributors to test every possible edge case.  Testing mainline scenarios and the most common

docs/guide/toc.md

@@ -1,5 +1,11 @@
-# [Introduction](introduction.md)
-# [SQL Tools JSON-RPC Protocol](jsonrpc_protocol.md)
-# [Using the JSON-RPC API](using_the_jsonrpc_api.md)
-# [Building the SQL Tools API](building_sqltoolsservice.md)
-# [Using the .NET API](using_the_dotnet_api.md)
+# Table of Contents
+
+## [Introduction](introduction.md)
+
+## [SQL Tools JSON-RPC Protocol](jsonrpc_protocol.md)
+
+## [Using the JSON-RPC API](using_the_jsonrpc_api.md)
+
+## [Building the SQL Tools API](building_sqltoolsservice.md)
+
+## [Using the .NET API](using_the_dotnet_api.md)

docs/guide/using_the_jsonrpc_api.md

@@ -1,4 +1,5 @@
 # Using the SQL Tools JSON-RPC API
+
 The SQL Tools JSON-RPC API is the best way to consume the services
 functionality in SQL tools.  The JSON-RPC API available through stdio
 of the SQL Tools Service process.

src/Microsoft.SqlTools.ServiceLayer/Microsoft.SqlTools.ServiceLayer.csproj

@@ -50,6 +50,7 @@
 		<PackageReference Include="Microsoft.SqlServer.Management.SqlParser" />
 		<PackageReference Include="System.Configuration.ConfigurationManager" />
 		<PackageReference Include="System.Text.Encoding.CodePages" />
+		<PackageReference Include="Microsoft.SqlServer.XEvent.XELite" />
 		<PackageReference Include="Microsoft.SqlServer.TransactSql.ScriptDom.NRT">
 			<Aliases>ASAScriptDom</Aliases>
 		</PackageReference>

src/Microsoft.SqlTools.ServiceLayer/Profiler/Contracts/GetXEventSessionsRequest.cs

@@ -22,7 +22,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler.Contracts
     public class GetXEventSessionsResult
     {
         /// <summary>
-        /// Session ID that was started
+        /// List of XE session names
         /// </summary>
         public List<string> Sessions { get; set; }
     }

src/Microsoft.SqlTools.ServiceLayer/Profiler/Contracts/ProfilerEventsAvailableNotification.cs

@@ -19,6 +19,9 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler.Contracts
         public bool EventsLost { get; set; }
     }
 
+    /// <summary>
+    /// Profiler Event available notification mapping entry 
+    /// </summary>
     public class ProfilerEventsAvailableNotification
     {
         public static readonly

src/Microsoft.SqlTools.ServiceLayer/Profiler/Contracts/ProfilerSessionCreatedNotification.cs

@@ -18,6 +18,9 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler.Contracts
         public string TemplateName { get; set; }
     }
 
+    /// <summary>
+    /// Profiler Session created notification mapping entry 
+    /// </summary>
     public class ProfilerSessionCreatedNotification
     {
         public static readonly

src/Microsoft.SqlTools.ServiceLayer/Profiler/Contracts/ProfilerSessionStoppedNotification.cs

@@ -13,9 +13,25 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler.Contracts
     {
         public string OwnerUri { get; set; }
 
+        /// <summary>
+        /// Numeric session id that is only unique on the server where the session resides
+        /// </summary>
         public int SessionId { get; set; }
+
+        /// <summary>
+        /// An key that uniquely identifies a session across all servers
+        /// </summary>
+        public string UniqueSessionId { get; set; }
+
+        /// <summary>
+        /// The error that stopped the session, if any.
+        /// </summary>
+        public string ErrorMessage { get; set; }
     }
 
+    /// <summary>
+    /// Profiler Session stopped notification mapping entry 
+    /// </summary>
     public class ProfilerSessionStoppedNotification
     {
         public static readonly

src/Microsoft.SqlTools.ServiceLayer/Profiler/Contracts/StartProfilingRequest.cs

@@ -17,10 +17,39 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler.Contracts
     {
         public string OwnerUri { get; set; }
 
+        /// <summary>
+        /// For RemoteSession sessions, the name of the remote session.
+        /// For LocalFile sessions, the full path of the XEL file to open.
+        /// </summary>
         public string SessionName { get; set; }
+
+        /// <summary>
+        /// Identifies which type of target the session name identifies.
+        /// </summary>
+        public ProfilingSessionType SessionType { get; set; } = ProfilingSessionType.RemoteSession;
     }
 
-    public class StartProfilingResult{}
+    public enum ProfilingSessionType
+    {
+        RemoteSession,
+        LocalFile
+    }
+
+    /// <summary>
+    /// Provides information about the session that was started
+    /// </summary>
+    public class StartProfilingResult
+    {
+        /// <summary>
+        /// A unique key to identify the session
+        /// </summary>
+        public string UniqueSessionId { get; set; }
+
+        /// <summary>
+        /// Whether the profiling session supports the Pause operation.
+        /// </summary>
+        public bool CanPause { get; set; }
+    }
 
     /// <summary>
     /// Start Profile request type

src/Microsoft.SqlTools.ServiceLayer/Profiler/IProfilerSessionListener.cs

@@ -14,6 +14,6 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
     {
         void EventsAvailable(string sessionId, List<ProfilerEvent> events, bool eventsLost);
 
-        void SessionStopped(string viewerId, int sessionId);
+        void SessionStopped(string viewerId, SessionId sessionId, string errorMessage);
     }
 }

src/Microsoft.SqlTools.ServiceLayer/Profiler/IXEventSession.cs

@@ -5,6 +5,9 @@
 
 #nullable disable
 
+using System;
+using System.Threading;
+
 namespace Microsoft.SqlTools.ServiceLayer.Profiler
 {
     /// <summary>
@@ -15,7 +18,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// <summary>
         /// Gets unique XEvent session Id
         /// </summary>
-        int Id { get; }
+        SessionId Id { get; }
 
         /// <summary>
         /// Starts XEvent session
@@ -32,4 +35,47 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// </summary>
         string GetTargetXml();
     }
+
+    public interface IObservableXEventSession : IXEventSession
+    {
+        IObservable<Contracts.ProfilerEvent> ObservableSessionEvents { get; }
+    }
+
+    /// <summary>
+    /// Strong type for use as a dictionary key, simply wraps a string.
+    /// Using a class helps distinguish instances from other strings like viewer id.
+    /// </summary>
+    public class SessionId
+    {
+        private readonly string sessionId;
+
+        // SQL Server starts session counters at around 64k, so it's unlikely that this process-scoped counter would collide with a real session id
+        // Eventually the profiler extension in ADS will use the string instead of the number.
+        private static int numericIdCurrent = -1;
+
+        /// <summary>
+        /// Constructs a new sessionId 
+        /// </summary>
+        /// <param name="sessionId">The true unique identifier string, opaque to the client.</param>
+        /// <param name="numericId">An optional numeric identifier used to identify the session to older clients that don't consume the string yet</param>
+        public SessionId(string sessionId, int? numericId = null)
+        {
+            this.sessionId = sessionId ?? throw new ArgumentNullException(nameof(sessionId));
+            NumericId = numericId ?? Interlocked.Increment(ref numericIdCurrent);
+        }
+
+        public int NumericId { get; private set; }
+
+        public override int GetHashCode() => sessionId.GetHashCode();
+
+        public override bool Equals(object obj)
+        {
+            return (obj is SessionId id) && id.sessionId.Equals(sessionId);
+        }
+
+        public override string ToString()
+        {
+            return sessionId;
+        }
+    }
 }

src/Microsoft.SqlTools.ServiceLayer/Profiler/IXEventSessionFactory.cs

@@ -24,5 +24,12 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// Creates an XEvent session with the given create statement and name
         /// </summary>
         IXEventSession CreateXEventSession(string createStatement, string sessionName, ConnectionInfo connInfo);
+
+        /// <summary>
+        /// Opens a session whose events are streamed from a local XEL file
+        /// </summary>
+        /// <param name="filePath"></param>
+        /// <returns></returns>
+        IXEventSession OpenLocalFileSession(string filePath);
     }
 }

src/Microsoft.SqlTools.ServiceLayer/Profiler/ObservableXEventSession.cs

@@ -0,0 +1,170 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+#nullable disable
+
+using Microsoft.SqlServer.XEvent.XELite;
+using Microsoft.SqlTools.ServiceLayer.Profiler.Contracts;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.SqlTools.ServiceLayer.Profiler
+{
+    /// <summary>
+    /// Wrapper XEventSession for IXEventFetcher instances
+    /// </summary>
+    class ObservableXEventSession : XEventSession, IObservableXEventSession
+    {
+        private readonly XeStreamObservable observableSession;
+        private readonly SessionId sessionId;
+        public IObservable<ProfilerEvent> ObservableSessionEvents => observableSession;
+        
+        public override void Start()
+        {
+            Session?.Start();
+            observableSession.Start();
+        }
+
+        public override void Stop()
+        {
+            observableSession.Close();
+            Session?.Stop();
+        }
+
+        public ObservableXEventSession(Func<IXEventFetcher> xeventFetcher, SessionId sessionId)
+        {
+            observableSession = new XeStreamObservable(xeventFetcher);
+            this.sessionId = sessionId;
+        }
+
+        protected override SessionId GetSessionId()
+        {
+            return sessionId;
+        }
+    }
+
+    /// <summary>
+    /// Source of ProfilerEvent push notifications. Wraps IXEventFetcher.
+    /// </summary>
+    public class XeStreamObservable : IObservable<ProfilerEvent>
+    {
+        private readonly object syncObj = new object();
+        private readonly List<IObserver<ProfilerEvent>> observers = new List<IObserver<ProfilerEvent>>();
+        private CancellationTokenSource cancellationTokenSource;
+        private readonly Func<IXEventFetcher> xeventFetcher;
+
+        /// <summary>
+        /// Constructs a new XeStreamObservable that converts xevent data from the fetcher to ProfilerEvent instances
+        /// </summary>
+        /// <param name="fetcher"></param>
+        public XeStreamObservable(Func<IXEventFetcher> fetcher)
+        {
+            xeventFetcher = fetcher;
+        }
+
+        /// <summary>
+        /// Starts processing xevents from the source.
+        /// </summary>
+        public void Start()
+        {
+            try
+            {
+                cancellationTokenSource = new CancellationTokenSource();
+                var xeventFetcherFuncCallBack = xeventFetcher();
+                var xeventFetcherTask = xeventFetcherFuncCallBack.ReadEventStream(OnEventRead, cancellationTokenSource.Token);
+                xeventFetcherTask.ContinueWith(OnStreamClosed);
+            } catch (Exception ex)
+            {
+                Task.FromException<IXEventFetcher>(ex).ContinueWith(OnStreamClosed);
+            }
+            
+        }
+
+        /// <summary>
+        /// Stops the xevent fetching task and informs all listeners that the event stream has ended and clears the list of listeners.
+        /// Start could be called again, but only new subscribers will see the data.
+        /// </summary>
+        public void Close()
+        {
+            cancellationTokenSource.Cancel();
+            var currentObservers = CurrentObservers;
+            currentObservers.ForEach(o => o.OnCompleted());
+            lock (syncObj)
+            {
+                currentObservers.ForEach(o => observers.Remove(o));
+            }
+        }
+
+        /// <summary>
+        /// Adds the observer to the listener list
+        /// </summary>
+        /// <param name="observer"></param>
+        /// <returns>An IDisposable for the listener to call when it no longer wishes to receive events</returns>
+        public IDisposable Subscribe(IObserver<ProfilerEvent> observer)
+        {
+            lock (syncObj)
+            {
+                if (!observers.Contains(observer))
+                {
+                    observers.Add(observer);
+                }
+                return new Unsubscriber(observers, observer);
+            }
+        }
+
+        private List<IObserver<ProfilerEvent>> CurrentObservers
+        {
+            get
+            {
+                lock (syncObj)
+                {
+                    return new List<IObserver<ProfilerEvent>>(observers);
+                }
+            }
+        }
+
+        private void OnStreamClosed(Task fetcherTask)
+        {
+            if (fetcherTask.IsFaulted)
+            {
+                CurrentObservers.ForEach(o => o.OnError(fetcherTask.Exception));
+            }
+            Close();
+        }
+
+        private Task OnEventRead(IXEvent xEvent)
+        {
+            ProfilerEvent profileEvent = new ProfilerEvent(xEvent.Name, xEvent.Timestamp.ToString());
+            foreach (var kvp in xEvent.Fields)
+            {
+                profileEvent.Values.Add(kvp.Key, kvp.Value.ToString());
+            }
+            CurrentObservers.ForEach(o => o.OnNext(profileEvent));
+            return Task.FromResult(0);
+        }
+
+        private class Unsubscriber : IDisposable
+        {
+            private readonly List<IObserver<ProfilerEvent>> _observers;
+            private readonly IObserver<ProfilerEvent> _observer;
+
+            public Unsubscriber(List<IObserver<ProfilerEvent>> observers, IObserver<ProfilerEvent> observer)
+            {
+                _observers = observers;
+                _observer = observer;
+            }
+
+            public void Dispose()
+            {
+                if (_observer != null && _observers.Contains(_observer))
+                {
+                    _observers.Remove(_observer);
+                }
+            }
+        }
+    }
+}

src/Microsoft.SqlTools.ServiceLayer/Profiler/ProfilerException.cs

@@ -0,0 +1,19 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+using System;
+
+namespace Microsoft.SqlTools.ServiceLayer.Profiler
+{
+    class ProfilerException : Exception
+    {
+        public ProfilerException(string message) : base(message)
+        {
+        }
+
+        public ProfilerException(string message, Exception innerException) : base(message, innerException)
+        {
+        }
+    }
+}

src/Microsoft.SqlTools.ServiceLayer/Profiler/ProfilerService.cs

@@ -8,11 +8,11 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.SqlServer.Management.Sdk.Sfc;
 using Microsoft.SqlServer.Management.XEvent;
 using Microsoft.SqlServer.Management.XEventDbScoped;
+using Microsoft.SqlServer.XEvent.XELite;
 using Microsoft.SqlTools.Hosting.Protocol;
 using Microsoft.SqlTools.ServiceLayer.Connection;
 using Microsoft.SqlTools.ServiceLayer.Hosting;
@@ -120,7 +120,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                 out connInfo);
             if (connInfo == null)
             {
-                throw new Exception(SR.ProfilerConnectionNotFound);
+                throw new ProfilerException(SR.ProfilerConnectionNotFound);
             }
             else if (parameters.SessionName == null)
             {
@@ -143,7 +143,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                 }
                 catch { }
 
-                // create a new XEvent session and Profiler session
+                // create a new XEvent session and Profiler session, if it doesn't exist
                 xeSession ??= this.XEventSessionFactory.CreateXEventSession(parameters.Template.CreateStatement, parameters.SessionName, connInfo);
 
                 // start monitoring the profiler session
@@ -161,6 +161,11 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// </summary>
         internal async Task HandleStartProfilingRequest(StartProfilingParams parameters, RequestContext<StartProfilingResult> requestContext)
         {
+            if (parameters.SessionType == ProfilingSessionType.LocalFile)
+            {
+                await StartLocalFileSession(parameters, requestContext);
+                return;
+            }
             ConnectionInfo connInfo;
             ConnectionServiceInstance.TryFindConnection(
                 parameters.OwnerUri,
@@ -169,25 +174,44 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             {
                 // create a new XEvent session and Profiler session
                 var xeSession = this.XEventSessionFactory.GetXEventSession(parameters.SessionName, connInfo);
+
                 // start monitoring the profiler session
                 monitor.StartMonitoringSession(parameters.OwnerUri, xeSession);
 
-                var result = new StartProfilingResult();
+                var result = new StartProfilingResult() { CanPause = true, UniqueSessionId = xeSession.Id.ToString() };
                 await requestContext.SendResult(result);
             }
             else
             {
-                throw new Exception(SR.ProfilerConnectionNotFound);
+                throw new ProfilerException(SR.ProfilerConnectionNotFound);
             }
         }
 
+        private async Task StartLocalFileSession(StartProfilingParams parameters, RequestContext<StartProfilingResult> requestContext)
+        {
+            var xeSession = XEventSessionFactory.OpenLocalFileSession(parameters.SessionName);
+            monitor.StartMonitoringSession(parameters.OwnerUri, xeSession);
+            xeSession.Start();
+            var result = new StartProfilingResult() { UniqueSessionId = xeSession.Id.ToString(), CanPause = false };
+            await requestContext.SendResult(result);
+        }
+
+        public IXEventSession OpenLocalFileSession(string filePath)
+        {
+            return new ObservableXEventSession(() => initIXEventFetcher(filePath), new SessionId(filePath));
+        }
+
+        public IXEventFetcher initIXEventFetcher(string filePath)
+        {
+            return new XEFileEventStreamer(filePath);
+        }
+
         /// <summary>
         /// Handle request to stop a profiling session
         /// </summary>
         internal async Task HandleStopProfilingRequest(StopProfilingParams parameters, RequestContext<StopProfilingResult> requestContext)
         {
-            ProfilerSession session;
-            monitor.StopMonitoringSession(parameters.OwnerUri, out session);
+            monitor.StopMonitoringSession(parameters.OwnerUri, out ProfilerSession session);
 
             if (session != null)
             {
@@ -199,6 +223,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                     try
                     {
                         session.XEventSession.Stop();
+                        session.Dispose();
                         await requestContext.SendResult(new StopProfilingResult { });
                         break;
                     }
@@ -209,13 +234,13 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                         {
                             throw;
                         }
-                        Thread.Sleep(500);
+                        await Task.Delay(500);
                     }
                 }
             }
             else
             {
-                throw new Exception(SR.SessionNotFound);
+                throw new ProfilerException(SR.SessionNotFound);
             }
         }
 
@@ -241,11 +266,11 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                 out connInfo);
             if (connInfo == null)
             {
-                await requestContext.SendError(new Exception(SR.ProfilerConnectionNotFound));
+                await requestContext.SendError(new ProfilerException(SR.ProfilerConnectionNotFound));
             }
             else
             {
-                List<string> sessions = GetXEventSessionList(parameters.OwnerUri, connInfo);
+                List<string> sessions = GetXEventSessionList(connInfo);
                 result.Sessions = sessions;
                 await requestContext.SendResult(result);
             }
@@ -254,9 +279,10 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// <summary>
         /// Handle request to disconnect a session
         /// </summary>
-        internal async Task HandleDisconnectSessionRequest(DisconnectSessionParams parameters, RequestContext<DisconnectSessionResult> requestContext)
+        internal Task HandleDisconnectSessionRequest(DisconnectSessionParams parameters, RequestContext<DisconnectSessionResult> requestContext)
         {
             monitor.StopMonitoringSession(parameters.OwnerUri, out _);
+            return Task.CompletedTask;
         }
 
         /// <summary>
@@ -265,7 +291,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// <returns>
         /// A list of the names of all running XEvent sessions
         /// </returns>
-        internal List<string> GetXEventSessionList(string ownerUri, ConnectionInfo connInfo)
+        internal List<string> GetXEventSessionList(ConnectionInfo connInfo)
         {
             var sqlConnection = ConnectionService.OpenSqlConnection(connInfo);
             SqlStoreConnection connection = new SqlStoreConnection(sqlConnection);
@@ -311,16 +337,19 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             Session session = store.Sessions[sessionName] ?? throw new Exception(SR.SessionNotFound);
 
             // start the session if it isn't already running
-            if (session != null && !session.IsRunning)
-            {
-                session.Start();
-            }
 
-            // create xevent session wrapper
-            return new XEventSession()
+            session = session ?? throw new ProfilerException(SR.SessionNotFound);
+
+            var xeventSession = new XEventSession()
             {
                 Session = session
             };
+
+            if (!session.IsRunning)
+            {
+                xeventSession.Start();
+            }
+            return xeventSession;
         }
 
         /// <summary>
@@ -336,7 +365,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             // session shouldn't already exist
             if (session != null)
             {
-                throw new Exception(SR.SessionAlreadyExists(sessionName));
+                throw new ProfilerException(SR.SessionAlreadyExists(sessionName));
             }
 
             var statement = createStatement.Replace("{sessionName}", sessionName);
@@ -345,7 +374,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             session = store.Sessions[sessionName];
             if (session == null)
             {
-                throw new Exception(SR.SessionNotFound);
+                throw new ProfilerException(SR.SessionNotFound);
             }
             if (!session.IsRunning)
             {
@@ -378,7 +407,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// <summary>
         /// Callback when the XEvent session is closed unexpectedly
         /// </summary>
-        public void SessionStopped(string viewerId, int sessionId)
+        public void SessionStopped(string viewerId, SessionId sessionId, string errorMessage)
         {
             // notify the client that their session closed
             this.ServiceHost.SendEvent(
@@ -386,7 +415,8 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                 new ProfilerSessionStoppedParams()
                 {
                     OwnerUri = viewerId,
-                    SessionId = sessionId
+                    SessionId = sessionId.NumericId,
+                    ErrorMessage = errorMessage
                 });
         }
 

src/Microsoft.SqlTools.ServiceLayer/Profiler/ProfilerSession.cs

@@ -7,7 +7,10 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
+using System.Threading;
+using System.Xml;
 using Microsoft.SqlTools.ServiceLayer.Connection;
 using Microsoft.SqlTools.ServiceLayer.Profiler.Contracts;
 
@@ -16,29 +19,43 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
     /// <summary>
     /// Profiler session class
     /// </summary>
-    public class ProfilerSession
+    public class ProfilerSession : IDisposable
     {
         private static readonly TimeSpan DefaultPollingDelay = TimeSpan.FromSeconds(1);
         private object pollingLock = new object();
         private bool isPolling = false;
         private DateTime lastPollTime = DateTime.Now.Subtract(DefaultPollingDelay);
-        private TimeSpan pollingDelay = DefaultPollingDelay;
         private ProfilerEvent lastSeenEvent = null;
-
+        private readonly SessionObserver sessionObserver;
+        private readonly IXEventSession xEventSession;
+        private readonly IDisposable observerDisposable;
         private bool eventsLost = false;
         int lastSeenId = -1;
 
-        public bool pollImmediatly = false;
+        public bool pollImmediately = false;
 
         /// <summary>
         /// Connection to use for the session
         /// </summary>
         public ConnectionInfo ConnectionInfo { get; set; }
 
+        /// <summary>
+        /// Constructs a new ProfilerSession to watch the given IXeventSession's incoming events
+        /// </summary>
+        /// <param name="xEventSession"></param>
+        public ProfilerSession(IXEventSession xEventSession)
+        {
+            this.xEventSession = xEventSession;
+            if (xEventSession is IObservableXEventSession observableSession)
+            {
+                observerDisposable = observableSession.ObservableSessionEvents?.Subscribe(sessionObserver = new SessionObserver());
+            }
+        }    
+
         /// <summary>
         /// Underlying XEvent session wrapper
         /// </summary>
-        public IXEventSession XEventSession { get; set; }
+        public IXEventSession XEventSession => xEventSession;
 
         /// <summary>
         /// Try to set the session into polling mode if criteria is meet
@@ -48,11 +65,11 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         {
             lock (this.pollingLock)
             {
-                if (pollImmediatly || (!this.isPolling && DateTime.Now.Subtract(this.lastPollTime) >= pollingDelay))
+                if (pollImmediately || (!this.isPolling && DateTime.Now.Subtract(this.lastPollTime) >= PollingDelay))
                 {
                     this.isPolling = true;
                     this.lastPollTime = DateTime.Now;
-                    this.pollImmediatly = false;
+                    this.pollImmediately = false;
                     return true;
                 }
                 else
@@ -83,13 +100,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         /// <summary>
         /// The delay between session polls
         /// </summary>
-        public TimeSpan PollingDelay
-        {
-            get
-            {
-                return pollingDelay;
-            }
-        }
+        public TimeSpan PollingDelay { get; } = DefaultPollingDelay;
 
         /// <summary>
         /// Could events have been lost in the last poll
@@ -197,5 +208,129 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                 events.Clear();
             }
         }
+
+        /// <summary>
+        /// Indicates if the current session has completed processing and will provide no new events
+        /// </summary>
+        public bool Completed
+        {
+            get
+            {
+                return (sessionObserver != null) ? sessionObserver.Completed : error != null;
+            }
+        }
+
+        private Exception error;
+        /// <summary>
+        /// Provides any fatal error encountered when processing a session
+        /// </summary>
+        public Exception Error
+        {
+            get
+            {
+                return sessionObserver?.Error ?? error;
+            }
+        }
+
+        /// Returns the current set of events in the session buffer.
+        /// For RingBuffer sessions, returns the content of the session ring buffer by querying the server.
+        /// For LiveTarget and LocalFile sessions, returns the events buffered in memory since the last call to GetCurrentEvents.
+        /// </summary>
+        /// <returns></returns>
+        public IEnumerable<ProfilerEvent> GetCurrentEvents()
+        {
+            if (XEventSession == null && sessionObserver == null)
+            {
+                return Enumerable.Empty<ProfilerEvent>();
+            }
+            if (sessionObserver == null)
+            {
+                try
+                {
+                    var targetXml = XEventSession.GetTargetXml();
+
+                    XmlDocument xmlDoc = new XmlDocument();
+                    xmlDoc.LoadXml(targetXml);
+
+                    var nodes = xmlDoc.DocumentElement.GetElementsByTagName("event");
+                    var rawEvents = nodes.Cast<XmlNode>().Select(ParseProfilerEvent).ToList();
+                    FilterOldEvents(rawEvents);
+                    return rawEvents;
+                }
+                catch (Exception e)
+                {
+                    error ??= e;
+                    return Enumerable.Empty<ProfilerEvent>();
+                }
+            }
+            return sessionObserver.CurrentEvents;
+        }
+
+        /// <summary>
+        /// Parse a single event node from XEvent XML
+        /// </summary>
+        private static ProfilerEvent ParseProfilerEvent(XmlNode node)
+        {
+            var name = node.Attributes["name"];
+            var timestamp = node.Attributes["timestamp"];
+
+            var profilerEvent = new ProfilerEvent(name.InnerText, timestamp.InnerText);
+
+            foreach (XmlNode childNode in node.ChildNodes)
+            {
+                var childName = childNode.Attributes["name"];
+                XmlNode typeNode = childNode.SelectSingleNode("type");
+                var typeName = typeNode.Attributes["name"];
+                XmlNode valueNode = childNode.SelectSingleNode("value");
+
+                if (!profilerEvent.Values.ContainsKey(childName.InnerText))
+                {
+                    profilerEvent.Values.Add(childName.InnerText, valueNode.InnerText);
+                }
+            }
+
+            return profilerEvent;
+        }
+
+        public void Dispose()
+        {
+            observerDisposable?.Dispose();
+        }
+    }
+
+    [DebuggerDisplay("SessionObserver. Current:{writeBuffer.Count} Total:{eventCount}")]
+    class SessionObserver : IObserver<ProfilerEvent>
+    {
+        private List<ProfilerEvent> writeBuffer = new List<ProfilerEvent>();
+        private Int64 eventCount = 0;
+        public void OnCompleted()
+        {
+            Completed = true;
+        }
+
+        public void OnError(Exception error)
+        {
+            Error = error;
+        }
+
+        public void OnNext(ProfilerEvent value)
+        {
+            writeBuffer.Add(value);
+            eventCount++;
+        }
+
+        public bool Completed { get; private set; }
+
+        public Exception Error { get; private set; }
+
+        public IEnumerable<ProfilerEvent> CurrentEvents
+        {
+            get
+            {
+                var newBuffer = new List<ProfilerEvent>();
+                var oldBuffer = Interlocked.Exchange(ref writeBuffer, newBuffer);
+                return oldBuffer;
+            }
+        }
     }
 }

src/Microsoft.SqlTools.ServiceLayer/Profiler/ProfilerSessionMonitor.cs

@@ -7,14 +7,10 @@
 
 using System;
 using System.Collections.Generic;
-using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using System.Xml;
-using Microsoft.SqlServer.Management.XEvent;
 using Microsoft.SqlTools.ServiceLayer.Profiler.Contracts;
-using Microsoft.SqlTools.Utility;
 
 namespace Microsoft.SqlTools.ServiceLayer.Profiler
 {
@@ -38,9 +34,9 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             public string Id { get; set; }
             public bool active { get; set; }
 
-            public int xeSessionId { get; set; }
+            public SessionId xeSessionId { get; set; }
 
-            public Viewer(string Id, bool active, int xeId)
+            public Viewer(string Id, bool active, SessionId xeId)
             {
                 this.Id = Id;
                 this.active = active;
@@ -49,15 +45,15 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         };
 
         // XEvent Session Id's matched to the Profiler Id's watching them
-        private Dictionary<int, List<string>> sessionViewers = new Dictionary<int, List<string>>();
+        private readonly Dictionary<SessionId, List<string>> sessionViewers = new Dictionary<SessionId, List<string>>();
 
         // XEvent Session Id's matched to their Profiler Sessions
-        private Dictionary<int, ProfilerSession> monitoredSessions = new Dictionary<int, ProfilerSession>();
+        private readonly Dictionary<SessionId, ProfilerSession> monitoredSessions = new Dictionary<SessionId, ProfilerSession>();
 
         // ViewerId -> Viewer objects
-        private Dictionary<string, Viewer> allViewers = new Dictionary<string, Viewer>();
+        private readonly Dictionary<string, Viewer> allViewers = new Dictionary<string, Viewer>();
 
-        private List<IProfilerSessionListener> listeners = new List<IProfilerSessionListener>();
+        private readonly List<IProfilerSessionListener> listeners = new List<IProfilerSessionListener>();
 
         /// <summary>
         /// Registers a session event Listener to receive a callback when events arrive
@@ -83,8 +79,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                 // create new profiling session if needed
                 if (!this.monitoredSessions.ContainsKey(session.Id))
                 {
-                    var profilerSession = new ProfilerSession();
-                    profilerSession.XEventSession = session;
+                    var profilerSession = new ProfilerSession(session);
 
                     this.monitoredSessions.Add(session.Id, profilerSession);
                 }
@@ -151,7 +146,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             }
         }
 
-        private bool RemoveSession(int sessionId, out ProfilerSession session)
+        private bool RemoveSession(SessionId sessionId, out ProfilerSession session)
         {
             lock (this.sessionsLock)
             {
@@ -181,11 +176,11 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             }
         }
 
-        public void PollSession(int sessionId)
+        public void PollSession(SessionId sessionId)
         {
             lock (this.sessionsLock)
             {
-                this.monitoredSessions[sessionId].pollImmediatly = true;
+                this.monitoredSessions[sessionId].pollImmediately = true;
             }
             lock (this.pollingLock)
             {
@@ -228,20 +223,33 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             {
                 Task.Factory.StartNew(() =>
                 {
-                    var events = PollSession(session);
-                    bool eventsLost = session.EventsLost;
-                    if (events.Count > 0 || eventsLost)
+                    try
                     {
-                        // notify all viewers for the polled session
-                        List<string> viewerIds = this.sessionViewers[session.XEventSession.Id];
-                        foreach (string viewerId in viewerIds)
+                        var events = PollSession(session);
+                        bool eventsLost = session.EventsLost;
+                        if (events.Count > 0 || eventsLost)
                         {
-                            if (allViewers[viewerId].active)
+                            // notify all viewers for the polled session
+                            List<string> viewerIds = this.sessionViewers[session.XEventSession.Id];
+                            foreach (string viewerId in viewerIds)
                             {
-                                SendEventsToListeners(viewerId, events, eventsLost);
+                                if (allViewers[viewerId].active)
+                                {
+                                    SendEventsToListeners(viewerId, events, eventsLost);
+                                }
                             }
                         }                        
                     }
+                    finally
+                    {
+                        session.IsPolling = false;
+                    }
+                    if (session.Completed)
+                    {
+                        SendStoppedSessionInfoToListeners(session.XEventSession.Id, session.Error?.Message);
+                        RemoveSession(session.XEventSession.Id, out ProfilerSession tempSession);
+                        tempSession.Dispose();
+                    }
                 });
             }
         }
@@ -249,51 +257,20 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
         private List<ProfilerEvent> PollSession(ProfilerSession session)
         {
             var events = new List<ProfilerEvent>();
-            try
+            if (session == null || session.XEventSession == null)
             {
-                if (session == null || session.XEventSession == null)
-                {
-                    return events;
-                }
-
-                var targetXml = session.XEventSession.GetTargetXml();
-
-                XmlDocument xmlDoc = new XmlDocument();
-                xmlDoc.LoadXml(targetXml);
-
-                var nodes = xmlDoc.DocumentElement.GetElementsByTagName("event");
-                foreach (XmlNode node in nodes)
-                {
-                    var profilerEvent = ParseProfilerEvent(node);
-                    if (profilerEvent != null)
-                    {
-                        events.Add(profilerEvent);
-                    }
-                }
-            }
-            catch (XEventException)
-            {
-                SendStoppedSessionInfoToListeners(session.XEventSession.Id);
-                ProfilerSession tempSession;
-                RemoveSession(session.XEventSession.Id, out tempSession);
-            }
-            catch (Exception ex)
-            {
-                Logger.Write(TraceEventType.Warning, "Failed to poll session. error: " + ex.Message);
-            }
-            finally
-            {
-                session.IsPolling = false;
+                return events;
             }
 
-            session.FilterOldEvents(events);
+            events.AddRange(session.GetCurrentEvents());
+            
             return session.FilterProfilerEvents(events);
         }
 
         /// <summary>
         /// Notify listeners about closed sessions
         /// </summary>
-        private void SendStoppedSessionInfoToListeners(int sessionId)
+        private void SendStoppedSessionInfoToListeners(SessionId sessionId, string errorMessage)
         {
             lock (listenersLock)
             {
@@ -301,7 +278,7 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
                 {
                     foreach(string viewerId in sessionViewers[sessionId])
                     {
-                        listener.SessionStopped(viewerId, sessionId);
+                        listener.SessionStopped(viewerId, sessionId, errorMessage);
                     }
                 }
             }
@@ -321,30 +298,5 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
             }
         }
 
-        /// <summary>
-        /// Parse a single event node from XEvent XML
-        /// </summary>
-        private ProfilerEvent ParseProfilerEvent(XmlNode node)
-        {
-            var name = node.Attributes["name"];
-            var timestamp = node.Attributes["timestamp"];
-
-            var profilerEvent = new ProfilerEvent(name.InnerText, timestamp.InnerText);
-
-            foreach (XmlNode childNode in node.ChildNodes)
-            {
-                var childName = childNode.Attributes["name"];
-                XmlNode typeNode = childNode.SelectSingleNode("type");
-                var typeName = typeNode.Attributes["name"];
-                XmlNode valueNode = childNode.SelectSingleNode("value");
-
-                if (!profilerEvent.Values.ContainsKey(childName.InnerText))
-                {
-                    profilerEvent.Values.Add(childName.InnerText, valueNode.InnerText);
-                }
-            }
-
-            return profilerEvent;
-        }
     }
 }

src/Microsoft.SqlTools.ServiceLayer/Profiler/XEventSession.cs

@@ -17,22 +17,27 @@ namespace Microsoft.SqlTools.ServiceLayer.Profiler
     {
         public Session Session { get; set; }
 
-        public int Id
+        private SessionId sessionId;
+        public SessionId Id
         {
-            get { return Session.ID; }
+            get { return sessionId ??= GetSessionId(); }
         }
 
-        public void Start()
+        protected virtual SessionId GetSessionId()
+        {
+            return new SessionId($"{Session.Parent.Name}_{Session.ID}", Session?.ID);
+        }
+        public virtual void Start()
         {
             this.Session.Start();
         }
 
-        public void Stop()
+        public virtual void Stop()
         {
             this.Session.Stop();
         }
 
-        public string GetTargetXml()
+        public virtual string GetTargetXml()
         {
             if (this.Session == null)
             {

test/Microsoft.SqlTools.ServiceLayer.IntegrationTests/Connection/ConnectionServiceTests.cs

@@ -53,7 +53,7 @@ namespace Microsoft.SqlTools.ServiceLayer.IntegrationTests.Connection
                 query.Execute();
                 query.ExecutionTask.Wait();
 
-                // We should still have 2 DbConnections
+                // We should see 1 DbConnections
                 Assert.AreEqual(1, connectionInfo.CountConnections);
 
                 // If we disconnect, we should remain in a consistent state to do it over again

test/Microsoft.SqlTools.ServiceLayer.IntegrationTests/Profiler/ProfilerServiceTests.cs

@@ -15,71 +15,124 @@ using Microsoft.SqlTools.ServiceLayer.Profiler.Contracts;
 using Microsoft.SqlTools.ServiceLayer.Test.Common;
 using Moq;
 using NUnit.Framework;
+using Microsoft.SqlServer.Management.Sdk.Sfc;
+using System.Linq;
+using Microsoft.SqlTools.ServiceLayer.Connection;
+using Microsoft.SqlServer.Management.XEvent;
 
 namespace Microsoft.SqlTools.ServiceLayer.IntegrationTests.Profiler
 {
     public class ProfilerServiceTests
     {
+
         /// <summary>
         /// Verify that a start profiling request starts a profiling session
         /// </summary>
-        //[Test]
+        [Test]
         public async Task TestHandleStartAndStopProfilingRequests()
         {
             using (SelfCleaningTempFile queryTempFile = new SelfCleaningTempFile())
             {
+
                 var connectionResult = await LiveConnectionHelper.InitLiveConnectionInfoAsync("master", queryTempFile.FilePath);
-
+                var sqlConnection = ConnectionService.OpenSqlConnection(connectionResult.ConnectionInfo);
+                SqlStoreConnection connection = new SqlStoreConnection(sqlConnection);
+                var xeStore = new XEStore(connection);
                 ProfilerService profilerService = new ProfilerService();
-
-                // start a new session
-                var startParams = new StartProfilingParams();
-                startParams.OwnerUri = connectionResult.ConnectionInfo.OwnerUri;
-                startParams.SessionName = "Standard";
-
-                string sessionId = null;
-                var startContext = new Mock<RequestContext<StartProfilingResult>>();
-                startContext.Setup(rc => rc.SendResult(It.IsAny<StartProfilingResult>()))
-                    .Returns<StartProfilingResult>((result) =>
-                    {
-                        // capture the session id for sending the stop message
-                        return Task.FromResult(0);
-                    });
-
-                await profilerService.HandleStartProfilingRequest(startParams, startContext.Object);
-
-                startContext.VerifyAll();
-
-                // wait a bit for the session monitoring to initialize
-                Thread.Sleep(TimeSpan.FromHours(1));
-
-                // stop the session
-                var stopParams = new StopProfilingParams()
+                var sessionName = await StartStandardSession(profilerService, connectionResult.ConnectionInfo.OwnerUri);
+                xeStore.Sessions.Refresh();
+                Assert.Multiple(() =>
                 {
-                    OwnerUri = sessionId
-                };
+                    Assert.That(xeStore.Sessions.Cast<Session>().Select(s => s.Name), Has.Member(sessionName), "ProfilerService should have created the session");
+                    Assert.That(xeStore.Sessions[sessionName].IsRunning, Is.True, "Session should be running when created by ProfilerService");
+                });
 
-                var stopContext = new Mock<RequestContext<StopProfilingResult>>();
-                stopContext.Setup(rc => rc.SendResult(It.IsAny<StopProfilingResult>()))
-                    .Returns(Task.FromResult(0));
 
-                await profilerService.HandleStopProfilingRequest(stopParams, stopContext.Object);
+                try
+                {
+                    var xeSession = xeStore.Sessions[sessionName];
+                    xeSession.Stop();
+                    // start a new session
+                    var startParams = new StartProfilingParams
+                    {
+                        OwnerUri = connectionResult.ConnectionInfo.OwnerUri,
+                        SessionName = sessionName
+                    };
 
-                stopContext.VerifyAll();
+                    string sessionId = null;
+                    var startContext = new Mock<RequestContext<StartProfilingResult>>();
+                    startContext.Setup(rc => rc.SendResult(It.IsAny<StartProfilingResult>()))
+                        .Returns<StartProfilingResult>((result) =>
+                        {
+                            // capture the session id for sending the stop message
+                            sessionId = result.UniqueSessionId;
+                            return Task.FromResult(0);
+                        });
+
+                    
+                    await profilerService.HandleStartProfilingRequest(startParams, startContext.Object);
+                    Assert.That(sessionId, Does.Contain(connectionResult.ConnectionInfo.ConnectionDetails.ServerName), "UniqueSessionId");
+                    startContext.VerifyAll();
+
+                    // wait a bit for the session monitoring to initialize
+                    Thread.Sleep(TimeSpan.FromSeconds(30));
+
+                    xeSession.Refresh();
+                    Assert.That(xeSession.IsRunning, Is.True, "Session should be running due to HandleStartProfilingRequest");
+
+                    // stop the session
+                    var stopParams = new StopProfilingParams()
+                    {
+                        OwnerUri = connectionResult.ConnectionInfo.OwnerUri
+                    };
+
+                    var stopContext = new Mock<RequestContext<StopProfilingResult>>();
+                    stopContext.Setup(rc => rc.SendResult(It.IsAny<StopProfilingResult>()))
+                        .Returns(Task.FromResult(0));
+
+                    await profilerService.HandleStopProfilingRequest(stopParams, stopContext.Object);
+
+                    xeSession.Refresh();
+                    Assert.That(xeSession.IsRunning, Is.False, "Session should be stopped due to HandleStopProfilingRequest");
+                    stopContext.VerifyAll();
+                }
+                finally
+                {
+                    try
+                    {
+                        xeStore.Sessions.Refresh();
+                        if (xeStore.Sessions.Contains(sessionName))
+                        {
+                            try
+                            {
+                                xeStore.Sessions[sessionName].Stop();
+                            }
+                            catch 
+                            { }
+                            xeStore.Sessions[sessionName].Drop();
+                        }
+                    }
+                    catch
+                    { }
+                }
             }
         }
 
-        /// <summary>
-        /// Verify  the profiler service XEvent session factory
-        /// </summary>
-        //[Test]
-        public void TestCreateXEventSession()
+        private async Task<string> StartStandardSession(ProfilerService profilerService, string ownerUri)
         {
-            var liveConnection = LiveConnectionHelper.InitLiveConnectionInfo("master");
-            ProfilerService profilerService = new ProfilerService();
-            IXEventSession xeSession = profilerService.GetXEventSession("Profiler", liveConnection.ConnectionInfo);
-            Assert.NotNull(xeSession);
-            Assert.NotNull(xeSession.GetTargetXml());
+            const string sessionName = "ADS_Standard_Test";
+            var template = Newtonsoft.Json.JsonConvert.DeserializeObject<ProfilerSessionTemplate>(standardSessionJson.Substring(1, standardSessionJson.Length - 2));
+
+            var createParams = new CreateXEventSessionParams() { OwnerUri = ownerUri, SessionName = sessionName, Template = template };
+            var requestContext = new Mock<RequestContext<CreateXEventSessionResult>>();
+            requestContext.Setup(c => c.SendResult(It.IsAny<CreateXEventSessionResult>()))
+              .Returns<CreateXEventSessionResult>((result) => { return Task.FromResult(0); });
+            var serviceHostMock = new Mock<IProtocolEndpoint>();
+            profilerService.ServiceHost = serviceHostMock.Object;
+            await profilerService.HandleCreateXEventSessionRequest(createParams, requestContext.Object);
+            return sessionName;
         }
+
+        const string standardSessionJson = /*lang=json,strict*/ "[{\"name\": \"Standard_OnPrem\", \"defaultView\": \"Standard View\", \"engineTypes\": [\"Standalone\"], \"createStatement\": \"CREATE EVENT SESSION [{sessionName}] ON SERVER ADD EVENT sqlserver.attention(ACTION(package0.event_sequence,sqlserver.client_app_name,sqlserver.client_pid,sqlserver.database_id,sqlserver.nt_username,sqlserver.query_hash,sqlserver.server_principal_name,sqlserver.session_id) WHERE ([package0].[equal_boolean]([sqlserver].[is_system],(0)))), ADD EVENT sqlserver.existing_connection(SET collect_options_text=(1) ACTION(package0.event_sequence,sqlserver.client_app_name,sqlserver.client_pid,sqlserver.nt_username,sqlserver.server_principal_name,sqlserver.session_id)), ADD EVENT sqlserver.login(SET collect_options_text=(1) ACTION(package0.event_sequence,sqlserver.client_app_name,sqlserver.client_pid,sqlserver.nt_username,sqlserver.server_principal_name,sqlserver.session_id)), ADD EVENT sqlserver.logout( ACTION(package0.event_sequence,sqlserver.client_app_name,sqlserver.client_pid,sqlserver.nt_username,sqlserver.server_principal_name,sqlserver.session_id)), ADD EVENT sqlserver.rpc_completed( ACTION(package0.event_sequence,sqlserver.client_app_name,sqlserver.client_pid,sqlserver.database_id,sqlserver.database_name,sqlserver.nt_username,sqlserver.query_hash,sqlserver.server_principal_name,sqlserver.session_id) WHERE ([package0].[equal_boolean]([sqlserver].[is_system],(0)))), ADD EVENT sqlserver.sql_batch_completed( ACTION(package0.event_sequence,sqlserver.client_app_name,sqlserver.client_pid,sqlserver.database_id,sqlserver.database_name,sqlserver.nt_username,sqlserver.query_hash,sqlserver.server_principal_name,sqlserver.session_id) WHERE ([package0].[equal_boolean]([sqlserver].[is_system],(0)))), ADD EVENT sqlserver.sql_batch_starting( ACTION(package0.event_sequence,sqlserver.client_app_name,sqlserver.client_pid,sqlserver.database_id,sqlserver.database_name,sqlserver.nt_username,sqlserver.query_hash,sqlserver.server_principal_name,sqlserver.session_id) WHERE ([package0].[equal_boolean]([sqlserver].[is_system],(0)))) ADD TARGET package0.ring_buffer(SET max_events_limit=(1000),max_memory=(51200)) WITH (MAX_MEMORY=8192 KB,EVENT_RETENTION_MODE=ALLOW_SINGLE_EVENT_LOSS,MAX_DISPATCH_LATENCY=5 SECONDS,MAX_EVENT_SIZE=0 KB,MEMORY_PARTITION_MODE=PER_CPU,TRACK_CAUSALITY=ON,STARTUP_STATE=OFF)\"}]";
     }
 }

test/Microsoft.SqlTools.ServiceLayer.UnitTests/Microsoft.SqlTools.ServiceLayer.UnitTests.csproj

@@ -45,4 +45,9 @@
 	<ItemGroup>
 		<EmbeddedResource Include="ShowPlan\TestShowPlanRecommendations.xml" />
 	</ItemGroup>
+	<ItemGroup>
+	  <None Update="Profiler\TestXel_0.xel">
+	    <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+	  </None>
+	</ItemGroup>
 </Project>

test/Microsoft.SqlTools.ServiceLayer.UnitTests/Profiler/ProfilerServiceTests.cs

@@ -16,10 +16,12 @@ using Microsoft.SqlTools.ServiceLayer.Profiler.Contracts;
 using Microsoft.SqlTools.ServiceLayer.UnitTests.Utility;
 using Moq;
 using NUnit.Framework;
+using System.IO;
+using System.Reflection;
+using System.Linq;
 
 namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
 {
-    [TestFixture]
     /// <summary>
     /// Unit tests for ProfilerService
     /// </summary>
@@ -29,66 +31,39 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
         /// Test starting a profiling session and receiving event callback
         /// </summary>
         /// <returns></returns>
-        // TODO: Fix flaky test. See https://github.com/Microsoft/sqltoolsservice/issues/459
-        //[Test]
-        public async Task TestStartProfilingRequest()
+        [Test]
+        public async Task StartProfilingRequest_creates_pausable_remote_session()
         {
-            string sessionId = null;
-            bool recievedEvents = false;
+            var sessionId = new SessionId("testsession_1", 1);
             string testUri = "profiler_uri";
             var requestContext = new Mock<RequestContext<StartProfilingResult>>();
             requestContext.Setup(rc => rc.SendResult(It.IsAny<StartProfilingResult>()))
                 .Returns<StartProfilingResult>((result) =>
                 {
-                    // capture the session id for sending the stop message
+                    Assert.Multiple(() =>
+                    {
+                        Assert.That(result.CanPause, Is.True, "Result.CanPause for RingBuffer sessions");
+                        Assert.That(result.UniqueSessionId, Is.EqualTo(sessionId.ToString()), "Result.UniqueSessionId");
+                    });
                     return Task.FromResult(0);
                 });
 
             // capture Listener event notifications
-            var mockListener = new Mock<IProfilerSessionListener>();
-            mockListener.Setup(p => p.EventsAvailable(It.IsAny<string>(), It.IsAny<List<ProfilerEvent>>(), It.IsAny<bool>())).Callback(() =>
-                {
-                    recievedEvents = true;
-                });
-
             var profilerService = new ProfilerService();
-            profilerService.SessionMonitor.AddSessionListener(mockListener.Object);
             profilerService.ConnectionServiceInstance = TestObjects.GetTestConnectionService();
             ConnectionInfo connectionInfo = TestObjects.GetTestConnectionInfo();
             profilerService.ConnectionServiceInstance.OwnerToConnectionMap.Add(testUri, connectionInfo);
             profilerService.XEventSessionFactory = new TestXEventSessionFactory();
 
-            var requestParams = new StartProfilingParams();
-            requestParams.OwnerUri = testUri;
-            requestParams.SessionName = "Standard";
+            var requestParams = new StartProfilingParams
+            {
+                OwnerUri = testUri,
+                SessionName = "Standard"
+            };
 
             // start profiling session
             await profilerService.HandleStartProfilingRequest(requestParams, requestContext.Object);            
-
-            profilerService.SessionMonitor.PollSession(1);
-            // simulate a short polling delay
-            Thread.Sleep(200);
-            profilerService.SessionMonitor.PollSession(1);
-
-            // wait for polling to finish, or for timeout
-            System.Timers.Timer pollingTimer = new System.Timers.Timer();
-            pollingTimer.Interval = 10000;
-            pollingTimer.Start();
-            bool timeout = false;
-            pollingTimer.Elapsed += new System.Timers.ElapsedEventHandler((s_, e_) => {timeout = true;});
-            while (sessionId == null && !timeout)
-            {
-                Thread.Sleep(250);
-            }
-            pollingTimer.Stop();
-
             requestContext.VerifyAll();            
-
-            // Check that the correct XEvent session was started
-            Assert.AreEqual("1", sessionId);
-
-            // check that the proper owner Uri was used
-            Assert.True(recievedEvents);
         }
 
         /// <summary>
@@ -118,6 +93,9 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
                     stopped = true;
                 });
 
+            mockSession.Setup(p => p.GetTargetXml()).Returns("<RingBufferTarget/>");
+
+            mockSession.Setup(p => p.Id).Returns(new SessionId("test_1", 1));
             var sessionListener = new TestSessionListener();
             var profilerService = new ProfilerService();
             profilerService.SessionMonitor.AddSessionListener(sessionListener);
@@ -136,8 +114,8 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             requestContext.VerifyAll();
 
             // check that session was succesfully stopped and stop was called
-            Assert.True(success);
-            Assert.True(stopped);
+            Assert.True(success, nameof(success));
+            Assert.True(stopped, nameof(stopped));
 
             // should not be able to remove the session, it should already be gone
             ProfilerSession ps;
@@ -185,9 +163,9 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             profilerService.SessionMonitor.StartMonitoringSession(testUri, new TestXEventSession1());
 
             // poll the session
-            profilerService.SessionMonitor.PollSession(1);
+            profilerService.SessionMonitor.PollSession(new SessionId("testsession_1", 1));
             Thread.Sleep(500);
-            profilerService.SessionMonitor.PollSession(1);
+            profilerService.SessionMonitor.PollSession(new SessionId("testsession_1", 1));
 
             // wait for polling to finish, or for timeout
             System.Timers.Timer pollingTimer = new System.Timers.Timer();
@@ -211,7 +189,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             recievedEvents = false;
             success = false;
 
-            profilerService.SessionMonitor.PollSession(1);
+            profilerService.SessionMonitor.PollSession(new SessionId("testsession_1", 1));
 
             // confirm that no events were sent to paused Listener
             Assert.False(recievedEvents);
@@ -220,7 +198,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             await profilerService.HandlePauseProfilingRequest(requestParams, requestContext.Object);
             Assert.True(success);
 
-            profilerService.SessionMonitor.PollSession(1);
+            profilerService.SessionMonitor.PollSession(new SessionId("testsession_1", 1));
 
             // wait for polling to finish, or for timeout
             timeout = false;
@@ -240,7 +218,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
         /// Test notifications for stopped sessions
         /// </summary>
         [Test]
-        public async Task TestStoppedSessionNotification()
+        public void TestStoppedSessionNotification()
         {
             bool sessionStopped = false;
             string testUri = "profiler_uri";
@@ -251,9 +229,9 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
                 {
                     throw new XEventException();
                 });
-
+            mockSession.Setup(p => p.Id).Returns(new SessionId("test_1", 1));
             var mockListener = new Mock<IProfilerSessionListener>();
-            mockListener.Setup(p => p.SessionStopped(It.IsAny<string>(), It.IsAny<int>())).Callback(() =>
+            mockListener.Setup(p => p.SessionStopped(It.IsAny<string>(), It.IsAny<SessionId>(), It.IsAny<string>())).Callback(() =>
             {
                 sessionStopped = true;
             });
@@ -282,5 +260,114 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             // check that a stopped session notification was sent
             Assert.True(sessionStopped);
         }
+
+        [Test]
+        public void StartProfilingRequest_defaults_to_remote()
+        { 
+            var param = new StartProfilingParams();
+            Assert.That(param.SessionType, Is.EqualTo(ProfilingSessionType.RemoteSession), nameof(param.SessionType));
+        }
+
+        [Test]
+        public async Task StartProfilingRequest_creates_a_LocalFile_session_on_request()
+        {
+            var filePath = @"c:\folder\file.xel";
+            var param = new StartProfilingParams() { OwnerUri = "someUri", SessionType = ProfilingSessionType.LocalFile, SessionName =  filePath};
+            var mockSession = new Mock<IObservableXEventSession>();
+            mockSession.Setup(p => p.GetTargetXml()).Callback(() =>
+            {
+                throw new XEventException();
+            });
+            mockSession.Setup(p => p.Id).Returns(new SessionId("test_1", 1));
+            var requestContext = new Mock<RequestContext<StartProfilingResult>>();
+            requestContext.Setup(rc => rc.SendResult(It.IsAny<StartProfilingResult>()))
+                .Returns<StartProfilingResult>((result) =>
+                {
+                    return Task.FromResult(0);
+                });
+            var sessionFactory = new Mock<IXEventSessionFactory>();
+            sessionFactory.Setup(s => s.OpenLocalFileSession(filePath))
+                .Returns (mockSession.Object)
+                .Verifiable();
+            var profilerService = new ProfilerService() { XEventSessionFactory = sessionFactory.Object };
+            await profilerService.HandleStartProfilingRequest(param, requestContext.Object);
+            sessionFactory.Verify();
+            requestContext.VerifyAll();
+        }
+
+        [Test]
+        public async Task ProfilerService_processes_localfile_session()
+        {
+            var viewerId = "someUri";
+            var filePath = Path.Combine(Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location), "Profiler", "TestXel_0.xel");
+            var param = new StartProfilingParams() { OwnerUri = viewerId, SessionType = ProfilingSessionType.LocalFile, SessionName = filePath };
+            var requestContext = new Mock<RequestContext<StartProfilingResult>>();
+            requestContext.Setup(rc => rc.SendResult(It.IsAny<StartProfilingResult>()))
+                .Returns<StartProfilingResult>((result) =>
+                {
+                    Assert.Multiple(() =>
+                    {
+                        Assert.That(result.CanPause, Is.False, "local file session cannot be paused");
+                        Assert.That(result.UniqueSessionId, Is.EqualTo(filePath), "UniqueSessionId should match file path");
+                    });
+                    return Task.FromResult(0);
+                });
+            var profilerService = new ProfilerService();
+            var listener = new TestSessionListener();
+            profilerService.SessionMonitor.AddSessionListener(listener);
+            await profilerService.HandleStartProfilingRequest(param, requestContext.Object);
+            var retries = 100;
+            while (retries-- > 0 && !listener.StoppedSessions.Contains(viewerId))
+            {
+                Thread.Sleep(100);
+            }
+            Assert.Multiple(() =>
+            {
+                Assert.That(listener.StoppedSessions, Has.Member(viewerId), "session should have been stopped after reading the file");
+                Assert.That(listener.AllEvents.Keys, Has.Member(viewerId), "session should have events logged for it");
+                Assert.That(listener.AllEvents[viewerId]?.Count, Is.EqualTo(149), "all events from the xel should be in the buffer");
+            });
+        }
+
+        [Test]
+        public async Task ProfilerService_includes_ErrorMessage_in_session_stop_notification()
+        {
+            var param = new StartProfilingParams() { OwnerUri = "someUri", SessionName = "someSession" };
+            var mockSession = new Mock<IXEventSession>();
+            mockSession.Setup(p => p.GetTargetXml()).Callback(() =>
+            {
+                throw new XEventException("test!");
+            });
+            mockSession.Setup(p => p.Id).Returns(new SessionId("test_1", 1));
+            var requestContext = new Mock<RequestContext<StartProfilingResult>>();
+            requestContext.Setup(rc => rc.SendResult(It.IsAny<StartProfilingResult>()))
+                .Returns<StartProfilingResult>((result) =>
+                {
+                    return Task.FromResult(0);
+                });
+            var sessionFactory = new Mock<IXEventSessionFactory>();
+            sessionFactory.Setup(s => s.GetXEventSession(It.IsAny<string>(), It.IsAny<ConnectionInfo>()))
+                .Returns(mockSession.Object)
+                .Verifiable();
+            var profilerService = new ProfilerService() { XEventSessionFactory = sessionFactory.Object };
+            profilerService.ConnectionServiceInstance = TestObjects.GetTestConnectionService();
+            var connectionInfo = TestObjects.GetTestConnectionInfo();
+            profilerService.ConnectionServiceInstance.OwnerToConnectionMap.Add("someUri", connectionInfo);
+
+            var listener = new TestSessionListener();
+            profilerService.SessionMonitor.AddSessionListener(listener);
+            await profilerService.HandleStartProfilingRequest(param, requestContext.Object);
+            var retries = 10;
+            while (retries-- > 0 && !listener.StoppedSessions.Any())
+            {
+                Thread.Sleep(100);
+            }
+            Assert.Multiple(() =>
+            {
+                Assert.That(listener.ErrorMessages, Is.EqualTo(new[] { "test!" }), "listener.ErrorMessages");
+                Assert.That(listener.StoppedSessions, Has.Member("someUri"), "listener.StoppedSessions");
+            });
+            sessionFactory.Verify();
+        }
     }
 }

test/Microsoft.SqlTools.ServiceLayer.UnitTests/Profiler/ProfilerSessionTests.cs

@@ -25,7 +25,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
         public void TestFilterOldEvents()
         {
             // create a profiler session and get some test events
-            var profilerSession = new ProfilerSession();
+            var profilerSession = new ProfilerSession(new XEventSession());
             var allEvents = ProfilerTestObjects.TestProfilerEvents;
             var profilerEvents = ProfilerTestObjects.TestProfilerEvents;
 
@@ -62,7 +62,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
         public void TestFilterProfilerEvents()
         {
             // create a profiler session and get some test events
-            var profilerSession = new ProfilerSession();
+            var profilerSession = new ProfilerSession(new XEventSession());
             var profilerEvents = ProfilerTestObjects.TestProfilerEvents;
                         
             int expectedEventCount = profilerEvents.Count;
@@ -86,7 +86,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
         public void TestEventsLost()
         {
             // create a profiler session and get some test events
-            var profilerSession = new ProfilerSession();
+            var profilerSession = new ProfilerSession(new XEventSession());
             var profilerEvents = ProfilerTestObjects.TestProfilerEvents;
 
             // filter all the results from the first poll
@@ -135,7 +135,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             DateTime startTime = DateTime.Now;
 
             // create new profiler session
-            var profilerSession = new ProfilerSession();
+            var profilerSession = new ProfilerSession(new XEventSession());
             
             // enter the polling block
             Assert.True(profilerSession.TryEnterPolling());

test/Microsoft.SqlTools.ServiceLayer.UnitTests/Profiler/ProfilerTestObjects.cs

@@ -5,6 +5,7 @@
 
 #nullable disable
 
+using System;
 using System.Collections.Generic;
 using Microsoft.SqlTools.ServiceLayer.Connection;
 using Microsoft.SqlTools.ServiceLayer.Profiler;
@@ -37,21 +38,24 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
 
     public class TestSessionListener : IProfilerSessionListener
     {
-        public string PreviousSessionId { get; set; }
+        public readonly Dictionary<string, List<ProfilerEvent>> AllEvents = new Dictionary<string, List<ProfilerEvent>>();
 
-        public List<ProfilerEvent> PreviousEvents { get; set; }
-
-        public bool Stopped { get; set; }
+        public readonly List<string> StoppedSessions = new List<string>();
+        public readonly List<string> ErrorMessages = new List<string>();
 
         public void EventsAvailable(string sessionId, List<ProfilerEvent> events, bool eventsLost)
         {
-            this.PreviousSessionId = sessionId;
-            this.PreviousEvents = events;
+            if (!AllEvents.ContainsKey(sessionId))
+            {
+                AllEvents[sessionId] = new List<ProfilerEvent>();
+            }
+            AllEvents[sessionId].AddRange(events);            
         }
 
-        public void SessionStopped(string viewerId, int sessionId)
+        public void SessionStopped(string viewerId, SessionId sessionId, string errorMessage)
         {
-            Stopped = true;
+            StoppedSessions.Add(viewerId);
+            ErrorMessages.Add(errorMessage);
         }
     }
 
@@ -197,7 +201,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
 
 
 
-        public int Id { get { return 51; } }
+        public SessionId Id { get { return new SessionId("testsession_51"); } }
 
         public void Start(){}
 
@@ -282,7 +286,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             "	</event>" +
             "</RingBufferTarget>";
 
-        public int Id { get { return 1; } }
+        public SessionId Id { get { return new SessionId("testsession_1"); } }
 
         public void Start(){}
 
@@ -373,7 +377,7 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
             "	</event>" +
             "</RingBufferTarget>";
 
-        public int Id { get { return 2; } }
+        public SessionId Id { get { return new SessionId("testsession_2"); } }
 
         public void Start(){}
 
@@ -420,5 +424,10 @@ namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
                 return new TestXEventSession2();
             }
         }
+
+        public IXEventSession OpenLocalFileSession(string filePath)
+        {
+            throw new NotImplementedException();
+        }
     }
 }

test/Microsoft.SqlTools.ServiceLayer.UnitTests/Profiler/XeStreamObservableTests.cs

@@ -0,0 +1,113 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+#nullable disable
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using NUnit.Framework;
+using System.Reflection;
+using Microsoft.SqlTools.ServiceLayer.Profiler.Contracts;
+using Microsoft.SqlTools.ServiceLayer.Profiler;
+using System.Threading;
+using System.Linq;
+using Moq;
+
+namespace Microsoft.SqlTools.ServiceLayer.UnitTests.Profiler
+{
+    public class XeStreamObservableTests
+    { 
+
+        /// <summary>
+        /// this might technically be an integration test but putting it here because it doesn't require any connectivity.
+        /// </summary>
+        [Test]
+        public void XeStreamObservable_reads_entire_xel_file()
+        {
+            var observer = InitializeFileObserver();
+            observer.Observable.Start();
+            var retries = 100;
+            while (!observer.Completed && retries-- > 0)
+            {
+                Thread.Sleep(100);
+            }
+            Assert.That(observer.Completed, Is.True, $"Reading the file didn't complete in 10 seconds. Events read: {observer.ProfilerEvents.Count}");
+            Assert.Multiple(() =>
+            {
+                Assert.That(observer.ProfilerEvents.Count, Is.EqualTo(149), "Number of events read");
+                Assert.That(observer.ProfilerEvents[0].Name, Is.EqualTo("rpc_completed"), "First event in the file");
+                Assert.That(observer.ProfilerEvents.Last().Name, Is.EqualTo("sql_batch_completed"), "Last event in the file");
+
+            });
+        }
+
+        [Test]
+        public void XeStreamObservable_calls_OnError_when_the_fetcher_fails()
+        {
+            var observer = InitializeFileObserver("thispathdoesnotexist.xel");
+            observer.Observable.Start();
+            var retries = 10;
+            while (!observer.Completed && retries-- > 0)
+            {
+                Thread.Sleep(100);
+            }
+            Assert.Multiple(() =>
+            {
+                Assert.That(observer.Completed, Is.True, $"Reading the missing file didn't complete in 1 second.");
+                Assert.That(observer.Error?.GetBaseException(), Is.InstanceOf<FileNotFoundException>(), $"Expected Error from missing file. Error:{observer.Error}");
+            });
+        }
+
+        private XeStreamObserver InitializeFileObserver(string filePath = null)
+        {
+            filePath ??= Path.Combine(Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location), "Profiler", "TestXel_0.xel");
+            var profilerService = SetupProfilerService(filePath);
+            var xeStreamObservable = new XeStreamObservable(() =>
+            {
+                return profilerService.initIXEventFetcher(filePath);
+            });
+            var observer = new XeStreamObserver() { Observable = xeStreamObservable };
+            xeStreamObservable.Subscribe(observer);
+            return observer;
+        }
+
+        private ProfilerService SetupProfilerService (string filePath = null)
+        {
+            filePath ??= Path.Combine(Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location), "Profiler", "TestXel_0.xel");
+            var sessionFactory = new Mock<IXEventSessionFactory>();
+            var profilerService = new ProfilerService() { XEventSessionFactory = sessionFactory.Object };
+            return profilerService;
+        }
+    }
+
+    sealed class XeStreamObserver : IObserver<ProfilerEvent>
+    {
+        public XeStreamObservable Observable { get; set; }
+        public readonly List<ProfilerEvent> ProfilerEvents = new List<ProfilerEvent>();
+
+        public bool Completed { get; private set; }
+
+        public Exception Error { get; private set; }
+
+        public void OnCompleted()
+        {
+            Completed = true;
+        }
+
+        public void OnError(Exception error)
+        {
+            Error = error;
+        }
+
+        public void OnNext(ProfilerEvent value)
+        {
+            ProfilerEvents.Add(value);
+            OnEventAdded?.Invoke(this, EventArgs.Empty);
+        }
+
+        public event EventHandler OnEventAdded;
+    }
+}