-
Notifications
You must be signed in to change notification settings - Fork 476
/
PartitionSupervisorCore.cs
93 lines (83 loc) · 3.61 KB
/
PartitionSupervisorCore.cs
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
85
86
87
88
89
90
91
92
93
//------------------------------------------------------------
// Copyright (c) Microsoft Corporation. All rights reserved.
//------------------------------------------------------------
namespace Microsoft.Azure.Cosmos.ChangeFeed.FeedManagement
{
using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Cosmos.ChangeFeed.Exceptions;
using Microsoft.Azure.Cosmos.ChangeFeed.FeedProcessing;
using Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement;
using Microsoft.Azure.Cosmos.ChangeFeed.Utils;
internal sealed class PartitionSupervisorCore : PartitionSupervisor
{
private readonly DocumentServiceLease lease;
private readonly ChangeFeedObserver observer;
private readonly FeedProcessor processor;
private readonly LeaseRenewer renewer;
private readonly CancellationTokenSource renewerCancellation = new CancellationTokenSource();
private CancellationTokenSource processorCancellation;
public PartitionSupervisorCore(DocumentServiceLease lease, ChangeFeedObserver observer, FeedProcessor processor, LeaseRenewer renewer)
{
this.lease = lease;
this.observer = observer;
this.processor = processor;
this.renewer = renewer;
}
public override async Task RunAsync(CancellationToken shutdownToken)
{
await this.observer.OpenAsync(this.lease.CurrentLeaseToken).ConfigureAwait(false);
this.processorCancellation = CancellationTokenSource.CreateLinkedTokenSource(shutdownToken);
Task processorTask = this.processor.RunAsync(this.processorCancellation.Token);
processorTask.ContinueWith(_ => this.renewerCancellation.Cancel()).LogException();
Task renewerTask = this.renewer.RunAsync(this.renewerCancellation.Token);
renewerTask.ContinueWith(_ => this.processorCancellation.Cancel()).LogException();
ChangeFeedObserverCloseReason closeReason = shutdownToken.IsCancellationRequested ?
ChangeFeedObserverCloseReason.Shutdown :
ChangeFeedObserverCloseReason.Unknown;
try
{
await Task.WhenAll(processorTask, renewerTask).ConfigureAwait(false);
}
catch (LeaseLostException)
{
closeReason = ChangeFeedObserverCloseReason.LeaseLost;
throw;
}
catch (FeedRangeGoneException)
{
closeReason = ChangeFeedObserverCloseReason.LeaseGone;
throw;
}
catch (CosmosException)
{
closeReason = ChangeFeedObserverCloseReason.CosmosException;
throw;
}
catch (OperationCanceledException) when (shutdownToken.IsCancellationRequested)
{
closeReason = ChangeFeedObserverCloseReason.Shutdown;
}
catch (ChangeFeedProcessorUserException)
{
closeReason = ChangeFeedObserverCloseReason.ObserverError;
throw;
}
catch (Exception) when (processorTask.IsFaulted)
{
closeReason = ChangeFeedObserverCloseReason.Unknown;
throw;
}
finally
{
await this.observer.CloseAsync(this.lease.CurrentLeaseToken, closeReason).ConfigureAwait(false);
}
}
public override void Dispose()
{
this.processorCancellation?.Dispose();
this.renewerCancellation.Dispose();
}
}
}