This repository was archived by the owner on Feb 16, 2019. It is now read-only.
Repository navigation
Expand file tree
/
Copy pathEventStorePersistentSubscription.cs
More file actions
79 lines (69 loc) · 3.17 KB
/
Copy pathEventStorePersistentSubscription.cs
File metadata and controls
79 lines (69 loc) · 3.17 KB
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
using System;
using System.Threading.Tasks;
using EventStore.ClientAPI.ClientOperations;
using EventStore.ClientAPI.Internal;
using EventStore.ClientAPI.SystemData;
namespace EventStore.ClientAPI
{
internal struct PersistentSubscriptionResolvedEvent : IResolvedEvent
{
/// <summary>
///
/// </summary>
public readonly int? RetryCount;
/// <summary>
///
/// </summary>
public readonly ResolvedEvent Event;
/// <summary>
///
/// </summary>
/// <param name="event"></param>
/// <param name="retryCount"></param>
internal PersistentSubscriptionResolvedEvent(ResolvedEvent @event, int? retryCount)
{
Event = @event;
RetryCount = retryCount;
}
string IResolvedEvent.OriginalStreamId => Event.OriginalStreamId;
long IResolvedEvent.OriginalEventNumber => Event.OriginalEventNumber;
RecordedEvent IResolvedEvent.OriginalEvent => Event.OriginalEvent;
Position? IResolvedEvent.OriginalPosition => Event.OriginalPosition;
/// <summary>
///
/// </summary>
/// <param name="x"></param>
/// <returns></returns>
public static implicit operator ResolvedEvent(PersistentSubscriptionResolvedEvent x) => x.Event;
}
/// <summary>
/// Represents a persistent subscription connection.
/// </summary>
public class EventStorePersistentSubscription : EventStorePersistentSubscriptionBase
{
private readonly EventStoreConnectionLogicHandler _handler;
internal EventStorePersistentSubscription(
string subscriptionId, string streamId,
Func<EventStorePersistentSubscriptionBase, ResolvedEvent, int?, Task> eventAppeared,
Action<EventStorePersistentSubscriptionBase, SubscriptionDropReason, Exception> subscriptionDropped,
UserCredentials userCredentials, ILogger log, bool verboseLogging, ConnectionSettings settings,
EventStoreConnectionLogicHandler handler, int bufferSize = 10, bool autoAck = true)
: base(
subscriptionId, streamId, eventAppeared, subscriptionDropped, userCredentials, log, verboseLogging,
settings, bufferSize, autoAck)
{
_handler = handler;
}
internal override Task<PersistentEventStoreSubscription> StartSubscription(
string subscriptionId, string streamId, int bufferSize, UserCredentials userCredentials,
Func<EventStoreSubscription, PersistentSubscriptionResolvedEvent, Task> onEventAppeared,
Action<EventStoreSubscription, SubscriptionDropReason, Exception> onSubscriptionDropped, ConnectionSettings settings)
{
var source = new TaskCompletionSource<PersistentEventStoreSubscription>(TaskCreationOptions.RunContinuationsAsynchronously);
_handler.EnqueueMessage(new StartPersistentSubscriptionMessage(source, subscriptionId, streamId, bufferSize,
userCredentials, onEventAppeared,
onSubscriptionDropped, settings.MaxRetries, settings.OperationTimeout));
return source.Task;
}
}
}