diff --git a/README.md b/README.md index df57e34..9211760 100644 --- a/README.md +++ b/README.md @@ -1,24 +1,13 @@ -# EventStore .NET Core Client - -Official direct port of the Event Store .NET client to .NET Core targetting .netstandard 2.0 - -Works with EventStore server 4.0.0 and above. +# Deprecated -[![wercker status](https://app.wercker.com/status/21204a69779b2b4543242fd543d0aaab/s/master "wercker status")](https://app.wercker.com/project/byKey/21204a69779b2b4543242fd543d0aaab) -[![Appveyor status](https://ci.appveyor.com/api/projects/status/8tcaw05oa59fkba2/branch/master?svg=true)](https://ci.appveyor.com/project/EventStore/clientapi-netcore/branch/master) - -## Installation - -You can install the Eventstore .NET Core client by copying and pasting the following command into your Package Manager Console within Visual Studio (Tools > NuGet Package Manager > Package Manager Console). +# EventStore .NET Core Client -```bash -Install-Package EventStore.ClientAPI.NetCore -``` +This repository contained an Event Store .NET client for .NET Core, targetting netstandard-2.0. -Alternatively if you're using the .NET Core CLI then you can install the client via the command line interface with the following command: +As of version 5.0.0, the official client maintained as part of [github.com/EventStore/EventStore](github.com/EventStore/EventStore) now targets all supported .NET platforms: -```bash -dotnet add package EventStore.ClientAPI.NetCore -``` +- netstandard2.0 +- net452 +- net46 -For more information see the [EventStore.ClientAPI.NetCore NuGet page](https://www.nuget.org/packages/EventStore.ClientAPI.NetCore/). +This code and repository is not maintained. diff --git a/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs b/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs index 45feb30..c503d3b 100644 --- a/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs +++ b/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs @@ -43,7 +43,7 @@ public static ClusterSettingsBuilder Create() /// /// Prefer a randomly selected node. /// - public bool PreferRandomNode; + public NodePreference NodePreference; /// /// Used if connecting with gossip seeds. @@ -51,15 +51,15 @@ public static ClusterSettingsBuilder Create() /// Endpoints for seeding gossip. /// Maximum number of attempts to discover the cluster. /// Timeout for cluster gossip. - /// Whether to prefer a random node selection or master. - internal ClusterSettings(GossipSeed[] gossipSeeds, int maxDiscoverAttempts, TimeSpan gossipTimeout, bool preferRandomNode) + /// Whether to prefer slave, random, or master node selection. + internal ClusterSettings(GossipSeed[] gossipSeeds, int maxDiscoverAttempts, TimeSpan gossipTimeout, NodePreference nodePreference) { ClusterDns = ""; MaxDiscoverAttempts = maxDiscoverAttempts; ExternalGossipPort = 0; GossipTimeout = gossipTimeout; GossipSeeds = gossipSeeds; - PreferRandomNode = preferRandomNode; + NodePreference = nodePreference; } /// @@ -69,8 +69,8 @@ internal ClusterSettings(GossipSeed[] gossipSeeds, int maxDiscoverAttempts, Time /// The maximum number of attempts for discovering endpoints. /// The well-known endpoint on which cluster managers are running. /// Timeout for cluster gossip. - /// Whether to prefer a random node selection or master. - internal ClusterSettings(string clusterDns, int maxDiscoverAttempts, int externalGossipPort, TimeSpan gossipTimeout, bool preferRandomNode) + /// Whether to prefer slave, random, or master node selection. + internal ClusterSettings(string clusterDns, int maxDiscoverAttempts, int externalGossipPort, TimeSpan gossipTimeout, NodePreference nodePreference) { Ensure.NotNullOrEmpty(clusterDns, "clusterDns"); if (maxDiscoverAttempts < -1) @@ -82,7 +82,7 @@ internal ClusterSettings(string clusterDns, int maxDiscoverAttempts, int externa ExternalGossipPort = externalGossipPort; GossipTimeout = gossipTimeout; GossipSeeds = new GossipSeed[0]; - PreferRandomNode = preferRandomNode; + NodePreference = nodePreference; } } } 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 152cca0..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; @@ -124,7 +128,7 @@ public static ConnectionSettingsBuilder Create() /// /// Whether to randomly choose a node that's alive from the known nodes. /// - public readonly bool PreferRandomNode; + public readonly NodePreference NodePreference; /// /// The interval after which a client will time out during connection. @@ -139,6 +143,7 @@ internal ConnectionSettings(ILogger log, int maxReconnections, bool requireMaster, TimeSpan reconnectionDelay, + TimeSpan queueTimeout, TimeSpan operationTimeout, TimeSpan operationTimeoutCheckPeriod, UserCredentials defaultUserCredentials, @@ -154,7 +159,7 @@ internal ConnectionSettings(ILogger log, int maxDiscoverAttempts, int externalGossipPort, TimeSpan gossipTimeout, - bool preferRandomNode) + NodePreference nodePreference) { Ensure.NotNull(log, "log"); Ensure.Positive(maxQueueSize, "maxQueueSize"); @@ -173,6 +178,7 @@ internal ConnectionSettings(ILogger log, MaxReconnections = maxReconnections; RequireMaster = requireMaster; ReconnectionDelay = reconnectionDelay; + QueueTimeout = queueTimeout; OperationTimeout = operationTimeout; OperationTimeoutCheckPeriod = operationTimeoutCheckPeriod; ClientConnectionTimeout = clientConnectionTimeout; @@ -189,7 +195,7 @@ internal ConnectionSettings(ILogger log, MaxDiscoverAttempts = maxDiscoverAttempts; ExternalGossipPort = externalGossipPort; GossipTimeout = gossipTimeout; - PreferRandomNode = preferRandomNode; + NodePreference = nodePreference; } } } diff --git a/src/EventStore.ClientAPI.NetCore/ConnectionSettingsBuilder.cs b/src/EventStore.ClientAPI.NetCore/ConnectionSettingsBuilder.cs index 1c1edfa..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; @@ -40,7 +41,7 @@ public class ConnectionSettingsBuilder private int _gossipExternalHttpPort = Consts.DefaultClusterManagerExternalHttpPort; private TimeSpan _gossipTimeout = TimeSpan.FromSeconds(1); private GossipSeed[] _gossipSeeds; - private bool _preferRandomNode = false; + private NodePreference _nodePreference = NodePreference.Master; internal ConnectionSettingsBuilder() @@ -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. /// @@ -350,7 +362,17 @@ public ConnectionSettingsBuilder SetGossipTimeout(TimeSpan timeout) /// A for further configuration. public ConnectionSettingsBuilder PreferRandomNode() { - _preferRandomNode = true; + _nodePreference = NodePreference.Random; + return this; + } + + /// + /// Whether to prioritize choosing a slave node that's alive from the known nodes. + /// + /// A for further configuration. + public ConnectionSettingsBuilder PreferSlaveNode() + { + _nodePreference = NodePreference.Slave; return this; } @@ -439,6 +461,7 @@ public ConnectionSettings Build() _maxReconnections, _requireMaster, _reconnectionDelay, + _queueTimeout, _operationTimeout, _operationTimeoutCheckPeriod, _defaultUserCredentials, @@ -454,7 +477,7 @@ public ConnectionSettings Build() _maxDiscoverAttempts, _gossipExternalHttpPort, _gossipTimeout, - _preferRandomNode); + _nodePreference); } } } \ No newline at end of file diff --git a/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs b/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs index 3786afc..9c0c2a1 100644 --- a/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs +++ b/src/EventStore.ClientAPI.NetCore/DnsClusterSettingsBuilder.cs @@ -14,7 +14,7 @@ public class DnsClusterSettingsBuilder private int _maxDiscoverAttempts = Consts.DefaultMaxClusterDiscoverAttempts; private int _managerExternalHttpPort = Consts.DefaultClusterManagerExternalHttpPort; private TimeSpan _gossipTimeout = TimeSpan.FromSeconds(1); - private bool _preferRandomNode = false; + private NodePreference _nodePreference = NodePreference.Master; /// /// Sets the DNS name under which cluster nodes are listed. @@ -54,13 +54,33 @@ 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. /// /// A for further configuration. public DnsClusterSettingsBuilder PreferRandomNode() { - _preferRandomNode = true; + _nodePreference = NodePreference.Random; + return this; + } + + /// + /// Whether to prioritize choosing a slave node that's alive from the known nodes. + /// + /// A for further configuration. + public DnsClusterSettingsBuilder PreferSlaveNode() + { + _nodePreference = NodePreference.Slave; return this; } @@ -103,7 +123,7 @@ public ClusterSettings Build() { this._maxDiscoverAttempts, this._managerExternalHttpPort, this._gossipTimeout, - this._preferRandomNode); + this._nodePreference); } } } \ No newline at end of file diff --git a/src/EventStore.ClientAPI.NetCore/EventStore.ClientAPI.NetCore.csproj b/src/EventStore.ClientAPI.NetCore/EventStore.ClientAPI.NetCore.csproj old mode 100755 new mode 100644 index 51ca040..62cba69 --- a/src/EventStore.ClientAPI.NetCore/EventStore.ClientAPI.NetCore.csproj +++ b/src/EventStore.ClientAPI.NetCore/EventStore.ClientAPI.NetCore.csproj @@ -1,32 +1,30 @@ - - - - netstandard2.0 - - - EventStore.ClientAPI - EventStore NetStandard ClientApi - Event Store LLP - Copyright © Event Store LLP. All rights reserved. - EventStore.ClientAPI - 4.0.2-alpha - - - The client API for the Event Store. Get the open source or commercial versions of the Event Store server from http://geteventstore.com - Event Store LLP - portable - EventStore.ClientAPI.NetCore - EventStore.ClientAPI.NetCore - eventstore;client;netstandard2.0 - http://geteventstore.com/assets/ouro-200.png - https://github.com/EventStore/netcore-client/blob/master/LICENSE.md - git - https://github.com/EventStore/ClientAPI.NetCore - - - - - - - + + + netstandard2.0 + EventStore.ClientAPI + + + EventStore.ClientAPI + EventStore NetStandard ClientApi + Event Store LLP + Copyright © Event Store LLP. All rights reserved. + EventStore.ClientAPI + 4.1.1-rc + + + The client API for the Event Store. Get the open source or commercial versions of the Event Store server from http://geteventstore.com + Event Store LLP + portable + EventStore.ClientAPI.NetCore + EventStore.ClientAPI.NetCore + eventstore;client;netstandard2.0 + http://geteventstore.com/assets/ouro-200.png + https://github.com/EventStore/netcore-client/blob/master/LICENSE.md + git + https://github.com/EventStore/ClientAPI.NetCore + + + + + 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 28867be..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, @@ -101,12 +100,12 @@ public static IEventStoreConnection Create(ConnectionSettings connectionSettings connectionSettings.ClientConnectionTimeout, connectionSettings.ClusterDns, connectionSettings.GossipSeeds, connectionSettings.MaxDiscoverAttempts, connectionSettings.ExternalGossipPort, connectionSettings.GossipTimeout, - connectionSettings.PreferRandomNode); + connectionSettings.NodePreference); } if (scheme == "discover") { var clusterSettings = new ClusterSettings(uri.Host, connectionSettings.MaxDiscoverAttempts, uri.Port, - connectionSettings.GossipTimeout, connectionSettings.PreferRandomNode); + connectionSettings.GossipTimeout, connectionSettings.NodePreference); Ensure.NotNull(connectionSettings, "connectionSettings"); Ensure.NotNull(clusterSettings, "clusterSettings"); @@ -116,7 +115,7 @@ public static IEventStoreConnection Create(ConnectionSettings connectionSettings clusterSettings.ExternalGossipPort, clusterSettings.GossipSeeds, clusterSettings.GossipTimeout, - clusterSettings.PreferRandomNode); + clusterSettings.NodePreference); return new EventStoreNodeConnection(connectionSettings, clusterSettings, endPointDiscoverer, connectionName); @@ -134,7 +133,7 @@ public static IEventStoreConnection Create(ConnectionSettings connectionSettings var clusterSettings = new ClusterSettings(connectionSettings.GossipSeeds, connectionSettings.MaxDiscoverAttempts, connectionSettings.GossipTimeout, - connectionSettings.PreferRandomNode); + connectionSettings.NodePreference); Ensure.NotNull(connectionSettings, "connectionSettings"); Ensure.NotNull(clusterSettings, "clusterSettings"); @@ -144,7 +143,7 @@ public static IEventStoreConnection Create(ConnectionSettings connectionSettings clusterSettings.ExternalGossipPort, clusterSettings.GossipSeeds, clusterSettings.GossipTimeout, - clusterSettings.PreferRandomNode); + clusterSettings.NodePreference); return new EventStoreNodeConnection(connectionSettings, clusterSettings, endPointDiscoverer, connectionName); @@ -210,9 +209,23 @@ public static IEventStoreConnection Create(ConnectionSettings connectionSettings clusterSettings.ExternalGossipPort, clusterSettings.GossipSeeds, clusterSettings.GossipTimeout, - clusterSettings.PreferRandomNode); + clusterSettings.NodePreference); 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 450547b..32f96d2 100644 --- a/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs +++ b/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs @@ -13,7 +13,7 @@ public class GossipSeedClusterSettingsBuilder private GossipSeed[] _gossipSeeds; private TimeSpan _gossipTimeout = TimeSpan.FromSeconds(1); private int _maxDiscoverAttempts = Consts.DefaultMaxClusterDiscoverAttempts; - private bool _preferRandomNode = false; + private NodePreference _nodePreference = NodePreference.Master; /// /// Sets gossip seed endpoints for the client. @@ -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. @@ -82,7 +92,17 @@ public GossipSeedClusterSettingsBuilder SetGossipTimeout(TimeSpan timeout) /// A for further configuration. public GossipSeedClusterSettingsBuilder PreferRandomNode() { - _preferRandomNode = true; + _nodePreference = NodePreference.Random; + return this; + } + + /// + /// hether to prioritize choosing a slave node that's alive from the known nodes. + /// + /// A for further configuration. + public GossipSeedClusterSettingsBuilder PreferSlaveNode() + { + _nodePreference = NodePreference.Slave; return this; } @@ -100,7 +120,7 @@ public static implicit operator ClusterSettings(GossipSeedClusterSettingsBuilder /// Builds a object from a . /// public ClusterSettings Build() { - return new ClusterSettings(this._gossipSeeds, this._maxDiscoverAttempts, this._gossipTimeout, this._preferRandomNode); + return new ClusterSettings(this._gossipSeeds, this._maxDiscoverAttempts, this._gossipTimeout, this._nodePreference); } } } 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 7df33d9..f227ffb 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/ClusterDnsEndPointDiscoverer.cs +++ b/src/EventStore.ClientAPI.NetCore/Internal/ClusterDnsEndPointDiscoverer.cs @@ -23,7 +23,7 @@ internal class ClusterDnsEndPointDiscoverer: IEndPointDiscoverer private readonly HttpAsyncClient _client; private ClusterMessages.MemberInfoDto[] _oldGossip; private TimeSpan _gossipTimeout; - private readonly bool _preferRandomNode; + private readonly NodePreference _nodePreference; public ClusterDnsEndPointDiscoverer(ILogger log, string clusterDns, @@ -31,7 +31,7 @@ public ClusterDnsEndPointDiscoverer(ILogger log, int managerExternalHttpPort, GossipSeed[] gossipSeeds, TimeSpan gossipTimeout, - bool preferRandomNode) + NodePreference nodePreference) { Ensure.NotNull(log, "log"); @@ -41,12 +41,15 @@ public ClusterDnsEndPointDiscoverer(ILogger log, _managerExternalHttpPort = managerExternalHttpPort; _gossipSeeds = gossipSeeds; _gossipTimeout = gossipTimeout; - _preferRandomNode = preferRandomNode; + _nodePreference = nodePreference; _client = new HttpAsyncClient(_gossipTimeout); } 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)); } @@ -83,7 +87,7 @@ public async Task DiscoverAsync(IPEndPoint failedTcpEndPoint ) if (gossip == null || gossip.Members == null || gossip.Members.Length == 0) continue; - var bestNode = TryDetermineBestNode(gossip.Members, _preferRandomNode); + var bestNode = TryDetermineBestNode(gossip.Members, _nodePreference); if (bestNode != null) { _oldGossip = gossip.Members; @@ -210,7 +214,7 @@ private ClusterMessages.ClusterInfoDto TryGetGossipFrom(GossipSeed endPoint) return result; } - private NodeEndPoints? TryDetermineBestNode(IEnumerable members, bool preferRandomNode) + private NodeEndPoints? TryDetermineBestNode(IEnumerable members, NodePreference nodePreference) { var notAllowedStates = new[] { @@ -222,14 +226,21 @@ private ClusterMessages.ClusterInfoDto TryGetGossipFrom(GossipSeed endPoint) .Where(x => !notAllowedStates.Contains(x.State)) .OrderByDescending(x => x.State) .ToArray(); - if (preferRandomNode) + + switch (nodePreference) { - RandomShuffle(nodes, 0, nodes.Length - 1); + case NodePreference.Random: + RandomShuffle(nodes, 0, nodes.Length - 1); + 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; } var node = nodes.FirstOrDefault(); - if (node == null) + if (node == default(ClusterMessages.MemberInfoDto)) { //_log.Info("Unable to locate suitable node. Gossip info:\n{0}.", string.Join("\n", members.Select(x => x.ToString()))); return null; @@ -243,4 +254,4 @@ private ClusterMessages.ClusterInfoDto TryGetGossipFrom(GossipSeed endPoint) return new NodeEndPoints(normTcp, secTcp); } } -} +} \ No newline at end of file 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/NodePreference.cs b/src/EventStore.ClientAPI.NetCore/NodePreference.cs new file mode 100644 index 0000000..a7c5544 --- /dev/null +++ b/src/EventStore.ClientAPI.NetCore/NodePreference.cs @@ -0,0 +1,25 @@ +using System; +using System.Collections.Generic; +using System.Text; + +namespace EventStore.ClientAPI +{ + /// + /// Indicates which order of preferred nodes for connecting to. + /// + public enum NodePreference + { + /// + /// When attempting connnection, prefers master node. + /// + Master, + /// + /// When attempting connnection, prefers slave node. + /// + Slave, + /// + /// When attempting connnection, has no node preference. + /// + Random + } +} diff --git a/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs b/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs index e98b758..65657f0 100644 --- a/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs +++ b/src/EventStore.ClientAPI.NetCore/Projections/QueryManager.cs @@ -1,15 +1,15 @@ +using System; +using System.Net; +using System.Threading.Tasks; using EventStore.ClientAPI.Common.Utils; using EventStore.ClientAPI.Common.Utils.Threading; using EventStore.ClientAPI.SystemData; using Newtonsoft.Json.Linq; -using System; -using System.Net; -using System.Threading.Tasks; 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 @@ -21,38 +21,13 @@ public class QueryManager /// Creates a new instance of . /// /// An instance of to use for logging. - /// HTTP IP endpoint of an Event Store server. + /// HTTP endpoint of an Event Store server. /// Timeout of projection API operations /// Timeout of query execution - public QueryManager(ILogger log, IPEndPoint ipHttpEndPoint, TimeSpan projectionOperationTimeout, TimeSpan queryTimeout) + public QueryManager(ILogger log, IPEndPoint httpEndPoint, TimeSpan projectionOperationTimeout, TimeSpan queryTimeout) { _queryTimeout = queryTimeout; - _projectionsManager = new ProjectionsManager(log, ipHttpEndPoint, projectionOperationTimeout); - } - - /// - /// Creates a new instance of . - /// - /// An instance of to use for logging. - /// HTTP DNS endpoint of an Event Store server. - /// Timeout of projection API operations - /// Timeout of query execution - public QueryManager(ILogger log, DnsEndPoint dnsHttpEndPoint, TimeSpan projectionOperationTimeout, TimeSpan queryTimeout) - { - _queryTimeout = queryTimeout; - _projectionsManager = new ProjectionsManager(log, dnsHttpEndPoint, projectionOperationTimeout); - } - - /// - /// Creates a new instance of . - /// - /// An instance of to use for logging. - /// HTTP URI endpoint of an Event Store server. - /// Timeout of projection API operations - /// Timeout of query execution - public QueryManager(ILogger log, Uri uri, TimeSpan projectionOperationTimeout, TimeSpan queryTimeout) - : this(log, new DnsEndPoint(uri.Host, uri.Port), projectionOperationTimeout, queryTimeout) - { + _projectionsManager = new ProjectionsManager(log, httpEndPoint, projectionOperationTimeout); } /// @@ -105,4 +80,4 @@ private async Task GetStatusAsync(string name, UserCredentials userCrede return projectionStatus.ParseJson()["status"].ToString(); } } -} \ No newline at end of file +} 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/EventStore.ClientAPI.NetCore.Tests.csproj b/test/EventStore.ClientAPI.NetCore.Tests/EventStore.ClientAPI.NetCore.Tests.csproj index a80677b..b0c215c 100755 --- a/test/EventStore.ClientAPI.NetCore.Tests/EventStore.ClientAPI.NetCore.Tests.csproj +++ b/test/EventStore.ClientAPI.NetCore.Tests/EventStore.ClientAPI.NetCore.Tests.csproj @@ -21,10 +21,10 @@ - - - - + + + + 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;