< Summary

Information
Class: dotnet_etcd.WatchManager
Assembly: dotnet-etcd
File(s): /home/runner/work/dotnet-etcd/dotnet-etcd/dotnet-etcd/watchclient/WatchManager.cs
Line coverage
92%
Covered lines: 393
Uncovered lines: 30
Coverable lines: 423
Total lines: 756
Line coverage: 92.9%
Branch coverage
76%
Covered branches: 60
Total branches: 78
Branch coverage: 76.9%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%22100%
WatchAsync()100%11100%
WrappedCallback()75%4481.81%
TrackResumeRevision(...)77.77%181892.59%
Watch(...)100%11100%
Watch(...)100%11100%
Callback()75%4480%
WatchRange(...)75%4493.33%
Watch(...)75%4493.33%
WatchRange(...)75%4493.54%
WatchAsync()100%11100%
WatchRangeAsync()100%11100%
WatchAsync()100%11100%
WatchRangeAsync()100%11100%
WatchRange(...)100%11100%
CancelWatch(...)66.66%66100%
Dispose()100%66100%
GetRangeEnd(...)50%4472.72%
GetServerWatchId(...)50%4477.77%
EnsureWatchStream(...)100%22100%
HandleConnectionFailure()100%1187.5%
get_WatchId()100%11100%
get_CancellationTokenSource()100%11100%
get_Request()100%11100%
get_Callback()100%11100%
get_NextRevision()100%11100%

File(s)

/home/runner/work/dotnet-etcd/dotnet-etcd/dotnet-etcd/watchclient/WatchManager.cs

#LineLine coverage
 1#nullable enable
 2using System;
 3using System.Collections.Concurrent;
 4using System.Collections.Generic;
 5using System.Text;
 6using System.Threading;
 7using System.Threading.Tasks;
 8using dotnet_etcd.interfaces;
 9using Etcdserverpb;
 10using Google.Protobuf;
 11using Grpc.Core;
 12using Mvccpb;
 13
 14namespace dotnet_etcd;
 15
 16/// <summary>
 17///     Manages watch streams and provides a way to cancel watches
 18/// </summary>
 19public class WatchManager : IWatchManager
 20{
 35421    private readonly object _lockObject = new();
 35422    private readonly ConcurrentDictionary<long, WatchCancellation> _watches = new();
 35423    private readonly ConcurrentDictionary<long, long> _watchIdMapping = new();
 24
 25    private readonly
 26        Func<Metadata?, DateTime?, CancellationToken, IAsyncDuplexStreamingCall<WatchRequest, WatchResponse>>
 27        _watchStreamFactory;
 28
 29    private bool _disposed;
 35430    private long _nextWatchId = 1;
 31    private Watcher? _watchStream;
 32
 33    /// <summary>
 34    ///     Creates a new WatchManager
 35    /// </summary>
 36    /// <param name="watchStreamFactory">A factory function that creates a new watch stream</param>
 35437    public WatchManager(
 35438        Func<Metadata?, DateTime?, CancellationToken, IAsyncDuplexStreamingCall<WatchRequest, WatchResponse>>
 70839            watchStreamFactory) => _watchStreamFactory =
 35440        watchStreamFactory ?? throw new ArgumentNullException(nameof(watchStreamFactory));
 41
 42    /// <summary>
 43    ///     Creates a new watch request
 44    /// </summary>
 45    /// <param name="request">The watch requests to create</param>
 46    /// <param name="callback">The callback to invoke when a watch event is received</param>
 47    /// <param name="headers">The initial metadata to send with the call</param>
 48    /// <param name="deadline">An optional deadline for the call</param>
 49    /// <param name="cancellationToken">An optional token for canceling the call</param>
 50    /// <returns>A watch ID that can be used to cancel the watch</returns>
 51    public async Task<long> WatchAsync(WatchRequest request, Action<WatchResponse> callback, Metadata? headers = null,
 52        DateTime? deadline = null, CancellationToken cancellationToken = default)
 6553    {
 6554        ObjectDisposedException.ThrowIf(_disposed, this);
 55
 56        // Create a new watch stream if needed
 6457        EnsureWatchStream(headers, deadline, cancellationToken);
 58
 59        // Generate a new watch ID
 6460        long watchId = Interlocked.Increment(ref _nextWatchId);
 61
 62        // Create a cancellation token source that can be used to cancel the watch
 6463        CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 64
 65        // Create a watch cancellation object
 6466        WatchCancellation watchCancellation = new()
 6467        {
 6468            WatchId = watchId,
 6469            CancellationTokenSource = cts,
 6470            Request = request,
 6471            Callback = WrappedCallback
 6472        };
 73
 6474        request.CreateRequest.WatchId = watchId;
 75        // Create the watch
 6476        await _watchStream!.CreateWatchAsync(request, WrappedCallback).ConfigureAwait(false);
 77
 78        // Add the watch cancellation to the dictionary
 6479        _watches[watchId] = watchCancellation;
 80
 81        // Since we don't get a server watch ID from CreateWatchAsync, we can't map it
 82        // The server will assign a watch ID and include it in the watch response
 83        // Our wrappedCallback will handle this mapping when it receives the response
 84
 6485        return watchId;
 86
 87        // Create a wrapper callback that checks if the watch has been canceled
 88        void WrappedCallback(WatchResponse response)
 7789        {
 7790            if (cts.IsCancellationRequested)
 091            {
 092                return;
 93            }
 94
 95            // If this is a creation response, update our watch ID mapping
 7796            if (response.Created)
 1297            {
 98                // Map the server-assigned watch ID to our client-generated watch ID
 1299                _watchIdMapping[response.WatchId] = watchId;
 12100            }
 101
 102            // Track the revision to resume from on reconnect so no events are missed in the gap.
 77103            TrackResumeRevision(watchId, response);
 104
 77105            callback(response);
 77106        }
 64107    }
 108
 109    /// <summary>
 110    ///     Advances the stored resume revision for a watch based on a received response, mirroring the
 111    ///     etcd clientv3 "nextRev" logic: after an event batch resume from lastEvent.ModRevision + 1;
 112    ///     for a created/progress notification with no events, advance to the header revision. A
 113    ///     compaction cancel resets the resume point to the compact revision. Only ever moves forward.
 114    /// </summary>
 115    private void TrackResumeRevision(long clientWatchId, WatchResponse response)
 77116    {
 77117        if (!_watches.TryGetValue(clientWatchId, out WatchCancellation? watch))
 0118        {
 0119            return;
 120        }
 121
 77122        long candidate = watch.NextRevision;
 123
 77124        if (response.Canceled && response.CompactRevision > 0)
 1125        {
 126            // Our resume point was compacted away; the earliest we can resume from is CompactRevision.
 1127            candidate = response.CompactRevision;
 1128        }
 76129        else if (response.Events != null && response.Events.Count > 0)
 62130        {
 62131            long lastModRevision = response.Events[^1].Kv.ModRevision;
 62132            if (lastModRevision + 1 > candidate)
 61133            {
 61134                candidate = lastModRevision + 1;
 61135            }
 62136        }
 14137        else if (response.Header != null && response.Header.Revision > 0)
 11138        {
 139            // Created response or progress notification (no events): everything up to the header
 140            // revision has been observed, so the next start revision is header.Revision + 1.
 11141            long boundary = response.Header.Revision + 1;
 11142            if (boundary > candidate)
 11143            {
 11144                candidate = boundary;
 11145            }
 11146        }
 147
 77148        watch.NextRevision = candidate;
 77149    }
 150
 151    /// <summary>
 152    ///     Creates a new watch request
 153    /// </summary>
 154    /// <param name="request">The watch request to create</param>
 155    /// <param name="callback">The callback to invoke when a watch event is received</param>
 156    /// <param name="headers">The initial metadata to send with the call</param>
 157    /// <param name="deadline">An optional deadline for the call</param>
 158    /// <param name="cancellationToken">An optional token for canceling the call</param>
 159    /// <returns>A watch ID that can be used to cancel the watch</returns>
 160    public long Watch(WatchRequest request, Action<WatchResponse> callback, Metadata? headers = null,
 161        DateTime? deadline = null, CancellationToken cancellationToken = default)
 34162    {
 163        // Run the async method synchronously
 68164        Task<long> task = Task.Run(() => WatchAsync(request, callback, headers, deadline, cancellationToken), cancellati
 34165        task.Wait(cancellationToken);
 34166        return task.Result;
 34167    }
 168
 169    /// <summary>
 170    ///     Watches a specific key
 171    /// </summary>
 172    /// <param name="key">Key to watch</param>
 173    /// <param name="action">Action to be executed when watch event is triggered</param>
 174    /// <param name="headers">The initial metadata to send with the call</param>
 175    /// <param name="deadline">An optional deadline for the call</param>
 176    /// <param name="cancellationToken">An optional token for canceling the call</param>
 177    /// <returns>Watch ID</returns>
 178    public long Watch(string key, Action<WatchEvent> action, Metadata? headers = null, DateTime? deadline = null,
 179        CancellationToken cancellationToken = default)
 2180    {
 2181        ObjectDisposedException.ThrowIf(_disposed, this);
 182
 183        // Create a watch request for the key
 1184        WatchRequest request = new()
 1185        {
 1186            CreateRequest = new WatchCreateRequest
 1187            {
 1188                Key = ByteString.CopyFromUtf8(key), ProgressNotify = true, PrevKv = true
 1189            }
 1190        };
 191
 192        // Call the Watch method with the request
 1193        return Watch(request, Callback, headers, deadline, cancellationToken);
 194
 195        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 196        void Callback(WatchResponse response)
 1197        {
 1198            if (response.Events == null)
 0199            {
 0200                return;
 201            }
 202
 5203            foreach (Event evt in response.Events)
 1204            {
 1205                WatchEvent watchEvent = new() { Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Ty
 1206                action(watchEvent);
 1207            }
 1208        }
 1209    }
 210
 211    /// <summary>
 212    ///     Watches a range of keys with a prefix
 213    /// </summary>
 214    /// <param name="prefixKey">Prefix key to watch</param>
 215    /// <param name="action">Action to be executed when watch event is triggered</param>
 216    /// <param name="headers">The initial metadata to send with the call</param>
 217    /// <param name="deadline">An optional deadline for the call</param>
 218    /// <param name="cancellationToken">An optional token for canceling the call</param>
 219    /// <returns>Watch ID</returns>
 220    public long WatchRange(string prefixKey, Action<WatchEvent> action, Metadata? headers = null,
 221        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1222    {
 1223        ObjectDisposedException.ThrowIf(_disposed, this);
 224
 225        // Create a watch request for the range
 1226        WatchRequest request = new()
 1227        {
 1228            CreateRequest = new WatchCreateRequest
 1229            {
 1230                Key = ByteString.CopyFromUtf8(prefixKey),
 1231                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1232                ProgressNotify = true,
 1233                PrevKv = true
 1234            }
 1235        };
 236
 237        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1238        Action<WatchResponse> callback = response =>
 1239        {
 1240            if (response.Events == null)
 0241            {
 0242                return;
 1243            }
 1244
 5245            foreach (Event evt in response.Events)
 1246            {
 1247                WatchEvent watchEvent = new()
 1248                {
 1249                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1250                };
 1251                action(watchEvent);
 1252            }
 2253        };
 254
 255        // Call the Watch method with the request
 1256        return Watch(request, callback, headers, deadline, cancellationToken);
 1257    }
 258
 259    /// <summary>
 260    ///     Watches a specific key with start revision
 261    /// </summary>
 262    /// <param name="key">Key to watch</param>
 263    /// <param name="startRevision">Start revision</param>
 264    /// <param name="action">Action to be executed when watch event is triggered</param>
 265    /// <param name="headers">The initial metadata to send with the call</param>
 266    /// <param name="deadline">An optional deadline for the call</param>
 267    /// <param name="cancellationToken">An optional token for canceling the call</param>
 268    /// <returns>Watch ID</returns>
 269    public long Watch(string key, long startRevision, Action<WatchEvent> action, Metadata? headers = null,
 270        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1271    {
 1272        ObjectDisposedException.ThrowIf(_disposed, this);
 273
 274        // Create a watch request for the key with start revision
 1275        WatchRequest request = new()
 1276        {
 1277            CreateRequest = new WatchCreateRequest
 1278            {
 1279                Key = ByteString.CopyFromUtf8(key),
 1280                StartRevision = startRevision,
 1281                ProgressNotify = true,
 1282                PrevKv = true
 1283            }
 1284        };
 285
 286        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1287        Action<WatchResponse> callback = response =>
 1288        {
 1289            if (response.Events == null)
 0290            {
 0291                return;
 1292            }
 1293
 5294            foreach (Event evt in response.Events)
 1295            {
 1296                WatchEvent watchEvent = new()
 1297                {
 1298                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1299                };
 1300                action(watchEvent);
 1301            }
 2302        };
 303
 304        // Call the Watch method with the request
 1305        return Watch(request, callback, headers, deadline, cancellationToken);
 1306    }
 307
 308    /// <summary>
 309    ///     Watches a range of keys with a prefix and start revision
 310    /// </summary>
 311    /// <param name="prefixKey">Prefix key to watch</param>
 312    /// <param name="startRevision">Start revision</param>
 313    /// <param name="action">Action to be executed when watch event is triggered</param>
 314    /// <param name="headers">The initial metadata to send with the call</param>
 315    /// <param name="deadline">An optional deadline for the call</param>
 316    /// <param name="cancellationToken">An optional token for canceling the call</param>
 317    /// <returns>Watch ID</returns>
 318    public long WatchRange(string prefixKey, long startRevision, Action<WatchEvent> action, Metadata? headers = null,
 319        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1320    {
 1321        ObjectDisposedException.ThrowIf(_disposed, this);
 322
 323        // Create a watch request for the range with start revision
 1324        WatchRequest request = new()
 1325        {
 1326            CreateRequest = new WatchCreateRequest
 1327            {
 1328                Key = ByteString.CopyFromUtf8(prefixKey),
 1329                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1330                StartRevision = startRevision,
 1331                ProgressNotify = true,
 1332                PrevKv = true
 1333            }
 1334        };
 335
 336        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1337        Action<WatchResponse> callback = response =>
 1338        {
 1339            if (response.Events == null)
 0340            {
 0341                return;
 1342            }
 1343
 5344            foreach (Event evt in response.Events)
 1345            {
 1346                WatchEvent watchEvent = new()
 1347                {
 1348                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1349                };
 1350                action(watchEvent);
 1351            }
 2352        };
 353
 354        // Call the Watch method with the request
 1355        return Watch(request, callback, headers, deadline, cancellationToken);
 1356    }
 357
 358    /// <summary>
 359    ///     Watches a specific key asynchronously
 360    /// </summary>
 361    /// <param name="key">Key to watch</param>
 362    /// <param name="action">Action to be executed when watch event is triggered</param>
 363    /// <param name="headers">The initial metadata to send with the call</param>
 364    /// <param name="deadline">An optional deadline for the call</param>
 365    /// <param name="cancellationToken">An optional token for canceling the call</param>
 366    /// <returns>Watch ID</returns>
 367    public async Task<long> WatchAsync(string key, Action<WatchEvent> action, Metadata? headers = null,
 368        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1369    {
 1370        ObjectDisposedException.ThrowIf(_disposed, this);
 371
 372        // Create a watch request for the key
 1373        WatchRequest request = new()
 1374        {
 1375            CreateRequest = new WatchCreateRequest
 1376            {
 1377                Key = ByteString.CopyFromUtf8(key), ProgressNotify = true, PrevKv = true
 1378            }
 1379        };
 380
 381        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1382        Action<WatchResponse> callback = response =>
 1383        {
 1384            if (response.Events == null)
 0385            {
 0386                return;
 1387            }
 1388
 5389            foreach (Event evt in response.Events)
 1390            {
 1391                WatchEvent watchEvent = new()
 1392                {
 1393                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1394                };
 1395                action(watchEvent);
 1396            }
 2397        };
 398
 399        // Call the WatchAsync method with the request
 1400        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1401    }
 402
 403    /// <summary>
 404    ///     Watches a range of keys with a prefix asynchronously
 405    /// </summary>
 406    /// <param name="prefixKey">Prefix key to watch</param>
 407    /// <param name="action">Action to be executed when watch event is triggered</param>
 408    /// <param name="headers">The initial metadata to send with the call</param>
 409    /// <param name="deadline">An optional deadline for the call</param>
 410    /// <param name="cancellationToken">An optional token for canceling the call</param>
 411    /// <returns>Watch ID</returns>
 412    public async Task<long> WatchRangeAsync(string prefixKey, Action<WatchEvent> action, Metadata? headers = null,
 413        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1414    {
 1415        ObjectDisposedException.ThrowIf(_disposed, this);
 416
 417        // Create a watch request for the range
 1418        WatchRequest request = new()
 1419        {
 1420            CreateRequest = new WatchCreateRequest
 1421            {
 1422                Key = ByteString.CopyFromUtf8(prefixKey),
 1423                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1424                ProgressNotify = true,
 1425                PrevKv = true
 1426            }
 1427        };
 428
 429        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1430        Action<WatchResponse> callback = response =>
 1431        {
 1432            if (response.Events == null)
 0433            {
 0434                return;
 1435            }
 1436
 5437            foreach (Event evt in response.Events)
 1438            {
 1439                WatchEvent watchEvent = new()
 1440                {
 1441                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1442                };
 1443                action(watchEvent);
 1444            }
 2445        };
 446
 447        // Call the WatchAsync method with the request
 1448        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1449    }
 450
 451    /// <summary>
 452    ///     Watches a specific key with start revision asynchronously
 453    /// </summary>
 454    /// <param name="key">Key to watch</param>
 455    /// <param name="startRevision">Start revision</param>
 456    /// <param name="action">Action to be executed when watch event is triggered</param>
 457    /// <param name="headers">The initial metadata to send with the call</param>
 458    /// <param name="deadline">An optional deadline for the call</param>
 459    /// <param name="cancellationToken">An optional token for canceling the call</param>
 460    /// <returns>Watch ID</returns>
 461    public async Task<long> WatchAsync(string key, long startRevision, Action<WatchEvent> action,
 462        Metadata? headers = null, DateTime? deadline = null, CancellationToken cancellationToken = default)
 1463    {
 1464        ObjectDisposedException.ThrowIf(_disposed, this);
 465
 466        // Create a watch request for the key with start revision
 1467        WatchRequest request = new()
 1468        {
 1469            CreateRequest = new WatchCreateRequest
 1470            {
 1471                Key = ByteString.CopyFromUtf8(key),
 1472                StartRevision = startRevision,
 1473                ProgressNotify = true,
 1474                PrevKv = true
 1475            }
 1476        };
 477
 478        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1479        Action<WatchResponse> callback = response =>
 1480        {
 1481            if (response.Events == null)
 0482            {
 0483                return;
 1484            }
 1485
 5486            foreach (Event evt in response.Events)
 1487            {
 1488                WatchEvent watchEvent = new()
 1489                {
 1490                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1491                };
 1492                action(watchEvent);
 1493            }
 2494        };
 495
 496        // Call the WatchAsync method with the request
 1497        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1498    }
 499
 500    /// <summary>
 501    ///     Watches a range of keys with a prefix and start revision asynchronously
 502    /// </summary>
 503    /// <param name="prefixKey">Prefix key to watch</param>
 504    /// <param name="startRevision">Start revision</param>
 505    /// <param name="action">Action to be executed when watch event is triggered</param>
 506    /// <param name="headers">The initial metadata to send with the call</param>
 507    /// <param name="deadline">An optional deadline for the call</param>
 508    /// <param name="cancellationToken">An optional token for canceling the call</param>
 509    /// <returns>Watch ID</returns>
 510    public async Task<long> WatchRangeAsync(string prefixKey, long startRevision, Action<WatchEvent> action,
 511        Metadata? headers = null, DateTime? deadline = null, CancellationToken cancellationToken = default)
 1512    {
 1513        ObjectDisposedException.ThrowIf(_disposed, this);
 514
 515        // Create a watch request for the range with start revision
 1516        WatchRequest request = new()
 1517        {
 1518            CreateRequest = new WatchCreateRequest
 1519            {
 1520                Key = ByteString.CopyFromUtf8(prefixKey),
 1521                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1522                StartRevision = startRevision,
 1523                ProgressNotify = true,
 1524                PrevKv = true
 1525            }
 1526        };
 527
 528        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1529        Action<WatchResponse> callback = response =>
 1530        {
 1531            if (response.Events == null)
 0532            {
 0533                return;
 1534            }
 1535
 5536            foreach (Event evt in response.Events)
 1537            {
 1538                WatchEvent watchEvent = new()
 1539                {
 1540                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1541                };
 1542                action(watchEvent);
 1543            }
 2544        };
 545
 546        // Call the WatchAsync method with the request
 1547        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1548    }
 549
 550    /// <summary>
 551    ///     Watches a key range
 552    /// </summary>
 553    /// <param name="path">The path to watch</param>
 554    /// <param name="callback">The callback to invoke when a watch event is received</param>
 555    /// <param name="headers">The initial metadata to send with the call</param>
 556    /// <param name="deadline">An optional deadline for the call</param>
 557    /// <param name="cancellationToken">An optional token for canceling the call</param>
 558    /// <returns>A watch ID that can be used to cancel the watch</returns>
 559    public long WatchRange(string path, Action<WatchResponse> callback, Metadata? headers = null,
 560        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1561    {
 1562        ObjectDisposedException.ThrowIf(_disposed, this);
 563
 564        // Create a watch request for the range
 1565        WatchRequest request = new()
 1566        {
 1567            CreateRequest = new WatchCreateRequest
 1568            {
 1569                Key = ByteString.CopyFromUtf8(path),
 1570                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(path)),
 1571                ProgressNotify = true,
 1572                PrevKv = true
 1573            }
 1574        };
 575
 576        // Call the Watch method with the request
 1577        return Watch(request, callback, headers, deadline, cancellationToken);
 1578    }
 579
 580    /// <summary>
 581    ///     Cancels a watch request
 582    /// </summary>
 583    /// <param name="watchId">The ID of the watch to cancel</param>
 584    public void CancelWatch(long watchId)
 2585    {
 2586        ObjectDisposedException.ThrowIf(_disposed, this);
 587
 2588        if (!_watches.TryRemove(watchId, out WatchCancellation? watchCancellation))
 1589        {
 1590            return;
 591        }
 592
 593        // Cancel the watch
 1594        watchCancellation.CancellationTokenSource.Cancel();
 595
 596        // Find the server watch ID that corresponds to our client watch ID
 1597        long serverWatchId = GetServerWatchId(watchId);
 598
 599        // Cancel the watch on the server if we found a mapping
 1600        if (serverWatchId != -1 && _watchStream != null)
 1601        {
 1602            _watchIdMapping.TryRemove(serverWatchId, out _);
 1603            _watchStream.CancelWatchAsync(serverWatchId).ContinueWith(_ =>
 1604            {
 1605                // Ignore exceptions
 2606            });
 1607        }
 608
 609        // Dispose the cancellation token source
 1610        watchCancellation.CancellationTokenSource.Dispose();
 2611    }
 612
 613    /// <summary>
 614    ///     Disposes the watch manager
 615    /// </summary>
 616    public void Dispose()
 103617    {
 103618        if (_disposed)
 1619        {
 1620            return;
 621        }
 622
 102623        _disposed = true;
 624
 625        // Cancel all watches
 356626        foreach (WatchCancellation watchCancellation in _watches.Values)
 25627        {
 25628            watchCancellation.CancellationTokenSource.Cancel();
 25629            watchCancellation.CancellationTokenSource.Dispose();
 25630        }
 631
 102632        _watches.Clear();
 633
 634        // Dispose the watch stream
 102635        if (_watchStream != null)
 25636        {
 25637            _watchStream.Dispose();
 25638            _watchStream = null;
 25639        }
 640
 102641        GC.SuppressFinalize(this);
 103642    }
 643
 644    private static string GetRangeEnd(string path)
 5645    {
 646        // Calculate the range end for the given path
 647        // This is the same logic used in EtcdClient.GetRangeEnd
 5648        byte[] bytes = Encoding.UTF8.GetBytes(path);
 10649        for (int i = bytes.Length - 1; i >= 0; i--)
 5650        {
 5651            if (bytes[i] >= 0xff)
 0652            {
 0653                continue;
 654            }
 655
 5656            bytes[i]++;
 5657            return Encoding.UTF8.GetString(bytes, 0, i + 1);
 658        }
 659
 0660        return string.Empty;
 5661    }
 662
 663    /// <summary>
 664    ///     Gets the server watch ID for a client watch ID
 665    /// </summary>
 666    /// <param name="clientWatchId">The client watch ID</param>
 667    /// <returns>The server watch ID, or -1 if not found</returns>
 668    private long GetServerWatchId(long clientWatchId)
 1669    {
 4670        foreach (KeyValuePair<long, long> kvp in _watchIdMapping)
 1671        {
 1672            if (kvp.Value == clientWatchId)
 1673            {
 1674                return kvp.Key;
 675            }
 0676        }
 677
 0678        return -1;
 1679    }
 680
 681    /// <param name="cancellationToken">An optional token for canceling the call</param>
 682    private void EnsureWatchStream(Metadata? headers, DateTime? deadline, CancellationToken cancellationToken)
 68683    {
 68684        lock (_lockObject)
 68685        {
 68686            if (_watchStream != null)
 13687            {
 13688                return;
 689            }
 690
 691            // Create a new watch stream
 55692            IAsyncDuplexStreamingCall<WatchRequest, WatchResponse> watchStreamCall =
 55693                _watchStreamFactory(headers, deadline, cancellationToken);
 55694            _watchStream = new Watcher(watchStreamCall, HandleConnectionFailure);
 55695        }
 68696    }
 697
 698    private void HandleConnectionFailure()
 4699    {
 4700        lock (_lockObject)
 4701        {
 4702            _watchStream = null;
 4703            _watchIdMapping.Clear();
 4704        }
 705
 706        // Must run async to avoid blocking the caller (which might be the dead stream loop)
 4707        Task.Run(async () =>
 4708        {
 4709            try
 4710            {
 4711                // Wait small delay to allow network to stabilize
 4712                await Task.Delay(500);
 4713
 4714                lock (_lockObject)
 4715                {
 4716                   if (_disposed) return;
 4717                   EnsureWatchStream(null, null, default);
 4718                }
 4719
 20720                foreach (var watch in _watches.Values)
 4721                {
 4722                    // Resume from the revision after the last observed event so events written while
 4723                    // the stream was down are replayed instead of lost. Falls back to "from now"
 4724                    // (StartRevision 0) only when nothing has been observed yet.
 4725                    if (watch.NextRevision > 0 && watch.Request.CreateRequest != null)
 4726                    {
 4727                        watch.Request.CreateRequest.StartRevision = watch.NextRevision;
 4728                    }
 4729
 4730                    await _watchStream!.CreateWatchAsync(watch.Request, watch.Callback).ConfigureAwait(false);
 4731                }
 4732            }
 0733            catch (Exception ex)
 0734            {
 0735                Console.Error.WriteLine($"Watch reconnection failed: {ex.Message}");
 4736                // Retry in 5s
 0737                _ = Task.Delay(5000).ContinueWith(_ => HandleConnectionFailure());
 0738            }
 8739        });
 4740    }
 741
 742    private class WatchCancellation
 743    {
 64744        public long WatchId { get; set; }
 116745        public required CancellationTokenSource CancellationTokenSource { get; set; }
 76746        public required WatchRequest Request { get; set; }
 68747        public required Action<WatchResponse> Callback { get; set; }
 748
 749        /// <summary>
 750        ///     The revision to resume this watch from if the stream is re-established. Tracks the
 751        ///     revision after the last event/notification observed, so a reconnect does not miss
 752        ///     events written during the gap (mirrors the etcd clientv3 nextRev behavior).
 753        /// </summary>
 162754        public long NextRevision { get; set; }
 755    }
 756}
Generated by: ReportGenerator 5.5.10.0
07/04/2026 - 18:14:37
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Func`4<Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken,dotnet_etcd.interfaces.IAsyncDuplexStreamingCall`2<Etcdserverpb.WatchRequest,Etcdserverpb.WatchResponse>>)
WatchAsync()
WrappedCallback()
TrackResumeRevision(System.Int64,Etcdserverpb.WatchResponse)
Watch(Etcdserverpb.WatchRequest,System.Action`1<Etcdserverpb.WatchResponse>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
Watch(System.String,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
Callback()
WatchRange(System.String,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
Watch(System.String,System.Int64,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
WatchRange(System.String,System.Int64,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
WatchAsync()
WatchRangeAsync()
WatchAsync()
WatchRangeAsync()
WatchRange(System.String,System.Action`1<Etcdserverpb.WatchResponse>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
CancelWatch(System.Int64)
Dispose()
GetRangeEnd(System.String)
GetServerWatchId(System.Int64)
EnsureWatchStream(Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
HandleConnectionFailure()
get_WatchId()
get_CancellationTokenSource()
get_Request()
get_Callback()
get_NextRevision()