How to use HandleAsyncMessage method of PuppeteerSharp.Helpers.AsyncMessageQueue class

1using System;
2using System.Collections.Generic;
3using System.Threading;
4using System.Threading.Tasks;
5using Microsoft.Extensions.Logging;
6using Microsoft.Extensions.Logging.Abstractions;
7using PuppeteerSharp.Messaging;
8
9namespace PuppeteerSharp.Helpers
10{
11    /// <summary>
12    /// Provides an async queue for responses for <see cref="CDPSession.SendAsync"/>, so that responses can be handled
13    /// async without risk callers causing a deadlock.
14    /// </summary>
15    internal class AsyncMessageQueue : IDisposable
16    {
17        private readonly List<MessageTask> _pendingTasks;
18        private readonly bool _enqueueAsyncMessages;
19        private readonly ILogger _logger;
20        private bool _disposed;
21
22        public AsyncMessageQueue(bool enqueueAsyncMessages, ILogger logger = null)
23        {
24            _enqueueAsyncMessages = enqueueAsyncMessages;
25            _logger = logger ?? NullLogger.Instance;
26            _pendingTasks = new List<MessageTask>();
27        }
28
29        public void Enqueue(MessageTask callback, ConnectionResponse obj)
30        {
31            if (_disposed)
32            {
33                throw new ObjectDisposedException(GetType().FullName);
34            }
35
36            if (!_enqueueAsyncMessages)
37            {
38                HandleAsyncMessage(callback, obj);
39                return;
40            }
41
42            // Keep a ref to this task until it completes. If it can't finish by the time we dispose this queue,
43            // then we'll find it and cancel it.
44            lock (_pendingTasks)
45            {
46                _pendingTasks.Add(callback);
47            }
48
49            var task = Task.Run(() => HandleAsyncMessage(callback, obj));
50
51            // Unhandled error handler
52            task.ContinueWith(
53                t =>
54                {
55                    _logger.LogError(t.Exception, "Failed to complete async handling of SendAsync for {callback}", callback.Method);
56                    callback.TaskWrapper.TrySetException(t.Exception!); // t.Exception is available since this runs only on faulted
57                },
58                CancellationToken.None,
59                TaskContinuationOptions.OnlyOnFaulted,
60                TaskScheduler.Default);
61
62            // Always remove from the queue when done, regardless of outcome.
63            task.ContinueWith(
64                _ =>
65                {
66                    lock (_pendingTasks)
67                    {
68                        _pendingTasks.Remove(callback);
69                    }
70                },
71                TaskScheduler.Default);
72        }
73
74        public void Dispose()
75        {
76            if (_disposed)
77            {
78                return;
79            }
80
81            // Ensure all tasks are finished since we're disposing now. Any pending tasks will be canceled.
82            MessageTask[] pendingTasks;
83            lock (_pendingTasks)
84            {
85                pendingTasks = _pendingTasks.ToArray();
86                _pendingTasks.Clear();
87            }
88
89            foreach (var pendingTask in pendingTasks)
90            {
91                pendingTask.TaskWrapper.TrySetCanceled();
92            }
93
94            _disposed = true;
95        }
96
97        private static void HandleAsyncMessage(MessageTask callback, ConnectionResponse obj)
98        {
99            if (obj.Error != null)
100            {
101                callback.TaskWrapper.TrySetException(new MessageException(callback, obj.Error));
102            }
103            else
104            {
105                callback.TaskWrapper.TrySetResult(obj.Result);
106            }
107        }
108    }
109}
110

Source: AsyncMessageQueue.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using PuppeteerSharp.Messaging;

namespace PuppeteerSharp.Helpers
{
    /// <summary>
    /// Provides an async queue for responses for <see cref="CDPSession.SendAsync"/>, so that responses can be handled
    /// async without risk callers causing a deadlock.
    /// </summary>
    internal class AsyncMessageQueue : IDisposable
    {
        private readonly List<MessageTask> _pendingTasks;
        private readonly bool _enqueueAsyncMessages;
        private readonly ILogger _logger;
        private bool _disposed;

        public AsyncMessageQueue(bool enqueueAsyncMessages, ILogger logger = null)
        {
            _enqueueAsyncMessages = enqueueAsyncMessages;
            _logger = logger ?? NullLogger.Instance;
            _pendingTasks = new List<MessageTask>();
        }

        public void Enqueue(MessageTask callback, ConnectionResponse obj)
        {
            if (_disposed)
            {
                throw new ObjectDisposedException(GetType().FullName);
            }

            if (!_enqueueAsyncMessages)
            {
                HandleAsyncMessage(callback, obj);
                return;
            }

            // Keep a ref to this task until it completes. If it can't finish by the time we dispose this queue,
            // then we'll find it and cancel it.
            lock (_pendingTasks)
            {
                _pendingTasks.Add(callback);
            }

            var task = Task.Run(() => HandleAsyncMessage(callback, obj));

            // Unhandled error handler
            task.ContinueWith(
                t =>
                {
                    _logger.LogError(t.Exception, "Failed to complete async handling of SendAsync for {callback}", callback.Method);
                    callback.TaskWrapper.TrySetException(t.Exception!); // t.Exception is available since this runs only on faulted
                },
                CancellationToken.None,
                TaskContinuationOptions.OnlyOnFaulted,
                TaskScheduler.Default);

            // Always remove from the queue when done, regardless of outcome.
            task.ContinueWith(
                _ =>
                {
                    lock (_pendingTasks)
                    {
                        _pendingTasks.Remove(callback);
                    }
                },
                TaskScheduler.Default);
        }

        public void Dispose()
        {
            if (_disposed)
            {
                return;
            }

            // Ensure all tasks are finished since we're disposing now. Any pending tasks will be canceled.
            MessageTask[] pendingTasks;
            lock (_pendingTasks)
            {
                pendingTasks = _pendingTasks.ToArray();
                _pendingTasks.Clear();
            }

            foreach (var pendingTask in pendingTasks)
            {
                pendingTask.TaskWrapper.TrySetCanceled();
            }

            _disposed = true;
        }

        private static void HandleAsyncMessage(MessageTask callback, ConnectionResponse obj)
        {
            if (obj.Error != null)
            {
                callback.TaskWrapper.TrySetException(new MessageException(callback, obj.Error));
            }
            else
            {
                callback.TaskWrapper.TrySetResult(obj.Result);
            }
        }
    }
}