| | | 1 | | #nullable enable |
| | | 2 | | using System; |
| | | 3 | | using System.Collections.Concurrent; |
| | | 4 | | using System.Threading; |
| | | 5 | | using System.Threading.Tasks; |
| | | 6 | | using dotnet_etcd.interfaces; |
| | | 7 | | using Etcdserverpb; |
| | | 8 | | using Grpc.Core; |
| | | 9 | | |
| | | 10 | | namespace dotnet_etcd; |
| | | 11 | | |
| | | 12 | | /// <summary> |
| | | 13 | | /// Manages a bidirectional streaming connection to the etcd watch API |
| | | 14 | | /// </summary> |
| | | 15 | | public class Watcher : IWatcher |
| | | 16 | | { |
| | 61 | 17 | | private readonly ConcurrentDictionary<long, Action<WatchResponse>> _callbacks = new(); |
| | 61 | 18 | | private readonly CancellationTokenSource _cts = new(); |
| | | 19 | | |
| | | 20 | | private readonly IAsyncDuplexStreamingCall<WatchRequest, WatchResponse> _streamingCall; |
| | | 21 | | private readonly Action? _onConnectionFailure; |
| | | 22 | | |
| | | 23 | | |
| | | 24 | | /// <summary> |
| | | 25 | | /// Creates a new Watcher |
| | | 26 | | /// </summary> |
| | | 27 | | /// <param name="streamingCall">The streaming call to use</param> |
| | | 28 | | /// <param name="onConnectionFailure">Action to invoke when connection fails</param> |
| | 61 | 29 | | public Watcher(IAsyncDuplexStreamingCall<WatchRequest, WatchResponse> streamingCall, Action? onConnectionFailure = n |
| | 61 | 30 | | { |
| | 61 | 31 | | _streamingCall = streamingCall ?? throw new ArgumentNullException(nameof(streamingCall)); |
| | 60 | 32 | | _onConnectionFailure = onConnectionFailure; |
| | 60 | 33 | | _ = ProcessWatchResponses(); |
| | 60 | 34 | | } |
| | | 35 | | |
| | | 36 | | |
| | | 37 | | /// <summary> |
| | | 38 | | /// Creates a watch for the specified request |
| | | 39 | | /// </summary> |
| | | 40 | | /// <param name="request">The watch request</param> |
| | | 41 | | /// <param name="callback">The callback to invoke when a watch event is received</param> |
| | | 42 | | /// <returns>A task that completes when the watch is created</returns> |
| | | 43 | | public async Task CreateWatchAsync(WatchRequest request, Action<WatchResponse> callback) |
| | 71 | 44 | | { |
| | 71 | 45 | | ArgumentNullException.ThrowIfNull(request); |
| | | 46 | | |
| | 70 | 47 | | ArgumentNullException.ThrowIfNull(callback); |
| | | 48 | | |
| | 69 | 49 | | _callbacks[request.CreateRequest.WatchId] = callback; |
| | | 50 | | |
| | | 51 | | // Send the watch request |
| | 69 | 52 | | await _streamingCall.RequestStream.WriteAsync(request); |
| | 69 | 53 | | } |
| | | 54 | | |
| | | 55 | | /// <summary> |
| | | 56 | | /// Cancels a watch with the specified ID |
| | | 57 | | /// </summary> |
| | | 58 | | /// <param name="watchId">The ID of the watch to cancel</param> |
| | | 59 | | /// <returns>A task that completes when the watch is canceled</returns> |
| | | 60 | | public async Task CancelWatchAsync(long watchId) |
| | 2 | 61 | | { |
| | | 62 | | // Send a cancel request |
| | 2 | 63 | | WatchRequest request = new() { CancelRequest = new WatchCancelRequest { WatchId = watchId } }; |
| | | 64 | | |
| | 2 | 65 | | await _streamingCall.RequestStream.WriteAsync(request); |
| | | 66 | | |
| | | 67 | | // Remove the callback |
| | 2 | 68 | | _callbacks.TryRemove(watchId, out _); |
| | 2 | 69 | | } |
| | | 70 | | |
| | | 71 | | private async Task ProcessWatchResponses() |
| | 60 | 72 | | { |
| | | 73 | | try |
| | 60 | 74 | | { |
| | 140 | 75 | | while (await _streamingCall.ResponseStream.MoveNext(_cts.Token)) |
| | 80 | 76 | | { |
| | 80 | 77 | | WatchResponse response = _streamingCall.ResponseStream.Current; |
| | 80 | 78 | | if (!_callbacks.TryGetValue(response.WatchId, out Action<WatchResponse>? cb)) |
| | 3 | 79 | | { |
| | 3 | 80 | | continue; |
| | | 81 | | } |
| | | 82 | | |
| | 77 | 83 | | cb(response); |
| | | 84 | | |
| | | 85 | | // If the watch was canceled, remove the callback after invoking it |
| | 77 | 86 | | if (response.Canceled) |
| | 2 | 87 | | { |
| | 2 | 88 | | _callbacks.TryRemove(response.WatchId, out _); |
| | 2 | 89 | | } |
| | 77 | 90 | | } |
| | 0 | 91 | | } |
| | 6 | 92 | | catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled) |
| | 1 | 93 | | { |
| | | 94 | | // This is expected when the stream is canceled |
| | 1 | 95 | | } |
| | 28 | 96 | | catch (OperationCanceledException) |
| | 28 | 97 | | { |
| | | 98 | | // This is expected when the token is canceled |
| | 28 | 99 | | } |
| | 5 | 100 | | catch (RpcException ex) |
| | 5 | 101 | | { |
| | | 102 | | // Log a simplified message for expected connection failures |
| | 5 | 103 | | Console.WriteLine($"Watch stream connection lost: {ex.StatusCode} - {ex.Message}"); |
| | 5 | 104 | | _onConnectionFailure?.Invoke(); |
| | 5 | 105 | | } |
| | 0 | 106 | | catch (Exception ex) |
| | 0 | 107 | | { |
| | | 108 | | // Log the exception |
| | 0 | 109 | | await Console.Error.WriteAsync($"Error processing watch responses: {ex}"); |
| | 0 | 110 | | _onConnectionFailure?.Invoke(); |
| | | 111 | | #if DEBUG |
| | | 112 | | // Only re-throw in debug mode to help with debugging |
| | 0 | 113 | | throw; |
| | | 114 | | #endif |
| | | 115 | | } |
| | 34 | 116 | | } |
| | | 117 | | |
| | | 118 | | /// <summary> |
| | | 119 | | /// Disposes the watch stream |
| | | 120 | | /// </summary> |
| | | 121 | | public void Dispose() |
| | 30 | 122 | | { |
| | 30 | 123 | | _cts.Cancel(); |
| | 30 | 124 | | _streamingCall.Dispose(); |
| | | 125 | | |
| | 30 | 126 | | GC.SuppressFinalize(this); |
| | 30 | 127 | | } |
| | | 128 | | } |