Jericho / ZoomNet / 650

Committed 02 Nov 2023 04:28PM UTC coverage: 18.602% (+0.8%) from 17.818%

Build # 650

Build Type

push

appveyor

Committed by

Jericho

Commit Message

Merge branch 'release/0.68.0'

Run Details

551 of 2962 relevant lines covered (18.6%)

3.33 hits per line

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

0.0

/Source/ZoomNet/ZoomWebSocketClient.cs

using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using System;
using System.Net;
using System.Net.Http;
using System.Net.WebSockets;
using System.Reactive.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Websocket.Client;
using ZoomNet.Models;
using ZoomNet.Models.Webhooks;
using ZoomNet.Utilities;

namespace ZoomNet
{
        /// <summary>
        /// Client for Zoom's WebSocket webhooks.
        /// </summary>
        public class ZoomWebSocketClient : IDisposable
        {
                private readonly string _subscriptionId;
                private readonly ILogger _logger;
                private readonly IWebProxy _proxy;
                private readonly Func<Event, CancellationToken, Task> _eventProcessor;

                private WebsocketClient _websocketClient;
                private HttpClient _httpClient;
                private ITokenHandler _tokenHandler;

                /// <summary>
                /// Initializes a new instance of the <see cref="ZoomWebSocketClient"/> class.
                /// </summary>
                /// <param name="connectionInfo">Connection information.</param>
                /// <param name="subscriptionId">Your subscirption Id.</param>
                /// <param name="eventProcessor">A delegate that will be invoked when a wehook message is received.</param>
                /// <param name="proxy">Allows you to specify a proxy.</param>
                /// <param name="logger">Logger.</param>
                public ZoomWebSocketClient(IConnectionInfo connectionInfo, string subscriptionId, Func<Event, CancellationToken, Task> eventProcessor, IWebProxy proxy = null, ILogger logger = null)
                {
                        // According to https://marketplace.zoom.us/docs/api-reference/websockets/, only Server-to-Server OAuth connections are supported
                        if (connectionInfo == null) throw new ArgumentNullException(nameof(connectionInfo));
                        if (connectionInfo is not OAuthConnectionInfo oAuthConnectionInfo || oAuthConnectionInfo.GrantType != OAuthGrantType.AccountCredentials)
                        {
                                throw new ArgumentException("WebSocket client only supports Server-to-Server OAuth connections");
                        }

                        _subscriptionId = subscriptionId ?? throw new ArgumentNullException(nameof(subscriptionId));
                        _eventProcessor = eventProcessor ?? throw new ArgumentNullException(nameof(eventProcessor));
                        _proxy = proxy;
                        _logger = logger ?? NullLogger.Instance;
                        _httpClient = new HttpClient(new HttpClientHandler { Proxy = _proxy, UseProxy = _proxy != null });
                        _tokenHandler = new OAuthTokenHandler(oAuthConnectionInfo, _httpClient);

                        var clientFactory = new Func<Uri, CancellationToken, Task<WebSocket>>(async (uri, cancellationToken) =>
                        {
                                _logger.LogTrace("Establishing connection to Zoom");

                                // The current value in the uri parameter must be ignored because it contains "access_token" which may have expired.
                                // The following line ensures the "access_token" is refreshed whenever it expires.
                                uri = new Uri($"wss://ws.zoom.us/ws?subscriptionId={_subscriptionId}&access_token={_tokenHandler.Token}");

                                var client = new ClientWebSocket()
                                {
                                        Options =
                                        {
                                                KeepAliveInterval = TimeSpan.Zero, // Turn off built-in "Keep Alive" feature because Zoom uses proprietary "heartbeat" every 30 seconds rather than standard "pong" messages at regular interval.
                                                Proxy = _proxy,
                                        }
                                };
                                client.Options.SetRequestHeader("ZoomNet-Version", ZoomClient.Version);

                                await client.ConnectAsync(uri, cancellationToken).ConfigureAwait(false);
                                return client;
                        });

                        _websocketClient = new WebsocketClient(new Uri("wss://ws.zoom.us"), clientFactory)
                        {
                                Name = "ZoomNet",
                                ReconnectTimeout = TimeSpan.FromSeconds(45), // Greater than 30 seconds because we send a heartbeat every 30 seconds
                                ErrorReconnectTimeout = TimeSpan.FromSeconds(45)
                        };
                        _websocketClient.ReconnectionHappened.Subscribe(info => _logger.LogTrace("Reconnection happened, type: {reconnectionReason}", info.Type));
                        _websocketClient.DisconnectionHappened.Subscribe(info => _logger.LogTrace("Disconnection happened, type: {disconnectionReason}", info.Type));
                }

                /// <summary>
                /// Start listening to incoming webhooks from Zoom.
                /// </summary>
                /// <param name="cancellationToken">The cancellation token.</param>
                /// <returns>Asynchronous task.</returns>
                public Task StartAsync(CancellationToken cancellationToken = default)
                {
                        _websocketClient.MessageReceived
                                .Select(response => Observable.FromAsync(() => ProcessMessage(response, cancellationToken)))
                                .Merge(5) // Allow up to 5 messages to be processed concurently. This number is arbitrary but it seems reasonable.
                                .Subscribe();

                        Task.Run(() => SendHeartbeat(_websocketClient, cancellationToken), cancellationToken);

                        return _websocketClient.Start();
                }

                /// <summary>
                /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
                /// </summary>
                public void Dispose()
                {
                        // Call 'Dispose' to release resources
                        Dispose(true);

                        // Tell the GC that we have done the cleanup and there is nothing left for the Finalizer to do
                        GC.SuppressFinalize(this);
                }

                /// <summary>
                /// Releases unmanaged and - optionally - managed resources.
                /// </summary>
                /// <param name="disposing"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
                protected virtual void Dispose(bool disposing)
                {
                        if (disposing)
                        {
                                ReleaseManagedResources();
                        }
                        else
                        {
                                // The object went out of scope and the Finalizer has been called.
                                // The GC will take care of releasing managed resources, therefore there is nothing to do here.
                        }

                        ReleaseUnmanagedResources();
                }

                private async Task SendHeartbeat(IWebsocketClient client, CancellationToken cancellationToken = default)
                {
                        while (!cancellationToken.IsCancellationRequested)
                        {
                                await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken); // Zoom requires a heartbeat every 30 seconds

                                if (!client.IsRunning)
                                {
                                        _logger.LogTrace("Client is not running. Skipping heartbeat");
                                        continue;
                                }

                                _logger.LogTrace("Sending heartbeat");

                                await client.SendInstant("{\"module\":\"heartbeat\"}").ConfigureAwait(false);
                        }
                }

                private async Task ProcessMessage(ResponseMessage msg, CancellationToken cancellationToken = default)
                {
                        var jsonDoc = JsonDocument.Parse(msg.Text);
                        var module = jsonDoc.RootElement.GetPropertyValue("module", string.Empty);
                        var success = jsonDoc.RootElement.GetPropertyValue("success", true);
                        var content = jsonDoc.RootElement.GetPropertyValue("content", string.Empty);

                        if (content.Equals("Invalid Token", StringComparison.OrdinalIgnoreCase))
                        {
                                _logger.LogTrace("{module}. Token is invalid (presumably expired). Requesting a new token...", module);
                                _tokenHandler.RefreshTokenIfNecessary(true);
                                return;
                        }
                        else if (!success)
                        {
                                _logger.LogTrace("FAILURE: Received message: {module}. {content}", module, content);
                                return;
                        }

                        switch (module)
                        {
                                case "build_connection":
                                        _logger.LogTrace("Received message: {module}. Connection has been established.", module);
                                        break;
                                case "heartbeat":
                                        _logger.LogTrace("Received message: {module}. Server is acknowledging heartbeat.", module);
                                        break;
                                case "message":
                                        var parser = new WebhookParser();
                                        var webhookEvent = parser.ParseEventWebhook(content);
                                        var eventType = webhookEvent.EventType;
                                        _logger.LogTrace("Received webhook event: {eventType}", eventType);
                                        try
                                        {
                                                await _eventProcessor(webhookEvent, cancellationToken).ConfigureAwait(false);
                                        }
                                        catch (Exception ex)
                                        {
                                                _logger.LogError(ex, "An error occurred while processing webhook event: {eventType}", eventType);
                                        }

                                        break;
                                default:
                                        _logger.LogError("Received unknown message: {module}", module);
                                        break;
                        }
                }

                private void ReleaseManagedResources()
                {
                        _tokenHandler = null;

                        if (_websocketClient != null)
                        {
                                if (_websocketClient.IsRunning)
                                {
                                        _websocketClient.Stop(WebSocketCloseStatus.NormalClosure, "Shutting down").GetAwaiter().GetResult();
                                }

                                _websocketClient.Dispose();
                                _websocketClient = null;
                        }

                        if (_httpClient != null)
                        {
                                _httpClient.Dispose();
                                _httpClient = null;
                        }
                }

                private void ReleaseUnmanagedResources()
                {
                        // We do not hold references to unmanaged resources
                }
        }
}

1	using Microsoft.Extensions.Logging;
2	using Microsoft.Extensions.Logging.Abstractions;
3	using System;
4	using System.Net;
5	using System.Net.Http;
6	using System.Net.WebSockets;
7	using System.Reactive.Linq;
8	using System.Text.Json;
9	using System.Threading;
10	using System.Threading.Tasks;
11	using Websocket.Client;
12	using ZoomNet.Models;
13	using ZoomNet.Models.Webhooks;
14	using ZoomNet.Utilities;
15
16	namespace ZoomNet
17	{
18	/// <summary>
19	/// Client for Zoom's WebSocket webhooks.
20	/// </summary>
21	public class ZoomWebSocketClient : IDisposable
22	{
23	private readonly string _subscriptionId;
24	private readonly ILogger _logger;
25	private readonly IWebProxy _proxy;
26	private readonly Func<Event, CancellationToken, Task> _eventProcessor;
27
28	private WebsocketClient _websocketClient;
29	private HttpClient _httpClient;
30	private ITokenHandler _tokenHandler;
31
32	/// <summary>
33	/// Initializes a new instance of the <see cref="ZoomWebSocketClient"/> class.
34	/// </summary>
35	/// <param name="connectionInfo">Connection information.</param>
36	/// <param name="subscriptionId">Your subscirption Id.</param>
37	/// <param name="eventProcessor">A delegate that will be invoked when a wehook message is received.</param>
38	/// <param name="proxy">Allows you to specify a proxy.</param>
39	/// <param name="logger">Logger.</param>
40	public ZoomWebSocketClient(IConnectionInfo connectionInfo, string subscriptionId, Func<Event, CancellationToken, Task> eventProcessor, IWebProxy proxy = null, ILogger logger = null)
41	{
42	// According to https://marketplace.zoom.us/docs/api-reference/websockets/, only Server-to-Server OAuth connections are supported
43	if (connectionInfo == null) throw new ArgumentNullException(nameof(connectionInfo));	×
44	if (connectionInfo is not OAuthConnectionInfo oAuthConnectionInfo \|\| oAuthConnectionInfo.GrantType != OAuthGrantType.AccountCredentials)	×
45	{
46	throw new ArgumentException("WebSocket client only supports Server-to-Server OAuth connections");	×
47	}
48
49	_subscriptionId = subscriptionId ?? throw new ArgumentNullException(nameof(subscriptionId));	×
50	_eventProcessor = eventProcessor ?? throw new ArgumentNullException(nameof(eventProcessor));	×
51	_proxy = proxy;	×
52	_logger = logger ?? NullLogger.Instance;	×
53	_httpClient = new HttpClient(new HttpClientHandler { Proxy = _proxy, UseProxy = _proxy != null });	×
54	_tokenHandler = new OAuthTokenHandler(oAuthConnectionInfo, _httpClient);	×
55
56	var clientFactory = new Func<Uri, CancellationToken, Task<WebSocket>>(async (uri, cancellationToken) =>	×
57	{	×
58	_logger.LogTrace("Establishing connection to Zoom");	×
59		×
60	// The current value in the uri parameter must be ignored because it contains "access_token" which may have expired.	×
61	// The following line ensures the "access_token" is refreshed whenever it expires.	×
62	uri = new Uri($"wss://ws.zoom.us/ws?subscriptionId={_subscriptionId}&access_token={_tokenHandler.Token}");	×
63		×
64	var client = new ClientWebSocket()	×
65	{	×
66	Options =	×
67	{	×
68	KeepAliveInterval = TimeSpan.Zero, // Turn off built-in "Keep Alive" feature because Zoom uses proprietary "heartbeat" every 30 seconds rather than standard "pong" messages at regular interval.	×
69	Proxy = _proxy,	×
70	}	×
71	};	×
72	client.Options.SetRequestHeader("ZoomNet-Version", ZoomClient.Version);	×
73		×
74	await client.ConnectAsync(uri, cancellationToken).ConfigureAwait(false);	×
75	return client;	×
76	});	×
77
78	_websocketClient = new WebsocketClient(new Uri("wss://ws.zoom.us"), clientFactory)	×
79	{	×
80	Name = "ZoomNet",	×
81	ReconnectTimeout = TimeSpan.FromSeconds(45), // Greater than 30 seconds because we send a heartbeat every 30 seconds	×
82	ErrorReconnectTimeout = TimeSpan.FromSeconds(45)	×
83	};	×
84	_websocketClient.ReconnectionHappened.Subscribe(info => _logger.LogTrace("Reconnection happened, type: {reconnectionReason}", info.Type));	×
85	_websocketClient.DisconnectionHappened.Subscribe(info => _logger.LogTrace("Disconnection happened, type: {disconnectionReason}", info.Type));	×
86	}	×
87
88	/// <summary>
89	/// Start listening to incoming webhooks from Zoom.
90	/// </summary>
91	/// <param name="cancellationToken">The cancellation token.</param>
92	/// <returns>Asynchronous task.</returns>
93	public Task StartAsync(CancellationToken cancellationToken = default)
94	{
95	_websocketClient.MessageReceived	×
96	.Select(response => Observable.FromAsync(() => ProcessMessage(response, cancellationToken)))	×
97	.Merge(5) // Allow up to 5 messages to be processed concurently. This number is arbitrary but it seems reasonable.	×
98	.Subscribe();	×
99
100	Task.Run(() => SendHeartbeat(_websocketClient, cancellationToken), cancellationToken);	×
101
102	return _websocketClient.Start();	×
103	}
104
105	/// <summary>
106	/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
107	/// </summary>
108	public void Dispose()
109	{
110	// Call 'Dispose' to release resources
111	Dispose(true);	×
112
113	// Tell the GC that we have done the cleanup and there is nothing left for the Finalizer to do
114	GC.SuppressFinalize(this);	×
115	}	×
116
117	/// <summary>
118	/// Releases unmanaged and - optionally - managed resources.
119	/// </summary>
120	/// <param name="disposing"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
121	protected virtual void Dispose(bool disposing)
122	{
123	if (disposing)	×
124	{
125	ReleaseManagedResources();	×
126	}
127	else
128	{
129	// The object went out of scope and the Finalizer has been called.
130	// The GC will take care of releasing managed resources, therefore there is nothing to do here.
131	}
132
133	ReleaseUnmanagedResources();	×
134	}	×
135
136	private async Task SendHeartbeat(IWebsocketClient client, CancellationToken cancellationToken = default)
137	{
138	while (!cancellationToken.IsCancellationRequested)
139	{
140	await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken); // Zoom requires a heartbeat every 30 seconds
141
142	if (!client.IsRunning)
143	{
144	_logger.LogTrace("Client is not running. Skipping heartbeat");
145	continue;
146	}
147
148	_logger.LogTrace("Sending heartbeat");
149
150	await client.SendInstant("{\"module\":\"heartbeat\"}").ConfigureAwait(false);
151	}
152	}
153
154	private async Task ProcessMessage(ResponseMessage msg, CancellationToken cancellationToken = default)
155	{
156	var jsonDoc = JsonDocument.Parse(msg.Text);
157	var module = jsonDoc.RootElement.GetPropertyValue("module", string.Empty);
158	var success = jsonDoc.RootElement.GetPropertyValue("success", true);
159	var content = jsonDoc.RootElement.GetPropertyValue("content", string.Empty);
160
161	if (content.Equals("Invalid Token", StringComparison.OrdinalIgnoreCase))
162	{
163	_logger.LogTrace("{module}. Token is invalid (presumably expired). Requesting a new token...", module);
164	_tokenHandler.RefreshTokenIfNecessary(true);
165	return;
166	}
167	else if (!success)
168	{
169	_logger.LogTrace("FAILURE: Received message: {module}. {content}", module, content);
170	return;
171	}
172
173	switch (module)
174	{
175	case "build_connection":
176	_logger.LogTrace("Received message: {module}. Connection has been established.", module);
177	break;
178	case "heartbeat":
179	_logger.LogTrace("Received message: {module}. Server is acknowledging heartbeat.", module);
180	break;
181	case "message":
182	var parser = new WebhookParser();
183	var webhookEvent = parser.ParseEventWebhook(content);
184	var eventType = webhookEvent.EventType;
185	_logger.LogTrace("Received webhook event: {eventType}", eventType);
186	try
187	{
188	await _eventProcessor(webhookEvent, cancellationToken).ConfigureAwait(false);
189	}
190	catch (Exception ex)
191	{
192	_logger.LogError(ex, "An error occurred while processing webhook event: {eventType}", eventType);
193	}
194
195	break;
196	default:
197	_logger.LogError("Received unknown message: {module}", module);
198	break;
199	}
200	}
201
202	private void ReleaseManagedResources()
203	{
204	_tokenHandler = null;	×
205
206	if (_websocketClient != null)	×
207	{
208	if (_websocketClient.IsRunning)	×
209	{
210	_websocketClient.Stop(WebSocketCloseStatus.NormalClosure, "Shutting down").GetAwaiter().GetResult();	×
211	}
212
213	_websocketClient.Dispose();	×
214	_websocketClient = null;	×
215	}
216
217	if (_httpClient != null)	×
218	{
219	_httpClient.Dispose();	×
220	_httpClient = null;	×
221	}
222	}	×
223
224	private void ReleaseUnmanagedResources()
225	{
226	// We do not hold references to unmanaged resources
227	}	×
228	}
229	}

Jericho / ZoomNet / 650

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