diff --git a/src/EventStore.ClientAPI.NetCore/Common/Log/DebugLogger.cs b/src/EventStore.ClientAPI.NetCore/Common/Log/DebugLogger.cs index 91c03dd..d3a803e 100644 --- a/src/EventStore.ClientAPI.NetCore/Common/Log/DebugLogger.cs +++ b/src/EventStore.ClientAPI.NetCore/Common/Log/DebugLogger.cs @@ -6,32 +6,32 @@ class DebugLogger : ILogger { public void Error(string format, params object[] args) { - System.Diagnostics.Debug.WriteLine(Log("ERROR", format, args)); + System.Diagnostics.Trace.WriteLine(Log("ERROR", format, args)); } public void Error(Exception ex, string format, params object[] args) { - System.Diagnostics.Debug.WriteLine(Log("ERROR", ex, format, args)); + System.Diagnostics.Trace.WriteLine(Log("ERROR", ex, format, args)); } public void Debug(string format, params object[] args) { - System.Diagnostics.Debug.WriteLine(Log("DEBUG", format, args)); + System.Diagnostics.Trace.WriteLine(Log("DEBUG", format, args)); } public void Debug(Exception ex, string format, params object[] args) { - System.Diagnostics.Debug.WriteLine(Log("DEBUG", ex, format, args)); + System.Diagnostics.Trace.WriteLine(Log("DEBUG", ex, format, args)); } public void Info(string format, params object[] args) { - System.Diagnostics.Debug.WriteLine(Log("INFO", format, args)); + System.Diagnostics.Trace.WriteLine(Log("INFO", format, args)); } public void Info(Exception ex, string format, params object[] args) { - System.Diagnostics.Debug.WriteLine(Log("INFO", ex, format, args)); + System.Diagnostics.Trace.WriteLine(Log("INFO", ex, format, args)); } diff --git a/src/EventStore.ClientAPI.NetCore/ConnectionSettings.cs b/src/EventStore.ClientAPI.NetCore/ConnectionSettings.cs index 5803a8e..5397411 100644 --- a/src/EventStore.ClientAPI.NetCore/ConnectionSettings.cs +++ b/src/EventStore.ClientAPI.NetCore/ConnectionSettings.cs @@ -61,6 +61,10 @@ public static ConnectionSettingsBuilder Create() /// public readonly TimeSpan ReconnectionDelay; /// + /// The amount of time a request for an operation is permitted to be queued awaiting transmission to the server. + /// + public readonly TimeSpan QueueTimeout; + /// /// The amount of time before an operation is considered to have timed out. /// public readonly TimeSpan OperationTimeout; @@ -139,6 +143,7 @@ internal ConnectionSettings(ILogger log, int maxReconnections, bool requireMaster, TimeSpan reconnectionDelay, + TimeSpan queueTimeout, TimeSpan operationTimeout, TimeSpan operationTimeoutCheckPeriod, UserCredentials defaultUserCredentials, @@ -173,6 +178,7 @@ internal ConnectionSettings(ILogger log, MaxReconnections = maxReconnections; RequireMaster = requireMaster; ReconnectionDelay = reconnectionDelay; + QueueTimeout = queueTimeout; OperationTimeout = operationTimeout; OperationTimeoutCheckPeriod = operationTimeoutCheckPeriod; ClientConnectionTimeout = clientConnectionTimeout; diff --git a/src/EventStore.ClientAPI.NetCore/ConnectionSettingsBuilder.cs b/src/EventStore.ClientAPI.NetCore/ConnectionSettingsBuilder.cs index b95ea42..ed8ef6f 100644 --- a/src/EventStore.ClientAPI.NetCore/ConnectionSettingsBuilder.cs +++ b/src/EventStore.ClientAPI.NetCore/ConnectionSettingsBuilder.cs @@ -23,6 +23,7 @@ public class ConnectionSettingsBuilder private bool _requireMaster = Consts.DefaultRequireMaster; private TimeSpan _reconnectionDelay = Consts.DefaultReconnectionDelay; + private TimeSpan _queueTimeout = Consts.DefaultQueueTimeout; private TimeSpan _operationTimeout = Consts.DefaultOperationTimeout; private TimeSpan _operationTimeoutCheckPeriod = Consts.DefaultOperationTimeoutCheckPeriod; @@ -214,6 +215,17 @@ public ConnectionSettingsBuilder SetReconnectionDelayTo(TimeSpan reconnectionDel return this; } + /// + /// Sets the maximum permitted time a request may be queued awaiting transmission; if exceeded an is thrown. + /// + /// + /// + public ConnectionSettingsBuilder SetQueueTimeoutTo(TimeSpan queueTimeout) + { + _queueTimeout = queueTimeout; + return this; + } + /// /// Sets the operation timeout duration. /// @@ -449,6 +461,7 @@ public ConnectionSettings Build() _maxReconnections, _requireMaster, _reconnectionDelay, + _queueTimeout, _operationTimeout, _operationTimeoutCheckPeriod, _defaultUserCredentials, diff --git a/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs b/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs index 837967d..9c0c2a1 100644 --- a/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs +++ b/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs @@ -54,6 +54,16 @@ public DnsClusterSettingsBuilder SetGossipTimeout(TimeSpan timeout) return this; } + /// + /// Allows infinite nodes discovery attempts. + /// + /// A for further configuration. + public DnsClusterSettingsBuilder KeepDiscovering() + { + _maxDiscoverAttempts = Int32.MaxValue; + return this; + } + /// /// Whether to randomly choose a node that's alive from the known nodes. /// diff --git a/src/EventStore.ClientAPI.NetCore/EventStoreCatchUpSubscription.cs b/src/EventStore.ClientAPI.NetCore/EventStoreCatchUpSubscription.cs index d5fb49d..98fac26 100644 --- a/src/EventStore.ClientAPI.NetCore/EventStoreCatchUpSubscription.cs +++ b/src/EventStore.ClientAPI.NetCore/EventStoreCatchUpSubscription.cs @@ -455,7 +455,7 @@ private async Task ProcessEventsAsync(long? lastCommitPosition, AllEventsS : slice.NextPosition >= new Position(lastCommitPosition.Value, lastCommitPosition.Value); if (!done && slice.IsEndOfStream) - Thread.Sleep(1); // we are waiting for server to flush its data + await Task.Delay(1).ConfigureAwait(false); // we are waiting for server to flush its data return done; } @@ -595,7 +595,7 @@ private async Task ProcessEventsAsync(long? lastEventNumber, StreamEventsS } if (!done && slice.IsEndOfStream) - Thread.Sleep(1); // we are waiting for server to flush its data + await Task.Delay(1).ConfigureAwait(false); // we are waiting for server to flush its data return done; } diff --git a/src/EventStore.ClientAPI.NetCore/EventStoreConnection.cs b/src/EventStore.ClientAPI.NetCore/EventStoreConnection.cs index 85a3855..342e3d9 100644 --- a/src/EventStore.ClientAPI.NetCore/EventStoreConnection.cs +++ b/src/EventStore.ClientAPI.NetCore/EventStoreConnection.cs @@ -4,7 +4,6 @@ using EventStore.ClientAPI.Common.Utils; using EventStore.ClientAPI.Internal; using System; -using EventStore.ClientApi.Core.Internal; using EventStore.ClientAPI.SystemData; namespace EventStore.ClientAPI @@ -93,7 +92,7 @@ public static IEventStoreConnection Create(ConnectionSettings connectionSettings connectionSettings.MaxQueueSize, connectionSettings.MaxConcurrentItems, connectionSettings.MaxRetries, connectionSettings.MaxReconnections, connectionSettings.RequireMaster, connectionSettings.ReconnectionDelay, - connectionSettings.OperationTimeout, + connectionSettings.QueueTimeout, connectionSettings.OperationTimeout, connectionSettings.OperationTimeoutCheckPeriod, credential, connectionSettings.UseSslConnection, connectionSettings.TargetHost, connectionSettings.ValidateServer, connectionSettings.FailOnNoServerResponse, @@ -214,5 +213,19 @@ public static IEventStoreConnection Create(ConnectionSettings connectionSettings return new EventStoreNodeConnection(connectionSettings, clusterSettings, endPointDiscoverer, connectionName); } + + /// + /// Creates a new using specific and a custom-defined + /// + /// The to apply to the new connection + /// The custom-defined to use for node discovery + /// Optional name of connection (will be generated automatically, if not provided) + /// a new + public static IEventStoreConnection Create(ConnectionSettings connectionSettings, IEndPointDiscoverer endPointDiscoverer, string connectionName = null) + { + Ensure.NotNull(connectionSettings, "settings"); + Ensure.NotNull(endPointDiscoverer, "endPointDiscoverer"); + return new EventStoreNodeConnection(connectionSettings, null, endPointDiscoverer, connectionName); + } } } diff --git a/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs b/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs index b97640c..0a37376 100644 --- a/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs +++ b/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs @@ -197,7 +197,7 @@ private async void ProcessQueue() { if (_subscription == null) { - Thread.Sleep(1); + await Task.Delay(1).ConfigureAwait(false); } else { diff --git a/src/EventStore.ClientAPI.NetCore/Exceptions/OperationExpiredException.cs b/src/EventStore.ClientAPI.NetCore/Exceptions/OperationExpiredException.cs new file mode 100644 index 0000000..47abe9f --- /dev/null +++ b/src/EventStore.ClientAPI.NetCore/Exceptions/OperationExpiredException.cs @@ -0,0 +1,35 @@ +using System; +using System.Collections.Generic; +using System.Text; + +namespace EventStore.ClientAPI.Exceptions +{ + /// + /// Exception thrown if an operation expires before it can be scheduled. + /// + public class OperationExpiredException : EventStoreConnectionException + { + /// + /// Constructs a new . + /// + public OperationExpiredException() + { + } + + /// + /// Constructs a new . + /// + /// + public OperationExpiredException(string message) : base(message) + { + } + + /// + /// Constructs a new . + /// + public OperationExpiredException(string message, Exception innerException):base(message, innerException) + { + + } + } +} diff --git a/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs b/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs index 105864a..32f96d2 100644 --- a/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs +++ b/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs @@ -50,6 +50,16 @@ public GossipSeedClusterSettingsBuilder SetGossipSeedEndPoints(params GossipSeed _gossipSeeds = gossipSeeds; return this; } + + /// + /// Allows infinite nodes discovery attempts. + /// + /// A for further configuration. + public GossipSeedClusterSettingsBuilder KeepDiscovering() + { + _maxDiscoverAttempts = Int32.MaxValue; + return this; + } /// /// Sets the maximum number of attempts for discovery. diff --git a/src/EventStore.ClientAPI.NetCore/IEndPointDiscoverer.cs b/src/EventStore.ClientAPI.NetCore/IEndPointDiscoverer.cs new file mode 100644 index 0000000..e44ac5f --- /dev/null +++ b/src/EventStore.ClientAPI.NetCore/IEndPointDiscoverer.cs @@ -0,0 +1,18 @@ +using System.Net; +using System.Threading.Tasks; + +namespace EventStore.ClientAPI +{ + /// + /// Represents a way of discovering cluster endpoints. This could be through gossip, consul, text files, etc + /// + public interface IEndPointDiscoverer + { + /// + /// Called to discover a new + /// + /// The The recently failed endpoint + /// + Task DiscoverAsync(IPEndPoint failedTcpEndPoint); + } +} \ No newline at end of file diff --git a/src/EventStore.ClientAPI.NetCore/Internal/ClusterDnsEndPointDiscoverer.cs b/src/EventStore.ClientAPI.NetCore/Internal/ClusterDnsEndPointDiscoverer.cs index d76af0c..f227ffb 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/ClusterDnsEndPointDiscoverer.cs +++ b/src/EventStore.ClientAPI.NetCore/Internal/ClusterDnsEndPointDiscoverer.cs @@ -47,6 +47,9 @@ public ClusterDnsEndPointDiscoverer(ILogger log, public async Task DiscoverAsync(IPEndPoint failedTcpEndPoint ) { + var maxDiscoverAttemptsStr = ""; + if(_maxDiscoverAttempts != Int32.MaxValue) + maxDiscoverAttemptsStr = "/" + _maxDiscoverAttempts; for (int attempt = 1; attempt <= _maxDiscoverAttempts; ++attempt) { //_log.Info("Discovering cluster. Attempt {0}/{1}...", attempt, _maxDiscoverAttempts); @@ -55,18 +58,19 @@ public async Task DiscoverAsync(IPEndPoint failedTcpEndPoint ) var endPoints = await DiscoverEndPoint(failedTcpEndPoint).ConfigureAwait(false); if (endPoints != null) { - _log.Info("Discovering attempt {0}/{1} successful: best candidate is {2}.", attempt, _maxDiscoverAttempts, endPoints); + _log.Info("Discovering attempt {0}{1} successful: best candidate is {2}.", attempt, maxDiscoverAttemptsStr, endPoints); return endPoints.Value; } _log.Info("Discovering attempt {0}/{1} failed: no candidate found.", attempt, _maxDiscoverAttempts); + } catch (Exception exc) { - _log.Info("Discovering attempt {0}/{1} failed with error: {2}.", attempt, _maxDiscoverAttempts, exc); + _log.Info("Discovering attempt {0}{1} failed with error: {2}.", attempt, maxDiscoverAttemptsStr, exc); } - Thread.Sleep(500); + await Task.Delay(500).ConfigureAwait(false); } throw new ClusterException(string.Format("Failed to discover candidate in {0} attempts.", _maxDiscoverAttempts)); } @@ -230,6 +234,7 @@ private ClusterMessages.ClusterInfoDto TryGetGossipFrom(GossipSeed endPoint) break; case NodePreference.Slave: nodes = nodes.OrderBy(nodeEntry => nodeEntry.State != ClusterMessages.VNodeState.Slave).ToArray(); // OrderBy is a stable sort and only affects order of matching entries + RandomShuffle(nodes, 0, nodes.Count(nodeEntry => nodeEntry.State == ClusterMessages.VNodeState.Slave) - 1); break; } diff --git a/src/EventStore.ClientAPI.NetCore/Internal/Consts.cs b/src/EventStore.ClientAPI.NetCore/Internal/Consts.cs index e2566e2..23bc70a 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/Consts.cs +++ b/src/EventStore.ClientAPI.NetCore/Internal/Consts.cs @@ -12,6 +12,7 @@ internal static class Consts public const bool DefaultRequireMaster = true; public static readonly TimeSpan DefaultReconnectionDelay = TimeSpan.FromMilliseconds(100); + public static readonly TimeSpan DefaultQueueTimeout = TimeSpan.Zero; //unlimited public static readonly TimeSpan DefaultOperationTimeout = TimeSpan.FromSeconds(7); public static readonly TimeSpan DefaultOperationTimeoutCheckPeriod = TimeSpan.FromSeconds(1); diff --git a/src/EventStore.ClientAPI.NetCore/Internal/EventStoreNodeConnection.cs b/src/EventStore.ClientAPI.NetCore/Internal/EventStoreNodeConnection.cs index dafa923..a9058f6 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/EventStoreNodeConnection.cs +++ b/src/EventStore.ClientAPI.NetCore/Internal/EventStoreNodeConnection.cs @@ -1,6 +1,5 @@ using System; using System.Collections.Generic; -using System.Threading; using System.Threading.Tasks; using EventStore.ClientAPI.ClientOperations; using EventStore.ClientAPI.Common; @@ -80,14 +79,15 @@ public Task DeleteStreamAsync(string stream, long expectedVersion, return DeleteStreamAsync(stream, expectedVersion, false, userCredentials); } - public Task DeleteStreamAsync(string stream, long expectedVersion, bool hardDelete, UserCredentials userCredentials = null) + public async Task DeleteStreamAsync(string stream, long expectedVersion, bool hardDelete, UserCredentials userCredentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new DeleteStreamOperation(Settings.Log, source, Settings.RequireMaster, - stream, expectedVersion, hardDelete, userCredentials)); - return source.Task; + var operation = new DeleteStreamOperation(Settings.Log, source, Settings.RequireMaster, + stream, expectedVersion, hardDelete, userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } public Task AppendToStreamAsync(string stream, long expectedVersion, params EventData[] events) @@ -106,20 +106,21 @@ public Task AppendToStreamAsync(string stream, long expectedVersion // ReSharper restore RedundantCast } - public Task AppendToStreamAsync(string stream, long expectedVersion, IEnumerable events, UserCredentials userCredentials = null) + public async Task AppendToStreamAsync(string stream, long expectedVersion, IEnumerable events, UserCredentials userCredentials = null) { // ReSharper disable PossibleMultipleEnumeration Ensure.NotNullOrEmpty(stream, "stream"); Ensure.NotNull(events, "events"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new AppendToStreamOperation(Settings.Log, source, Settings.RequireMaster, - stream, expectedVersion, events, userCredentials)); - return source.Task; + var operation = new AppendToStreamOperation(Settings.Log, source, Settings.RequireMaster, + stream, expectedVersion, events, userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); // ReSharper restore PossibleMultipleEnumeration } - public Task ConditionalAppendToStreamAsync(string stream, long expectedVersion, IEnumerable events, + public async Task ConditionalAppendToStreamAsync(string stream, long expectedVersion, IEnumerable events, UserCredentials userCredentials = null) { // ReSharper disable PossibleMultipleEnumeration @@ -127,20 +128,22 @@ public Task ConditionalAppendToStreamAsync(string stream Ensure.NotNull(events, "events"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new ConditionalAppendToStreamOperation(Settings.Log, source, Settings.RequireMaster, - stream, expectedVersion, events, userCredentials)); - return source.Task; + var operation = new ConditionalAppendToStreamOperation(Settings.Log, source, Settings.RequireMaster, + stream, expectedVersion, events, userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); // ReSharper restore PossibleMultipleEnumeration } - public Task StartTransactionAsync(string stream, long expectedVersion, UserCredentials userCredentials = null) + public async Task StartTransactionAsync(string stream, long expectedVersion, UserCredentials userCredentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new StartTransactionOperation(Settings.Log, source, Settings.RequireMaster, - stream, expectedVersion, this, userCredentials)); - return source.Task; + var operation = new StartTransactionOperation(Settings.Log, source, Settings.RequireMaster, + stream, expectedVersion, this, userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } public EventStoreTransaction ContinueTransaction(long transactionId, UserCredentials userCredentials = null) @@ -149,42 +152,44 @@ public EventStoreTransaction ContinueTransaction(long transactionId, UserCredent return new EventStoreTransaction(transactionId, userCredentials, this); } - Task IEventStoreTransactionConnection.TransactionalWriteAsync(EventStoreTransaction transaction, IEnumerable events, UserCredentials userCredentials) + async Task IEventStoreTransactionConnection.TransactionalWriteAsync(EventStoreTransaction transaction, IEnumerable events, UserCredentials userCredentials) { // ReSharper disable PossibleMultipleEnumeration Ensure.NotNull(transaction, "transaction"); Ensure.NotNull(events, "events"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new TransactionalWriteOperation(Settings.Log, source, Settings.RequireMaster, - transaction.TransactionId, events, userCredentials)); - return source.Task; + var operation = new TransactionalWriteOperation(Settings.Log, source, Settings.RequireMaster, + transaction.TransactionId, events, userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + await source.Task.ConfigureAwait(false); // ReSharper restore PossibleMultipleEnumeration } - Task IEventStoreTransactionConnection.CommitTransactionAsync(EventStoreTransaction transaction, UserCredentials userCredentials) + async Task IEventStoreTransactionConnection.CommitTransactionAsync(EventStoreTransaction transaction, UserCredentials userCredentials) { Ensure.NotNull(transaction, "transaction"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new CommitTransactionOperation(Settings.Log, source, Settings.RequireMaster, - transaction.TransactionId, userCredentials)); - return source.Task; + var operation = new CommitTransactionOperation(Settings.Log, source, Settings.RequireMaster, + transaction.TransactionId, userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } - public Task ReadEventAsync(string stream, long eventNumber, bool resolveLinkTos, UserCredentials userCredentials = null) + public async Task ReadEventAsync(string stream, long eventNumber, bool resolveLinkTos, UserCredentials userCredentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); if (eventNumber < -1) throw new ArgumentOutOfRangeException(nameof(eventNumber)); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var operation = new ReadEventOperation(Settings.Log, source, stream, eventNumber, resolveLinkTos, Settings.RequireMaster, userCredentials); - EnqueueOperation(operation); - return source.Task; + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } - public Task ReadStreamEventsForwardAsync(string stream, long start, int count, bool resolveLinkTos, UserCredentials userCredentials = null) + public async Task ReadStreamEventsForwardAsync(string stream, long start, int count, bool resolveLinkTos, UserCredentials userCredentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); Ensure.Nonnegative(start, "start"); @@ -193,11 +198,11 @@ public Task ReadStreamEventsForwardAsync(string stream, long var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var operation = new ReadStreamEventsForwardOperation(Settings.Log, source, stream, start, count, resolveLinkTos, Settings.RequireMaster, userCredentials); - EnqueueOperation(operation); - return source.Task; + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } - public Task ReadStreamEventsBackwardAsync(string stream, long start, int count, bool resolveLinkTos, UserCredentials userCredentials = null) + public async Task ReadStreamEventsBackwardAsync(string stream, long start, int count, bool resolveLinkTos, UserCredentials userCredentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); Ensure.Positive(count, "count"); @@ -205,37 +210,37 @@ public Task ReadStreamEventsBackwardAsync(string stream, long var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var operation = new ReadStreamEventsBackwardOperation(Settings.Log, source, stream, start, count, resolveLinkTos, Settings.RequireMaster, userCredentials); - EnqueueOperation(operation); - return source.Task; + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } - public Task ReadAllEventsForwardAsync(Position position, int maxCount, bool resolveLinkTos, UserCredentials userCredentials = null) + public async Task ReadAllEventsForwardAsync(Position position, int maxCount, bool resolveLinkTos, UserCredentials userCredentials = null) { Ensure.Positive(maxCount, "maxCount"); if (maxCount > ClientApiConstants.MaxReadSize) throw new ArgumentException(string.Format("Count should be less than {0}. For larger reads you should page.", ClientApiConstants.MaxReadSize)); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var operation = new ReadAllEventsForwardOperation(Settings.Log, source, position, maxCount, resolveLinkTos, Settings.RequireMaster, userCredentials); - EnqueueOperation(operation); - return source.Task; + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } - public Task ReadAllEventsBackwardAsync(Position position, int maxCount, bool resolveLinkTos, UserCredentials userCredentials = null) + public async Task ReadAllEventsBackwardAsync(Position position, int maxCount, bool resolveLinkTos, UserCredentials userCredentials = null) { Ensure.Positive(maxCount, "maxCount"); if (maxCount > ClientApiConstants.MaxReadSize) throw new ArgumentException(string.Format("Count should be less than {0}. For larger reads you should page.", ClientApiConstants.MaxReadSize)); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var operation = new ReadAllEventsBackwardOperation(Settings.Log, source, position, maxCount, resolveLinkTos, Settings.RequireMaster, userCredentials); - EnqueueOperation(operation); - return source.Task; + await EnqueueOperation(operation).ConfigureAwait(false); + return await source.Task.ConfigureAwait(false); } - private void EnqueueOperation(IClientOperation operation) + private async Task EnqueueOperation(IClientOperation operation) { while (_handler.TotalOperationCount >= Settings.MaxQueueSize) { - Thread.Sleep(1); + await Task.Delay(1).ConfigureAwait(false); } _handler.EnqueueMessage(new StartOperationMessage(operation, Settings.MaxRetries, Settings.OperationTimeout)); } @@ -403,23 +408,25 @@ public EventStorePersistentSubscription ConnectToPersistentSubscriptionForAll( */ - public Task CreatePersistentSubscriptionAsync(string stream, string groupName, PersistentSubscriptionSettings settings, UserCredentials credentials = null) { + public async Task CreatePersistentSubscriptionAsync(string stream, string groupName, PersistentSubscriptionSettings settings, UserCredentials credentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); Ensure.NotNullOrEmpty(groupName, "groupName"); Ensure.NotNull(settings, "settings"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new CreatePersistentSubscriptionOperation(Settings.Log, source, stream, groupName, settings, credentials)); - return source.Task; + var operation = new CreatePersistentSubscriptionOperation(Settings.Log, source, stream, groupName, settings, credentials); + await EnqueueOperation(operation).ConfigureAwait(false); + await source.Task.ConfigureAwait(false); } - public Task UpdatePersistentSubscriptionAsync(string stream, string groupName, PersistentSubscriptionSettings settings, UserCredentials credentials = null) + public async Task UpdatePersistentSubscriptionAsync(string stream, string groupName, PersistentSubscriptionSettings settings, UserCredentials credentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); Ensure.NotNullOrEmpty(groupName, "groupName"); Ensure.NotNull(settings, "settings"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new UpdatePersistentSubscriptionOperation(Settings.Log, source, stream, groupName, settings, credentials)); - return source.Task; + var operation = new UpdatePersistentSubscriptionOperation(Settings.Log, source, stream, groupName, settings, credentials); + await EnqueueOperation(operation).ConfigureAwait(false); + await source.Task.ConfigureAwait(false); } /* @@ -432,13 +439,14 @@ public Task CreatePersistentSubscriptionForA return source.Task; } - */ - public Task DeletePersistentSubscriptionAsync(string stream, string groupName, UserCredentials userCredentials = null) { +*/ + public async Task DeletePersistentSubscriptionAsync(string stream, string groupName, UserCredentials userCredentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); Ensure.NotNullOrEmpty(groupName, "groupName"); var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - EnqueueOperation(new DeletePersistentSubscriptionOperation(Settings.Log, source, stream, groupName, userCredentials)); - return source.Task; + var operation = new DeletePersistentSubscriptionOperation(Settings.Log, source, stream, groupName, userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + await source.Task.ConfigureAwait(false); } /* @@ -457,7 +465,7 @@ public Task SetStreamMetadataAsync(string stream, long expectedMeta return SetStreamMetadataAsync(stream, expectedMetastreamVersion, metadata.AsJsonBytes(), userCredentials); } - public Task SetStreamMetadataAsync(string stream, long expectedMetastreamVersion, byte[] metadata, UserCredentials userCredentials = null) + public async Task SetStreamMetadataAsync(string stream, long expectedMetastreamVersion, byte[] metadata, UserCredentials userCredentials = null) { Ensure.NotNullOrEmpty(stream, "stream"); if (SystemStreams.IsMetastream(stream)) @@ -466,14 +474,16 @@ public Task SetStreamMetadataAsync(string stream, long expectedMeta var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var metaevent = new EventData(Guid.NewGuid(), SystemEventTypes.StreamMetadata, true, metadata ?? Empty.ByteArray, null); - EnqueueOperation(new AppendToStreamOperation(Settings.Log, - source, - Settings.RequireMaster, - SystemStreams.MetastreamOf(stream), - expectedMetastreamVersion, - new[] { metaevent }, - userCredentials)); - return source.Task; + var operation = new AppendToStreamOperation(Settings.Log, + source, + Settings.RequireMaster, + SystemStreams.MetastreamOf(stream), + expectedMetastreamVersion, + new[] {metaevent}, + userCredentials); + await EnqueueOperation(operation).ConfigureAwait(false); + + return await source.Task.ConfigureAwait(false); } public Task GetStreamMetadataAsync(string stream, UserCredentials userCredentials = null) diff --git a/src/EventStore.ClientAPI.NetCore/Internal/IEndPointDiscoverer.cs b/src/EventStore.ClientAPI.NetCore/Internal/IEndPointDiscoverer.cs deleted file mode 100644 index f84f167..0000000 --- a/src/EventStore.ClientAPI.NetCore/Internal/IEndPointDiscoverer.cs +++ /dev/null @@ -1,10 +0,0 @@ -using System.Net; -using System.Threading.Tasks; - -namespace EventStore.ClientAPI.Internal -{ - internal interface IEndPointDiscoverer - { - Task DiscoverAsync(IPEndPoint failedTcpEndPoint); - } -} \ No newline at end of file diff --git a/src/EventStore.ClientAPI.NetCore/Internal/OperationsManager.cs b/src/EventStore.ClientAPI.NetCore/Internal/OperationsManager.cs index ee9b56f..e86b486 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/OperationsManager.cs +++ b/src/EventStore.ClientAPI.NetCore/Internal/OperationsManager.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -54,7 +55,7 @@ internal class OperationsManager private readonly string _connectionName; private readonly ConnectionSettings _settings; private readonly Dictionary _activeOperations = new Dictionary(); - private readonly Queue _waitingOperations = new Queue(); + private readonly ConcurrentQueue _waitingOperations = new ConcurrentQueue(); private readonly List _retryPendingOperations = new List(); private readonly object _lock = new object(); private int _totalOperationCount; @@ -82,7 +83,8 @@ public void CleanUp() operation.Operation.Fail(connectionClosedException); } _activeOperations.Clear(); - _waitingOperations.Clear(); + while (_waitingOperations.TryDequeue(out _)) {} + _retryPendingOperations.Clear(); _totalOperationCount = 0; } @@ -175,14 +177,43 @@ public void TryScheduleWaitingOperations(TcpPackageConnection connection) Ensure.NotNull(connection, "connection"); lock(_lock) { - while (_waitingOperations.Count > 0 && _activeOperations.Count < _settings.MaxConcurrentItems) + // We don't want to transmit or retain expired requests, so we trim any from before the cutoff implied by the current time + var cutoff = _settings.QueueTimeout == TimeSpan.Zero ? (DateTime?)null : DateTime.UtcNow - _settings.QueueTimeout; + + OperationItem operation; + while (_activeOperations.Count < _settings.MaxConcurrentItems) + { + if (!_waitingOperations.TryDequeue(out operation)) + break; + if (cutoff == null || !TryExpireItem(cutoff.Value, operation)) + ExecuteOperation(operation, connection); + } + + if (cutoff != null) { - ExecuteOperation(_waitingOperations.Dequeue(), connection); + // In case the active operations queue is at capacity, we trim expired items from the front of the queue + while (_waitingOperations.TryPeek(out operation) && TryExpireItem(cutoff.Value, operation)) + { + _waitingOperations.TryDequeue(out operation); + } } _totalOperationCount = _activeOperations.Count + _waitingOperations.Count; } } + bool TryExpireItem(DateTime cutoffDate, OperationItem operation) + { + if (operation.CreatedTime > cutoffDate) + return false; + + var err = string.Format("EventStoreConnection '{0}': request expired.\n" + + "UTC now: {1:HH:mm:ss.fff}, operation: {2}.", + _connectionName, DateTime.UtcNow, operation); + _settings.Log.Debug(err); + operation.Operation.Fail(new OperationExpiredException(err)); + return true; + } + public void ExecuteOperation(OperationItem operation, TcpPackageConnection connection) { operation.ConnectionId = connection.ConnectionId; operation.LastUpdated = DateTime.UtcNow; diff --git a/src/EventStore.ClientAPI.NetCore/Internal/SingleEndpointDiscoverer.cs b/src/EventStore.ClientAPI.NetCore/Internal/SingleEndpointDiscoverer.cs index d161efd..075e026 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/SingleEndpointDiscoverer.cs +++ b/src/EventStore.ClientAPI.NetCore/Internal/SingleEndpointDiscoverer.cs @@ -1,12 +1,10 @@ using System; -using System.Collections.Generic; using System.Linq; using System.Net; using System.Net.Sockets; using System.Threading.Tasks; -using EventStore.ClientAPI.Internal; -namespace EventStore.ClientApi.Core.Internal +namespace EventStore.ClientAPI.Internal { internal class SingleEndpointDiscoverer : IEndPointDiscoverer { diff --git a/src/EventStore.ClientAPI.NetCore/Internal/NodeEndPoints.cs b/src/EventStore.ClientAPI.NetCore/NodeEndPoints.cs similarity index 52% rename from src/EventStore.ClientAPI.NetCore/Internal/NodeEndPoints.cs rename to src/EventStore.ClientAPI.NetCore/NodeEndPoints.cs index 44db2ae..523ec5f 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/NodeEndPoints.cs +++ b/src/EventStore.ClientAPI.NetCore/NodeEndPoints.cs @@ -1,13 +1,28 @@ using System; using System.Net; -namespace EventStore.ClientAPI.Internal +namespace EventStore.ClientAPI { - internal struct NodeEndPoints + /// + /// Represents a node and its possible endpoints + /// + public struct NodeEndPoints { + /// + /// The tcp endpoint of the node. + /// public readonly IPEndPoint TcpEndPoint; + + /// + /// The ssl endpoint of the node. + /// public readonly IPEndPoint SecureTcpEndPoint; + /// + /// Called to create a new NodeEndPoints + /// + /// The tcp endpoint of the node + /// he ssl endpoint of the node public NodeEndPoints(IPEndPoint tcpEndPoint, IPEndPoint secureTcpEndPoint) { if ((tcpEndPoint ?? secureTcpEndPoint) == null) throw new ArgumentException("Both endpoints are null."); @@ -15,6 +30,9 @@ public NodeEndPoints(IPEndPoint tcpEndPoint, IPEndPoint secureTcpEndPoint) SecureTcpEndPoint = secureTcpEndPoint; } + /// + /// Formats the endpoints as a string + /// public override string ToString() { return string.Format("[{0}, {1}]", diff --git a/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs b/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs index 0e5797f..65657f0 100644 --- a/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs +++ b/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs @@ -9,7 +9,7 @@ namespace EventStore.ClientAPI.Projections { /// - /// API for executinmg queries in the Event Store through C# code. Communicates + /// API for executing queries in the Event Store through C# code. Communicates /// with the Event Store over the RESTful API. /// public class QueryManager diff --git a/src/EventStore.ClientAPI.NetCore/Transport.Http/EndpointExtensions.cs b/src/EventStore.ClientAPI.NetCore/Transport.Http/EndpointExtensions.cs index 622b3e8..2b63e38 100644 --- a/src/EventStore.ClientAPI.NetCore/Transport.Http/EndpointExtensions.cs +++ b/src/EventStore.ClientAPI.NetCore/Transport.Http/EndpointExtensions.cs @@ -11,7 +11,7 @@ public static string ToHttpUrl(this EndPoint endPoint, string schema, string raw if (endPoint is IPEndPoint) { var ipEndPoint = endPoint as IPEndPoint; - return CreateHttpUrl(schema, ipEndPoint.Address.ToString(), ipEndPoint.Port, rawUrl != null ? rawUrl.TrimStart('/') : string.Empty); + return CreateHttpUrl(schema, ipEndPoint.ToString(), rawUrl != null ? rawUrl.TrimStart('/') : string.Empty); } if (endPoint is DnsEndPoint) { @@ -26,7 +26,7 @@ public static string ToHttpUrl(this EndPoint endPoint, string schema, string for if (endPoint is IPEndPoint) { var ipEndPoint = endPoint as IPEndPoint; - return CreateHttpUrl(schema, ipEndPoint.Address.ToString(), ipEndPoint.Port, string.Format(formatString.TrimStart('/'), args)); + return CreateHttpUrl(schema, ipEndPoint.ToString(), string.Format(formatString.TrimStart('/'), args)); } if (endPoint is DnsEndPoint) { @@ -40,5 +40,10 @@ private static string CreateHttpUrl(string schema, string host, int port, string { return $"{schema}://{host}:{port}/{path}"; } + + private static string CreateHttpUrl(string schema, string address, string path) + { + return $"{schema}://{address}/{path}"; + } } } diff --git a/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_backward_should.cs b/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_backward_should.cs index aa2d157..a4b6ff2 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_backward_should.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_backward_should.cs @@ -99,7 +99,7 @@ public void be_able_to_read_events_slice_at_time() public void throw_when_got_int_max_value_as_maxcount() { - Assert.Throws( + Assert.ThrowsAsync( () => _conn.ReadAllEventsBackwardAsync(Position.Start, int.MaxValue, resolveLinkTos: false)); } } diff --git a/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_forward_should.cs b/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_forward_should.cs index 3b2242a..8bd70a4 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_forward_should.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/read_all_events_forward_should.cs @@ -96,7 +96,7 @@ public void return_partial_slice_if_not_enough_events() [Category("Network")] public void throw_when_got_int_max_value_as_maxcount() { - Assert.Throws( + Assert.ThrowsAsync( () => _conn.ReadAllEventsForwardAsync(Position.Start, int.MaxValue, resolveLinkTos: false)); } diff --git a/test/EventStore.ClientAPI.NetCore.Tests/read_event_should.cs b/test/EventStore.ClientAPI.NetCore.Tests/read_event_should.cs index c5f9e3e..f0bafbd 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/read_event_should.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/read_event_should.cs @@ -26,19 +26,19 @@ protected override void When() [Test, Category("Network")] public void throw_if_stream_id_is_null() { - Assert.Throws(() => _conn.ReadEventAsync(null, 0, resolveLinkTos: false)); + Assert.ThrowsAsync(() => _conn.ReadEventAsync(null, 0, resolveLinkTos: false)); } [Test, Category("Network")] public void throw_if_stream_id_is_empty() { - Assert.Throws(() => _conn.ReadEventAsync("", 0, resolveLinkTos: false)); + Assert.ThrowsAsync(() => _conn.ReadEventAsync("", 0, resolveLinkTos: false)); } [Test, Category("Network")] public void throw_if_event_number_is_less_than_minus_one() { - Assert.Throws(() => _conn.ReadEventAsync("stream", -2, resolveLinkTos: false)); + Assert.ThrowsAsync(() => _conn.ReadEventAsync("stream", -2, resolveLinkTos: false)); } [Test, Category("Network")] diff --git a/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_backward_should.cs b/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_backward_should.cs index f474770..82260aa 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_backward_should.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_backward_should.cs @@ -23,7 +23,7 @@ public void throw_if_count_le_zero() using (var store = BuildConnection()) { store.ConnectAsync().Wait(); - Assert.Throws(() => store.ReadStreamEventsBackwardAsync(stream, 0, 0, resolveLinkTos: false)); + Assert.ThrowsAsync(() => store.ReadStreamEventsBackwardAsync(stream, 0, 0, resolveLinkTos: false)); } } @@ -206,7 +206,7 @@ public void throw_when_got_int_max_value_as_maxcount() { store.ConnectAsync().Wait(); - Assert.Throws(() => store.ReadStreamEventsBackwardAsync("foo", StreamPosition.Start, int.MaxValue, resolveLinkTos: false)); + Assert.ThrowsAsync(() => store.ReadStreamEventsBackwardAsync("foo", StreamPosition.Start, int.MaxValue, resolveLinkTos: false)); } } diff --git a/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_forward_should.cs b/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_forward_should.cs index d77bb41..5fc5f4e 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_forward_should.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/read_event_stream_forward_should.cs @@ -23,7 +23,7 @@ public void throw_if_count_le_zero() using (var store = BuildConnection()) { store.ConnectAsync().Wait(); - Assert.Throws(() => store.ReadStreamEventsForwardAsync(stream, 0, 0, resolveLinkTos: false)); + Assert.ThrowsAsync(() => store.ReadStreamEventsForwardAsync(stream, 0, 0, resolveLinkTos: false)); } } @@ -35,7 +35,7 @@ public void throw_if_start_lt_zero() using (var store = BuildConnection()) { store.ConnectAsync().Wait(); - Assert.Throws(() => store.ReadStreamEventsForwardAsync(stream, -1, 1, resolveLinkTos: false)); + Assert.ThrowsAsync(() => store.ReadStreamEventsForwardAsync(stream, -1, 1, resolveLinkTos: false)); } } @@ -138,7 +138,7 @@ public void throw_when_got_int_max_value_as_maxcount() { store.ConnectAsync().Wait(); - Assert.Throws(() => store.ReadStreamEventsForwardAsync("foo", StreamPosition.Start, int.MaxValue, resolveLinkTos: false)); + Assert.ThrowsAsync(() => store.ReadStreamEventsForwardAsync("foo", StreamPosition.Start, int.MaxValue, resolveLinkTos: false)); } } diff --git a/test/EventStore.ClientAPI.NetCore.Tests/transaction.cs b/test/EventStore.ClientAPI.NetCore.Tests/transaction.cs index 0b31aff..aee95bb 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/transaction.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/transaction.cs @@ -1,18 +1,18 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; using EventStore.ClientAPI; using EventStore.ClientAPI.Common.Utils; using EventStore.ClientAPI.Exceptions; using EventStore.Core.Tests.ClientAPI.Helpers; using NUnit.Framework; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; namespace EventStore.Core.Tests.ClientAPI { [TestFixture, Category("LongRunning")] - public class transaction + public class transaction { protected virtual IEventStoreConnection BuildConnection() @@ -30,7 +30,7 @@ public void should_start_on_non_existing_stream_with_correct_exp_ver_and_create_ store.ConnectAsync().Wait(); using (var transaction = store.StartTransactionAsync(stream, ExpectedVersion.NoStream).Result) { - transaction.WriteAsync(new[] {TestEvent.NewTestEvent()}).Wait(); + transaction.WriteAsync(new[] { TestEvent.NewTestEvent() }).Wait(); Assert.AreEqual(0, transaction.CommitAsync().Result.NextExpectedVersion); } } @@ -46,7 +46,7 @@ public void should_start_on_non_existing_stream_with_exp_ver_any_and_create_stre store.ConnectAsync().Wait(); using (var transaction = store.StartTransactionAsync(stream, ExpectedVersion.Any).Result) { - transaction.WriteAsync(new[] {TestEvent.NewTestEvent()}).Wait(); + transaction.WriteAsync(new[] { TestEvent.NewTestEvent() }).Wait(); Assert.AreEqual(0, transaction.CommitAsync().Result.NextExpectedVersion); } } @@ -134,53 +134,51 @@ public void should_commit_when_writing_with_exp_ver_any_even_while_somene_is_wri const int totalTranWrites = 500; const int totalPlainWrites = 500; - + //500 events during transaction ThreadPool.QueueUserWorkItem(_ => { - Assert.DoesNotThrow(() => { - using (var store = BuildConnection()) + using (var store = BuildConnection()) + { + store.ConnectAsync().Wait(); + using (var transaction = store.StartTransactionAsync(stream, ExpectedVersion.Any).Result) { - store.ConnectAsync().Wait(); - using (var transaction = store.StartTransactionAsync(stream, ExpectedVersion.Any).Result) - { - - var writes = new List(); - for (int i = 0; i < totalTranWrites; i++) - { - writes.Add(transaction.WriteAsync(TestEvent.NewTestEvent(i.ToString(), "trans write"))); - } - Task.WaitAll(writes.ToArray()); - transaction.CommitAsync().Wait(); - transWritesCompleted.Set(); + var writes = new List(); + for (int i = 0; i < totalTranWrites; i++) + { + writes.Add(transaction.WriteAsync(TestEvent.NewTestEvent(i.ToString(), "trans write"))); } + + Task.WaitAll(writes.ToArray()); + transaction.CommitAsync().Wait(); + transWritesCompleted.Set(); } - }); + } + }); //500 events to same stream in parallel ThreadPool.QueueUserWorkItem(_ => { - Assert.DoesNotThrow(() => { - using (var store = BuildConnection()) + using (var store = BuildConnection()) + { + store.ConnectAsync().Wait(); + var writes = new List(); + for (int i = 0; i < totalPlainWrites; i++) { - store.ConnectAsync().Wait(); - var writes = new List(); - for (int i = 0; i < totalPlainWrites; i++) - { - writes.Add(store.AppendToStreamAsync(stream, - ExpectedVersion.Any, - new[] {TestEvent.NewTestEvent(i.ToString(), "plain write")})); - } - Task.WaitAll(writes.ToArray()); - writesToSameStreamCompleted.Set(); + writes.Add(store.AppendToStreamAsync(stream, + ExpectedVersion.Any, + new[] { TestEvent.NewTestEvent(i.ToString(), "plain write") })); } - }); + + Task.WaitAll(writes.ToArray()); + writesToSameStreamCompleted.Set(); + } }); - transWritesCompleted.Wait(); - writesToSameStreamCompleted.Wait(); + Assert.DoesNotThrow(() => transWritesCompleted.Wait()); + Assert.DoesNotThrow(() => writesToSameStreamCompleted.Wait()); // check all written using (var store = BuildConnection()) @@ -206,7 +204,7 @@ public void should_fail_to_commit_if_started_with_correct_ver_but_committing_wit store.ConnectAsync().Wait(); using (var transaction = store.StartTransactionAsync(stream, ExpectedVersion.EmptyStream).Result) { - store.AppendToStreamAsync(stream, ExpectedVersion.EmptyStream, new[] {TestEvent.NewTestEvent()}).Wait(); + store.AppendToStreamAsync(stream, ExpectedVersion.EmptyStream, new[] { TestEvent.NewTestEvent() }).Wait(); transaction.WriteAsync(TestEvent.NewTestEvent()).Wait(); Assert.That(() => transaction.CommitAsync().Wait(), Throws.Exception.TypeOf().With.InnerException.TypeOf()); @@ -223,7 +221,7 @@ public void should_not_fail_to_commit_if_started_with_wrong_ver_but_committing_w store.ConnectAsync().Wait(); using (var transaction = store.StartTransactionAsync(stream, 0).Result) { - store.AppendToStreamAsync(stream, ExpectedVersion.EmptyStream, new[] {TestEvent.NewTestEvent()}).Wait(); + store.AppendToStreamAsync(stream, ExpectedVersion.EmptyStream, new[] { TestEvent.NewTestEvent() }).Wait(); transaction.WriteAsync(TestEvent.NewTestEvent()).Wait(); Assert.AreEqual(1, transaction.CommitAsync().Result.NextExpectedVersion); } @@ -259,11 +257,11 @@ public void idempotency_is_correct_for_explicit_transactions_with_expected_versi var e = new EventData(Guid.NewGuid(), "SomethingHappened", true, Helper.UTF8NoBom.GetBytes("{Value:42}"), null); var transaction1 = store.StartTransactionAsync(streamId, ExpectedVersion.Any).Result; - transaction1.WriteAsync(new[] {e}).Wait(); + transaction1.WriteAsync(new[] { e }).Wait(); Assert.AreEqual(0, transaction1.CommitAsync().Result.NextExpectedVersion); var transaction2 = store.StartTransactionAsync(streamId, ExpectedVersion.Any).Result; - transaction2.WriteAsync(new[] {e}).Wait(); + transaction2.WriteAsync(new[] { e }).Wait(); Assert.AreEqual(0, transaction2.CommitAsync().Result.NextExpectedVersion); var res = store.ReadStreamEventsForwardAsync(streamId, 0, 100, false).Result;