Streaming Library Refactor (#5908)

microsoft · Oct 15, 2021 · da8348a · da8348a
1 parent 76ccffd
commit da8348a
Show file tree

Hide file tree

Showing 59 changed files with 6,178 additions and 69 deletions.
diff --git a/CodeCoverage.runsettings b/CodeCoverage.runsettings
@@ -40,6 +40,7 @@
                 <!--<ModulePath>.*\Microsoft.Bot.Configuration.dll$</ModulePath>-->
                 <ModulePath>.*\Microsoft.Bot.Connector.dll$</ModulePath>
                 <ModulePath>.*\Microsoft.Bot.Schema.dll$</ModulePath>
+                <ModulePath>.*\Microsoft.Bot.Connector.Streaming.dll$</ModulePath>
                 <ModulePath>.*\Microsoft.Bot.Streaming.dll$</ModulePath>
               </Include>
               <Exclude>

diff --git a/Microsoft.Bot.Builder.sln b/Microsoft.Bot.Builder.sln
@@ -225,6 +225,18 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.Bot.Builder.Dialo
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.Bot.Builder.Dialogs.Adaptive.Runtime", "libraries\Microsoft.Bot.Builder.Dialogs.Adaptive.Runtime\Microsoft.Bot.Builder.Dialogs.Adaptive.Runtime.csproj", "{2DB4E5B0-3209-425E-A912-005A330CC66A}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.Bot.Connector.Streaming", "libraries\Microsoft.Bot.Connector.Streaming\Microsoft.Bot.Connector.Streaming.csproj", "{80FA0E50-8F81-4C60-B265-1039391C1CEE}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.Bot.Connector.Streaming.Tests", "tests\Microsoft.Bot.Connector.Streaming.Tests\Microsoft.Bot.Connector.Streaming.Tests.csproj", "{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Streaming", "Streaming", "{EBFEF03F-9ACE-4312-89D7-2C8A147CDF9C}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.Bot.Connector.Streaming.Tests.Server", "tests\Microsoft.Bot.Connector.Streaming.Tests.Server\Microsoft.Bot.Connector.Streaming.Tests.Server.csproj", "{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.Bot.Connector.Streaming.Tests.Client", "tests\Microsoft.Bot.Connector.Streaming.Tests.Client\Microsoft.Bot.Connector.Streaming.Tests.Client.csproj", "{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.Bot.Connector.Streaming.Perf", "tests\Microsoft.Bot.Connector.Streaming.Perf\Microsoft.Bot.Connector.Streaming.Perf.csproj", "{B49A3201-5BEE-426C-A082-D92D52172E06}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -881,6 +893,46 @@ Global
 		{2DB4E5B0-3209-425E-A912-005A330CC66A}.Release|Any CPU.Build.0 = Release|Any CPU
 		{2DB4E5B0-3209-425E-A912-005A330CC66A}.Release-Windows|Any CPU.ActiveCfg = Release|Any CPU
 		{2DB4E5B0-3209-425E-A912-005A330CC66A}.Release-Windows|Any CPU.Build.0 = Release|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Debug-Windows|Any CPU.ActiveCfg = Debug|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Debug-Windows|Any CPU.Build.0 = Debug|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Release|Any CPU.Build.0 = Release|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Release-Windows|Any CPU.ActiveCfg = Release|Any CPU
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE}.Release-Windows|Any CPU.Build.0 = Release|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Debug-Windows|Any CPU.ActiveCfg = Debug|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Debug-Windows|Any CPU.Build.0 = Debug|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Release|Any CPU.Build.0 = Release|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Release-Windows|Any CPU.ActiveCfg = Release|Any CPU
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7}.Release-Windows|Any CPU.Build.0 = Release|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Debug-Windows|Any CPU.ActiveCfg = Debug|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Debug-Windows|Any CPU.Build.0 = Debug|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Release|Any CPU.Build.0 = Release|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Release-Windows|Any CPU.ActiveCfg = Release|Any CPU
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71}.Release-Windows|Any CPU.Build.0 = Release|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Debug-Windows|Any CPU.ActiveCfg = Debug|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Debug-Windows|Any CPU.Build.0 = Debug|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Release|Any CPU.Build.0 = Release|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Release-Windows|Any CPU.ActiveCfg = Release|Any CPU
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161}.Release-Windows|Any CPU.Build.0 = Release|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Debug-Windows|Any CPU.ActiveCfg = Debug|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Debug-Windows|Any CPU.Build.0 = Debug|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Release-Windows|Any CPU.ActiveCfg = Release|Any CPU
+		{B49A3201-5BEE-426C-A082-D92D52172E06}.Release-Windows|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -979,6 +1031,12 @@ Global
 		{0BF5E92D-D034-4D80-8921-07627F55F412} = {C40A300C-8988-4733-A760-A776C6309B57}
 		{D611AC03-9859-4EB6-BAB9-C26F493DFDB3} = {AD743B78-D61F-4FBF-B620-FA83CE599A50}
 		{2DB4E5B0-3209-425E-A912-005A330CC66A} = {4269F3C3-6B42-419B-B64A-3E6DC0F1574A}
+		{80FA0E50-8F81-4C60-B265-1039391C1CEE} = {4269F3C3-6B42-419B-B64A-3E6DC0F1574A}
+		{9EBA6EDB-7D67-4BC5-9F94-E0162A538CC7} = {EBFEF03F-9ACE-4312-89D7-2C8A147CDF9C}
+		{EBFEF03F-9ACE-4312-89D7-2C8A147CDF9C} = {AD743B78-D61F-4FBF-B620-FA83CE599A50}
+		{FB7ADCDF-C0A5-49EA-8ADC-CC77B6FB9D71} = {EBFEF03F-9ACE-4312-89D7-2C8A147CDF9C}
+		{2E5AD07C-4F6E-4B6B-BEFE-9FBE9F789161} = {EBFEF03F-9ACE-4312-89D7-2C8A147CDF9C}
+		{B49A3201-5BEE-426C-A082-D92D52172E06} = {EBFEF03F-9ACE-4312-89D7-2C8A147CDF9C}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {7173C9F3-A7F9-496E-9078-9156E35D6E16}

diff --git a/libraries/Microsoft.Bot.Builder.Dialogs.Adaptive.Runtime/CoreBotAdapter.cs b/libraries/Microsoft.Bot.Builder.Dialogs.Adaptive.Runtime/CoreBotAdapter.cs
@@ -13,7 +13,7 @@ internal class CoreBotAdapter : CloudAdapter
         public CoreBotAdapter(
             BotFrameworkAuthentication botFrameworkAuthentication,
             IEnumerable<IMiddleware> middlewares,
-            ILogger logger = null)
+            ILogger<CoreBotAdapter> logger = null)
             : base(botFrameworkAuthentication, logger)
         {
             // Pick up feature based middlewares such as telemetry or transcripts

diff --git a/libraries/Microsoft.Bot.Builder/CloudAdapterBase.cs b/libraries/Microsoft.Bot.Builder/CloudAdapterBase.cs
@@ -233,6 +233,23 @@ public override async Task CreateConversationAsync(string botAppId, string chann
             }
         }
 
+        /// <summary>
+        /// Gets the correct streaming connector factory that is processing the given activity.
+        /// </summary>
+        /// <param name="activity">The activity that is being processed.</param>
+        /// <returns>The Streaming Connector Factory responsible for processing the activity.</returns>
+        /// <remarks>
+        /// For HTTP requests, we usually create a new connector factory and reply to the activity over a new HTTP request.
+        /// However, when processing activities over a streaming connection, we need to reply over the same connection that is talking to a web socket.
+        /// This method will look up all active streaming connections in cloud adapter and return the connector factory that is processing the activity.
+        /// Messages between bot and channel go through the StreamingConnection (bot -> channel) and RequestHandler (channel -> bot), both created by the adapter.
+        /// However, proactive messages don't know which connection to talk to, so this method is designed to aid in the connection resolution for such proactive messages.
+        /// </remarks>
+        protected virtual ConnectorFactory GetStreamingConnectorFactory(Activity activity)
+        {
+            throw new NotImplementedException();
+        }
+
         /// <summary>
         /// The implementation for continue conversation.
         /// </summary>
@@ -247,7 +264,9 @@ protected async Task ProcessProactiveAsync(ClaimsIdentity claimsIdentity, Activi
             Logger.LogInformation($"ProcessProactiveAsync for Conversation Id: {continuationActivity.Conversation.Id}");
 
             // Create the connector factory.
-            var connectorFactory = BotFrameworkAuthentication.CreateConnectorFactory(claimsIdentity);
+            var connectorFactory = continuationActivity.IsFromStreamingConnection()
+                ? GetStreamingConnectorFactory(continuationActivity)
+                : BotFrameworkAuthentication.CreateConnectorFactory(claimsIdentity);
 
             // Create the connector client to use for outbound requests.
             using (var connectorClient = await connectorFactory.CreateAsync(continuationActivity.ServiceUrl, audience, cancellationToken).ConfigureAwait(false))

diff --git a/libraries/Microsoft.Bot.Builder/Microsoft.Bot.Builder.csproj b/libraries/Microsoft.Bot.Builder/Microsoft.Bot.Builder.csproj
@@ -24,6 +24,8 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.Bot.Streaming" Condition=" '$(ReleasePackageVersion)' == '' " Version="$(LocalPackageVersion)" />
     <PackageReference Include="Microsoft.Bot.Streaming" Condition=" '$(ReleasePackageVersion)' != '' " Version="$(ReleasePackageVersion)" />
+    <PackageReference Include="Microsoft.Bot.Connector.Streaming" Condition=" '$(ReleasePackageVersion)' == '' " Version="$(LocalPackageVersion)" />
+    <PackageReference Include="Microsoft.Bot.Connector.Streaming" Condition=" '$(ReleasePackageVersion)' != '' " Version="$(ReleasePackageVersion)" />
     <PackageReference Include="Microsoft.Bot.Connector" Condition=" '$(ReleasePackageVersion)' == '' " Version="$(LocalPackageVersion)" />
     <PackageReference Include="Microsoft.Bot.Connector" Condition=" '$(ReleasePackageVersion)' != '' " Version="$(ReleasePackageVersion)" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="2.1.0" />
@@ -33,6 +35,7 @@
   <ItemGroup>
     <ProjectReference Include="..\Microsoft.Bot.Connector\Microsoft.Bot.Connector.csproj" />
     <ProjectReference Include="..\Microsoft.Bot.Streaming\Microsoft.Bot.Streaming.csproj" />
+    <ProjectReference Include="..\Microsoft.Bot.Connector.Streaming\Microsoft.Bot.Connector.Streaming.csproj" />
   </ItemGroup>
 
   <ItemGroup>

diff --git a/libraries/Microsoft.Bot.Builder/Streaming/BotFrameworkHttpAdapterBase.cs b/libraries/Microsoft.Bot.Builder/Streaming/BotFrameworkHttpAdapterBase.cs
@@ -22,8 +22,10 @@ namespace Microsoft.Bot.Builder.Streaming
     /// <summary>
     /// An HTTP adapter base class.
     /// </summary>
-    public class BotFrameworkHttpAdapterBase : BotFrameworkAdapter, IStreamingActivityProcessor
+    public class BotFrameworkHttpAdapterBase : BotFrameworkAdapter, IStreamingActivityProcessor, IDisposable
     {
+        private bool _disposedValue;
+
         /// <summary>
         /// Initializes a new instance of the <see cref="BotFrameworkHttpAdapterBase"/> class.
         /// </summary>
@@ -220,7 +222,9 @@ public async Task<ResourceResponse> SendStreamingActivityAsync(Activity activity
                 var host = uri[uri.Length - 1];
                 await connection.ConnectAsync(new Uri(protocol + host + "/api/messages"), cancellationToken).ConfigureAwait(false);
 
+#pragma warning disable CA2000 // Dispose objects before losing scope (We'll dispose this when the adapter gets disposed or when elements are removed)
                 var handler = new StreamingRequestHandler(ConnectedBot, this, connection, Logger);
+#pragma warning restore CA2000 // Dispose objects before losing scope
 
                 if (RequestHandlers == null)
                 {
@@ -259,12 +263,49 @@ public async Task ConnectNamedPipeAsync(string pipeName, IBot bot, string audien
                 RequestHandlers = new List<StreamingRequestHandler>();
             }
 
+#pragma warning disable CA2000 // Dispose objects before losing scope (We'll dispose this when the adapter gets disposed or when elements are removed)
             var requestHandler = new StreamingRequestHandler(bot, this, pipeName, audience, Logger);
+#pragma warning restore CA2000 // Dispose objects before losing scope
             RequestHandlers.Add(requestHandler);
 
             await requestHandler.ListenAsync().ConfigureAwait(false);
         }
 
+        /// <inheritdoc/>
+        public void Dispose()
+        {
+            // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
+            Dispose(disposing: true);
+            GC.SuppressFinalize(this);
+        }
+
+        /// <summary>
+        /// Disposes resources of the <see cref="StreamingRequestHandler"/>.
+        /// </summary>
+        /// <param name="disposing">Whether we are disposing managed resources.</param>
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!_disposedValue)
+            {
+                if (disposing)
+                {
+                    if (RequestHandlers != null)
+                    {
+                        foreach (var handler in RequestHandlers)
+                        {
+                            if (handler is IDisposable disposable)
+                            {
+                                handler.Dispose();
+                            }
+                        }
+                    }
+                }
+
+                RequestHandlers = null;
+                _disposedValue = true;
+            }
+        }
+
         /// <summary>
         /// Evaluates if processing an outgoing activity is possible.
         /// </summary>