388024

Committed 11 Apr 2024 01:17PM UTC coverage: 78.163% (-0.02%) from 78.183%

Build # 388024

Build Type

push

CI-PR build

Committed by

web-flow

Commit Message

Update TimeoutException message (#6773)

Run Details

26183 of 33498 relevant lines covered (78.16%)

0.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.73

/libraries/Microsoft.Bot.Connector.Streaming/Application/StreamingConnection.cs

﻿// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

using System;
using System.Collections.Generic;
using System.IO.Pipelines;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Bot.Connector.Streaming.Session;
using Microsoft.Bot.Connector.Streaming.Transport;
using Microsoft.Bot.Streaming;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;

namespace Microsoft.Bot.Connector.Streaming.Application
{
    /// <summary>
    /// A streaming based connection that can listen for incoming requests and send them to a <see cref="RequestHandler"/>, 
    /// and can also send requests to the other end of the connection.
    /// </summary>
    public abstract class StreamingConnection : IDisposable
    {
        private readonly TaskCompletionSource<bool> _sessionInitializedTask = new TaskCompletionSource<bool>();

        private StreamingTransport _transport;
        private TransportHandler _application;
        private StreamingSession _session;

        private bool _disposedValue;

        /// <summary>
        /// Initializes a new instance of the <see cref="StreamingConnection"/> class.
        /// </summary>
        /// <param name="logger"><see cref="ILogger"/> for the connection.</param>
        protected StreamingConnection(ILogger logger)
        {
            Logger = logger ?? NullLogger.Instance;
        }

        /// <summary>
        /// Gets the <see cref="ILogger"/> instance for the streaming connection.
        /// </summary>
        /// <value>A <see cref="ILogger"/> for the streaming connection.</value>
        protected ILogger Logger { get; }

        /// <summary>
        /// Gets a value indicating whether this is currently connected.
        /// </summary>
        /// <value>
        /// True if this is currently connected, otherwise false.
        /// </value>
        protected bool IsConnected { get; private set; } = false;

        /// <summary>
        /// Sends a streaming request through the connection.
        /// </summary>
        /// <param name="request"><see cref="StreamingRequest"/> to be sent.</param>
        /// <param name="cancellationToken"><see cref="CancellationToken"/> to cancel the send process.</param>
        /// <returns>The <see cref="ReceiveResponse"/> returned from the client.</returns>
        public virtual async Task<ReceiveResponse> SendStreamingRequestAsync(StreamingRequest request, CancellationToken cancellationToken = default)
        {
            if (request == null)
            {
                throw new ArgumentNullException(nameof(request));
            }

            // This request could come fast while the session, transport and application are still being set up.
            // Wait for the session to signal that application and transport started before using the session.
            await _sessionInitializedTask.Task.ConfigureAwait(false);

            if (_session == null)
            {
                throw new InvalidOperationException("Cannot send streaming request since the session is not set up.");
            }

            try
            {
                return await _session.SendRequestAsync(request, cancellationToken).ConfigureAwait(false);
            }
            catch (TimeoutException ex)
            {
                var timeoutMessage = $"The connection to the client has been disconnected, and the request has timed out after waiting {TaskExtensions.DefaultTimeout.Seconds} seconds for a response.";
                if (IsConnected)
                {
                    timeoutMessage = $"The request sent to the client has timed out after waiting {TaskExtensions.DefaultTimeout.Seconds} seconds for a response.";
                }

                throw new OperationCanceledException(timeoutMessage, ex, cancellationToken);
            }
        }

        /// <summary>
        /// Opens the <see cref="StreamingConnection"/> and listens for incoming requests, which will
        /// be assembled and sent to the provided <see cref="RequestHandler"/>.
        /// </summary>
        /// <param name="requestHandler"><see cref="RequestHandler"/> to which incoming requests will be sent.</param>
        /// <param name="cancellationToken"><see cref="CancellationToken"/> that signals the need to stop the connection. 
        /// Once the token is cancelled, the connection will be gracefully shut down, finishing pending sends and receives.</param>
        /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
        public virtual async Task ListenAsync(RequestHandler requestHandler, CancellationToken cancellationToken = default)
        {
            _transport?.Dispose();
            _application?.Dispose();

            if (requestHandler == null)
            {
                throw new ArgumentNullException(nameof(requestHandler));
            }

            var duplexPipePair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);

            // Create transport and application
            _transport = CreateStreamingTransport(duplexPipePair.Application);
            _application = new TransportHandler(duplexPipePair.Transport, Logger);

            // Create session
            _session = new StreamingSession(requestHandler, _application, Logger, cancellationToken);

            // Start transport and application
            var transportTask = _transport.ConnectAsync((connected) => IsConnected = connected, cancellationToken);
            var applicationTask = _application.ListenAsync(cancellationToken);

            var tasks = new List<Task> { transportTask, applicationTask };

            // Signal that session is ready to be used
            _sessionInitializedTask.SetResult(true);

            // Let application and transport run
            await Task.WhenAll(tasks).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);
        }

        internal abstract StreamingTransport CreateStreamingTransport(IDuplexPipe application);

        /// <summary>
        /// Disposes managed and unmanaged resources of the underlying <see cref="StreamingConnection"/>.
        /// </summary>
        /// <param name="disposing">Whether we are disposing managed resources.</param>
        protected virtual void Dispose(bool disposing)
        {
            if (!_disposedValue)
            {
                if (disposing)
                {
                    _transport?.Dispose();
                    _application?.Dispose();
                }

                _disposedValue = true;
            }
        }
    }
}

1	// Copyright (c) Microsoft Corporation. All rights reserved.
2	// Licensed under the MIT License.
3
4	using System;
5	using System.Collections.Generic;
6	using System.IO.Pipelines;
7	using System.Threading;
8	using System.Threading.Tasks;
9	using Microsoft.Bot.Connector.Streaming.Session;
10	using Microsoft.Bot.Connector.Streaming.Transport;
11	using Microsoft.Bot.Streaming;
12	using Microsoft.Extensions.Logging;
13	using Microsoft.Extensions.Logging.Abstractions;
14
15	namespace Microsoft.Bot.Connector.Streaming.Application
16	{
17	/// <summary>
18	/// A streaming based connection that can listen for incoming requests and send them to a <see cref="RequestHandler"/>,
19	/// and can also send requests to the other end of the connection.
20	/// </summary>
21	public abstract class StreamingConnection : IDisposable
22	{
23	private readonly TaskCompletionSource<bool> _sessionInitializedTask = new TaskCompletionSource<bool>();	1✔
24
25	private StreamingTransport _transport;
26	private TransportHandler _application;
27	private StreamingSession _session;
28
29	private bool _disposedValue;
30
31	/// <summary>
32	/// Initializes a new instance of the <see cref="StreamingConnection"/> class.
33	/// </summary>
34	/// <param name="logger"><see cref="ILogger"/> for the connection.</param>
35	protected StreamingConnection(ILogger logger)	1✔
36	{
37	Logger = logger ?? NullLogger.Instance;	1✔
38	}	1✔
39
40	/// <summary>
41	/// Gets the <see cref="ILogger"/> instance for the streaming connection.
42	/// </summary>
43	/// <value>A <see cref="ILogger"/> for the streaming connection.</value>
44	protected ILogger Logger { get; }	1✔
45
46	/// <summary>
47	/// Gets a value indicating whether this is currently connected.
48	/// </summary>
49	/// <value>
50	/// True if this is currently connected, otherwise false.
51	/// </value>
52	protected bool IsConnected { get; private set; } = false;	1✔
53
54	/// <summary>
55	/// Sends a streaming request through the connection.
56	/// </summary>
57	/// <param name="request"><see cref="StreamingRequest"/> to be sent.</param>
58	/// <param name="cancellationToken"><see cref="CancellationToken"/> to cancel the send process.</param>
59	/// <returns>The <see cref="ReceiveResponse"/> returned from the client.</returns>
60	public virtual async Task<ReceiveResponse> SendStreamingRequestAsync(StreamingRequest request, CancellationToken cancellationToken = default)
61	{
62	if (request == null)	1✔
63	{
64	throw new ArgumentNullException(nameof(request));	×
65	}
66
67	// This request could come fast while the session, transport and application are still being set up.
68	// Wait for the session to signal that application and transport started before using the session.
69	await _sessionInitializedTask.Task.ConfigureAwait(false);	1✔
70
71	if (_session == null)	1✔
72	{
73	throw new InvalidOperationException("Cannot send streaming request since the session is not set up.");	×
74	}
75
76	try
77	{
78	return await _session.SendRequestAsync(request, cancellationToken).ConfigureAwait(false);	1✔
79	}
80	catch (TimeoutException ex)	×
81	{
82	var timeoutMessage = $"The connection to the client has been disconnected, and the request has timed out after waiting {TaskExtensions.DefaultTimeout.Seconds} seconds for a response.";	×
83	if (IsConnected)	×
84	{
85	timeoutMessage = $"The request sent to the client has timed out after waiting {TaskExtensions.DefaultTimeout.Seconds} seconds for a response.";	×
86	}
87
88	throw new OperationCanceledException(timeoutMessage, ex, cancellationToken);	×
89	}
90	}	1✔
91
92	/// <summary>
93	/// Opens the <see cref="StreamingConnection"/> and listens for incoming requests, which will
94	/// be assembled and sent to the provided <see cref="RequestHandler"/>.
95	/// </summary>
96	/// <param name="requestHandler"><see cref="RequestHandler"/> to which incoming requests will be sent.</param>
97	/// <param name="cancellationToken"><see cref="CancellationToken"/> that signals the need to stop the connection.
98	/// Once the token is cancelled, the connection will be gracefully shut down, finishing pending sends and receives.</param>
99	/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
100	public virtual async Task ListenAsync(RequestHandler requestHandler, CancellationToken cancellationToken = default)
101	{
102	_transport?.Dispose();	×
103	_application?.Dispose();	×
104
105	if (requestHandler == null)	1✔
106	{
107	throw new ArgumentNullException(nameof(requestHandler));	×
108	}
109
110	var duplexPipePair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);	1✔
111
112	// Create transport and application
113	_transport = CreateStreamingTransport(duplexPipePair.Application);	1✔
114	_application = new TransportHandler(duplexPipePair.Transport, Logger);	1✔
115
116	// Create session
117	_session = new StreamingSession(requestHandler, _application, Logger, cancellationToken);	1✔
118
119	// Start transport and application
120	var transportTask = _transport.ConnectAsync((connected) => IsConnected = connected, cancellationToken);	1✔
121	var applicationTask = _application.ListenAsync(cancellationToken);	1✔
122
123	var tasks = new List<Task> { transportTask, applicationTask };	1✔
124
125	// Signal that session is ready to be used
126	_sessionInitializedTask.SetResult(true);	1✔
127
128	// Let application and transport run
129	await Task.WhenAll(tasks).ConfigureAwait(false);	1✔
130	}	1✔
131
132	/// <inheritdoc />
133	public void Dispose()
134	{
135	// Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
136	Dispose(disposing: true);	1✔
137	GC.SuppressFinalize(this);	1✔
138	}	1✔
139
140	internal abstract StreamingTransport CreateStreamingTransport(IDuplexPipe application);
141
142	/// <summary>
143	/// Disposes managed and unmanaged resources of the underlying <see cref="StreamingConnection"/>.
144	/// </summary>
145	/// <param name="disposing">Whether we are disposing managed resources.</param>
146	protected virtual void Dispose(bool disposing)
147	{
148	if (!_disposedValue)	1✔
149	{
150	if (disposing)	1✔
151	{
152	_transport?.Dispose();	×
153	_application?.Dispose();	×
154	}
155
156	_disposedValue = true;	1✔
157	}
158	}	1✔
159	}
160	}

microsoft / botbuilder-dotnet / 388024

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous