Skip to content

Commit 29b8c62

Browse files
jagealljageall
committed
Added ConditionalAppendToStream API
1 parent e4efbc4 commit 29b8c62

7 files changed

Lines changed: 251 additions & 0 deletions

File tree

‎src/EventStore.ClientAPI.NetCore/ClientOperations/ConditionalAppendToStreamOperation.cs‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq;
4+
using System.Threading.Tasks;
5+
using EventStore.ClientAPI.Exceptions;
6+
using EventStore.ClientAPI.Messages;
7+
using EventStore.ClientAPI.SystemData;
8+
9+
namespace EventStore.ClientAPI.ClientOperations
10+
{
11+
internal class ConditionalAppendToStreamOperation : OperationBase<ConditionalWriteResult, ClientMessage.WriteEventsCompleted>
12+
{
13+
private readonly bool _requireMaster;
14+
private readonly string _stream;
15+
private readonly int _expectedVersion;
16+
private readonly IEnumerable<EventData> _events;
17+
18+
private bool _wasCommitTimeout;
19+
20+
public ConditionalAppendToStreamOperation(ILogger log,
21+
TaskCompletionSource<ConditionalWriteResult> source,
22+
bool requireMaster,
23+
string stream,
24+
int expectedVersion,
25+
IEnumerable<EventData> events,
26+
UserCredentials userCredentials)
27+
: base(log, source, TcpCommand.WriteEvents, TcpCommand.WriteEventsCompleted, userCredentials)
28+
{
29+
_requireMaster = requireMaster;
30+
_stream = stream;
31+
_expectedVersion = expectedVersion;
32+
_events = events;
33+
}
34+
35+
protected override object CreateRequestDto()
36+
{
37+
var dtos = _events.Select(x => new ClientMessage.NewEvent(x.EventId.ToByteArray(), x.Type, x.IsJson ? 1 : 0, 0, x.Data, x.Metadata)).ToArray();
38+
return new ClientMessage.WriteEvents(_stream, _expectedVersion, dtos, _requireMaster);
39+
}
40+
41+
protected override InspectionResult InspectResponse(ClientMessage.WriteEventsCompleted response)
42+
{
43+
switch (response.Result)
44+
{
45+
case ClientMessage.OperationResult.Success:
46+
if (_wasCommitTimeout)
47+
Log.Debug("IDEMPOTENT WRITE SUCCEEDED FOR {0}.", this);
48+
Succeed();
49+
return new InspectionResult(InspectionDecision.EndOperation, "Success");
50+
case ClientMessage.OperationResult.PrepareTimeout:
51+
return new InspectionResult(InspectionDecision.Retry, "PrepareTimeout");
52+
case ClientMessage.OperationResult.ForwardTimeout:
53+
return new InspectionResult(InspectionDecision.Retry, "ForwardTimeout");
54+
case ClientMessage.OperationResult.CommitTimeout:
55+
_wasCommitTimeout = true;
56+
return new InspectionResult(InspectionDecision.Retry, "CommitTimeout");
57+
case ClientMessage.OperationResult.WrongExpectedVersion:
58+
Succeed();
59+
return new InspectionResult(InspectionDecision.EndOperation, "ExpectedVersionMismatch");
60+
case ClientMessage.OperationResult.StreamDeleted:
61+
Succeed();
62+
return new InspectionResult(InspectionDecision.EndOperation, "StreamDeleted");
63+
case ClientMessage.OperationResult.InvalidTransaction:
64+
Fail(new InvalidTransactionException());
65+
return new InspectionResult(InspectionDecision.EndOperation, "InvalidTransaction");
66+
case ClientMessage.OperationResult.AccessDenied:
67+
Fail(new AccessDeniedException(string.Format("Write access denied for stream '{0}'.", _stream)));
68+
return new InspectionResult(InspectionDecision.EndOperation, "AccessDenied");
69+
default:
70+
throw new Exception(string.Format("Unexpected OperationResult: {0}.", response.Result));
71+
}
72+
}
73+
74+
protected override ConditionalWriteResult TransformResponse(ClientMessage.WriteEventsCompleted response)
75+
{
76+
if (response.Result == ClientMessage.OperationResult.WrongExpectedVersion)
77+
{
78+
return new ConditionalWriteResult(ConditionalWriteStatus.VersionMismatch);
79+
}
80+
if (response.Result == ClientMessage.OperationResult.StreamDeleted)
81+
{
82+
return new ConditionalWriteResult(ConditionalWriteStatus.StreamDeleted);
83+
}
84+
return new ConditionalWriteResult(response.LastEventNumber, new Position(response.PreparePosition ?? -1, response.CommitPosition ?? -1));
85+
}
86+
87+
public override string ToString()
88+
{
89+
return string.Format("Stream: {0}, ExpectedVersion: {1}", _stream, _expectedVersion);
90+
}
91+
}
92+
}

‎src/EventStore.ClientAPI.NetCore/ConditionalWriteFailureReason.cs‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
namespace EventStore.ClientAPI
2+
{
3+
/// <summary>
4+
/// The reason why a conditional write fails
5+
/// </summary>
6+
public enum ConditionalWriteStatus
7+
{
8+
/// <summary>
9+
/// The write operation succeeded
10+
/// </summary>
11+
Succeeded = 0,
12+
/// <summary>
13+
/// The expected version does not match actual stream version
14+
/// </summary>
15+
VersionMismatch = 1,
16+
/// <summary>
17+
/// The stream has been deleted
18+
/// </summary>
19+
StreamDeleted = 2
20+
}
21+
}

‎src/EventStore.ClientAPI.NetCore/ConditionalWriteResult.cs‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
using System;
2+
3+
namespace EventStore.ClientAPI
4+
{
5+
/// <summary>
6+
/// Result type returned after conditionally writing to a stream.
7+
/// </summary>
8+
public struct ConditionalWriteResult
9+
{
10+
/// <summary>
11+
/// Returns if the write was successful.
12+
/// </summary>
13+
public readonly ConditionalWriteStatus Status;
14+
15+
/// <summary>
16+
/// The next expected version for the stream.
17+
/// </summary>
18+
public readonly int? NextExpectedVersion;
19+
20+
/// <summary>
21+
/// The <see cref="LogPosition"/> of the write.
22+
/// </summary>
23+
public readonly Position? LogPosition;
24+
25+
/// <summary>
26+
/// Constructs a new <see cref="WriteResult"/>.
27+
/// </summary>
28+
/// <param name="nextExpectedVersion">The next expected version for the stream.</param>
29+
/// <param name="logPosition">The position of the write in the log</param>
30+
public ConditionalWriteResult(int nextExpectedVersion, Position logPosition)
31+
{
32+
LogPosition = logPosition;
33+
NextExpectedVersion = nextExpectedVersion;
34+
Status = ConditionalWriteStatus.Succeeded;
35+
}
36+
37+
/// <summary>
38+
/// Constructs a new <see cref="WriteResult"/>.
39+
/// </summary>
40+
/// <param name="status">The status of the write operation.</param>
41+
public ConditionalWriteResult(ConditionalWriteStatus status)
42+
{
43+
if (status == ConditionalWriteStatus.Succeeded)
44+
{
45+
throw new Exception("For successful write pass next expected version and log position.");
46+
}
47+
LogPosition = null;
48+
NextExpectedVersion = null;
49+
Status = status;
50+
}
51+
}
52+
}

‎src/EventStore.ClientAPI.NetCore/IEventStoreConnection.cs‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,26 @@ public interface IEventStoreConnection : IDisposable
115115
/// <param name="userCredentials">The optional user credentials to perform operation with.</param>
116116
Task<WriteResult> AppendToStreamAsync(string stream, int expectedVersion, IEnumerable<EventData> events, UserCredentials userCredentials = null);
117117

118+
/// <summary>
119+
/// Appends Events asynchronously to a stream if the stream version matches the <paramref name="expectedVersion"/>.
120+
/// </summary>
121+
/// <remarks>
122+
/// When appending events to a stream the <see cref="ExpectedVersion"/> choice can
123+
/// make a very large difference in the observed behavior. For example, if no stream exists
124+
/// and ExpectedVersion.Any is used, a new stream will be implicitly created when appending.
125+
///
126+
/// There are also differences in idempotency between different types of calls.
127+
/// If you specify an ExpectedVersion aside from ExpectedVersion.Any the Event Store
128+
/// will give you an idempotency guarantee. If using ExpectedVersion.Any the Event Store
129+
/// will do its best to provide idempotency but does not guarantee idempotency
130+
/// </remarks>
131+
/// <param name="stream">The name of the stream to append events to</param>
132+
/// <param name="expectedVersion">The <see cref="ExpectedVersion"/> of the stream to append to</param>
133+
/// <param name="events">The events to append to the stream</param>
134+
/// <param name="userCredentials">The optional user credentials to perform operation with.</param>
135+
/// <returns>If the operation succeeded and, if not, the reason for failure (which can be either stream version mismatch or trying to write to a deleted stream)</returns>
136+
Task<ConditionalWriteResult> ConditionalAppendToStreamAsync(string stream, int expectedVersion, IEnumerable<EventData> events, UserCredentials userCredentials = null);
137+
118138
/// <summary>
119139
/// Starts a transaction in the event store on a given stream asynchronously
120140
/// </summary>

‎src/EventStore.ClientAPI.NetCore/Internal/EventStoreNodeConnection.cs‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,20 @@ public Task<WriteResult> AppendToStreamAsync(string stream, int expectedVersion,
129129
// ReSharper restore PossibleMultipleEnumeration
130130
}
131131

132+
public Task<ConditionalWriteResult> ConditionalAppendToStreamAsync(string stream, int expectedVersion, IEnumerable<EventData> events,
133+
UserCredentials userCredentials = null)
134+
{
135+
// ReSharper disable PossibleMultipleEnumeration
136+
Ensure.NotNullOrEmpty(stream, "stream");
137+
Ensure.NotNull(events, "events");
138+
139+
var source = new TaskCompletionSource<ConditionalWriteResult>();
140+
EnqueueOperation(new ConditionalAppendToStreamOperation(_settings.Log, source, _settings.RequireMaster,
141+
stream, expectedVersion, events, userCredentials));
142+
return source.Task;
143+
// ReSharper restore PossibleMultipleEnumeration
144+
}
145+
132146
public Task<EventStoreTransaction> StartTransactionAsync(string stream, int expectedVersion, UserCredentials userCredentials = null)
133147
{
134148
Ensure.NotNullOrEmpty(stream, "stream");

‎test/EventStore.ClientAPI.NetCore.Tests/FakeEventStoreConnection.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -256,5 +256,10 @@ public Task<EventStorePersistentSubscriptionBase> ConnectToPersistentSubscriptio
256256
{
257257
throw new NotImplementedException();
258258
}
259+
260+
public Task<ConditionalWriteResult> ConditionalAppendToStreamAsync(string stream, int expectedVersion, IEnumerable<EventData> events, UserCredentials userCredentials = null)
261+
{
262+
throw new NotImplementedException();
263+
}
259264
}
260265
}

‎test/EventStore.ClientAPI.NetCore.Tests/append_to_stream.cs‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -359,5 +359,52 @@ public void can_append_multiple_events_at_once()
359359
Assert.AreEqual(99, store.AppendToStreamAsync(stream, ExpectedVersion.EmptyStream, events).Result.NextExpectedVersion);
360360
}
361361
}
362+
363+
[Test, Category("Network")]
364+
public void returns_failure_status_when_conditionally_appending_with_version_mismatch()
365+
{
366+
const string stream = "returns_failure_status_when_conditionally_appending_with_version_mismatch";
367+
using (var store = BuildConnection())
368+
{
369+
store.ConnectAsync().Wait();
370+
371+
var result = store.ConditionalAppendToStreamAsync(stream, 7, new[] {TestEvent.NewTestEvent()}).Result;
372+
373+
Assert.AreEqual(ConditionalWriteStatus.VersionMismatch, result.Status);
374+
}
375+
}
376+
377+
[Test, Category("Network")]
378+
public void returns_success_status_when_conditionally_appending_with_matching_version()
379+
{
380+
const string stream = "returns_success_status_when_conditionally_appending_with_matching_version";
381+
using (var store = BuildConnection())
382+
{
383+
store.ConnectAsync().Wait();
384+
385+
var result = store.ConditionalAppendToStreamAsync(stream, ExpectedVersion.Any, new[] { TestEvent.NewTestEvent() }).Result;
386+
387+
Assert.AreEqual(ConditionalWriteStatus.Succeeded, result.Status);
388+
Assert.IsNotNull(result.LogPosition);
389+
Assert.IsNotNull(result.NextExpectedVersion);
390+
}
391+
}
392+
393+
[Test, Category("Network")]
394+
public void returns_failure_status_when_conditionally_appending_to_a_deleted_stream()
395+
{
396+
const string stream = "returns_failure_status_when_conditionally_appending_to_a_deleted_stream";
397+
using (var store = BuildConnection())
398+
{
399+
store.ConnectAsync().Wait();
400+
401+
store.AppendToStreamAsync(stream, ExpectedVersion.Any, TestEvent.NewTestEvent()).Wait();
402+
store.DeleteStreamAsync(stream, ExpectedVersion.Any, true).Wait();
403+
404+
var result = store.ConditionalAppendToStreamAsync(stream, ExpectedVersion.Any, new[] { TestEvent.NewTestEvent() }).Result;
405+
406+
Assert.AreEqual(ConditionalWriteStatus.StreamDeleted, result.Status);
407+
}
408+
}
362409
}
363410
}

0 commit comments

Comments
 (0)