server/src/Transports.AspNetCore/WebSockets/BaseSubscriptionServer.Observer.cs at 8412702df3fea97ca739b8b8f21cfb6a810311c1 · graphql-dotnet/server · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
namespace GraphQL.Server.Transports.AspNetCore.WebSockets;

public abstract partial class BaseSubscriptionServer
{
    /// <summary>
    /// Handles messages from the event source.
    /// </summary>
    private class Observer : IObserver<ExecutionResult>
    {
        private readonly BaseSubscriptionServer _server;
        private readonly string _id;
        private readonly bool _closeAfterOnError;
        private readonly bool _closeAfterAnyError;
        private int _done;

        public Observer(BaseSubscriptionServer server, string id, bool closeAfterOnError, bool closeAfterAnyError)
        {
            _server = server;
            _id = id;
            _closeAfterOnError = closeAfterOnError;
            _closeAfterAnyError = closeAfterAnyError;
        }

        public void OnCompleted()
        {
            if (Interlocked.Exchange(ref _done, 1) == 1)
                return;
            try
            {
                _ = _server.SendCompletedAsync(_id);
            }
            catch { }
        }

        public async void OnError(Exception error)
        {
            if (Thread.VolatileRead(ref _done) == 1)
                return;
            if (_closeAfterOnError && Interlocked.Exchange(ref _done, 1) == 1)
                return;
            try
            {
                // although error should never be null, if the event source does call OnError(null!),
                // skip sending an error packet/message (allowed by spec)
                if (error != null)
                {
                    var executionError = error is ExecutionError ee ? ee : await _server.HandleErrorFromSourceAsync(error);
                    if (executionError != null)
                    {
                        var result = new ExecutionResult
                        {
                            Errors = new ExecutionErrors { executionError },
                        };
                        await _server.SendDataAsync(_id, result);
                    }
                }
            }
            catch { }
            try
            {
                if (_closeAfterOnError)
                    await _server.SendCompletedAsync(_id);
            }
            catch { }
        }

        public async void OnNext(ExecutionResult value)
        {
            if (value == null || Thread.VolatileRead(ref _done) == 1)
                return;
            try
            {
                await _server.SendDataAsync(_id, value);
                if (_closeAfterAnyError && value.Errors?.Count > 0)
                {
                    if (Interlocked.Exchange(ref _done, 1) == 1)
                        return;
                    await _server.SendCompletedAsync(_id);
                }
            }
            catch { }
        }
    }
}