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/appveyor.yml b/appveyor.yml index 14b2919..44eac44 100644 --- a/appveyor.yml +++ b/appveyor.yml @@ -1,4 +1,4 @@ -version: 4.0.2.{build} +version: 4.1.0.{build} image: Visual Studio 2017 @@ -22,9 +22,6 @@ build: publish_nuget: true verbosity: minimal -test_script: - - ps: .\scripts\appveyortest.ps1 - deploy: provider: NuGet on: diff --git a/scripts/installandruneventstore.ps1 b/scripts/installandruneventstore.ps1 index 7ede819..65f2a2d 100644 --- a/scripts/installandruneventstore.ps1 +++ b/scripts/installandruneventstore.ps1 @@ -1,3 +1,3 @@ -(New-Object Net.WebClient).DownloadFile('https://eventstore.org/downloads/EventStore-OSS-Win-v4.0.3.zip','c:\tools\eventstore.zip') +(New-Object Net.WebClient).DownloadFile('https://eventstore.org/downloads/EventStore-OSS-Win-v4.1.0.zip','c:\tools\eventstore.zip') 7z e C:\tools\eventstore.zip -oC:\tools\EventStore start-process -NoNewWindow "C:\tools\EventStore\EventStore.ClusterNode.exe" "--memdb --run-projections ALL" \ No newline at end of file diff --git a/src/EventStore.ClientAPI.NetCore/CatchUpSubscriptionSettings.cs b/src/EventStore.ClientAPI.NetCore/CatchUpSubscriptionSettings.cs index 50ec220..d6cc329 100644 --- a/src/EventStore.ClientAPI.NetCore/CatchUpSubscriptionSettings.cs +++ b/src/EventStore.ClientAPI.NetCore/CatchUpSubscriptionSettings.cs @@ -3,37 +3,37 @@ namespace EventStore.ClientAPI { /// - /// Settings for + /// Settings for . /// public class CatchUpSubscriptionSettings { /// - /// The maximum amount to cache when processing from live subscription. Going above will drop subscription. + /// The maximum amount of events to cache when processing from a live subscription. Going above this value will drop the subscription. /// public readonly int MaxLiveQueueSize; /// - /// The number of events to read per batch when reading history + /// The number of events to read per batch when reading the history. /// public readonly int ReadBatchSize; /// - /// Enables verbose logging on the subscription + /// Enables verbose logging on the subscription. /// public readonly bool VerboseLogging; /// - /// Whether or not to resolve link events + /// Whether to resolve link events. /// public readonly bool ResolveLinkTos; /// - /// The name of subscription. + /// The name of the subscription. /// public readonly string SubscriptionName; /// - /// Returns default settings + /// Returns default settings. /// public static readonly CatchUpSubscriptionSettings Default = new CatchUpSubscriptionSettings( Consts.CatchUpDefaultMaxPushQueueSize, @@ -42,19 +42,18 @@ public class CatchUpSubscriptionSettings resolveLinkTos: true, subscriptionName: String.Empty ); - /// - /// Constructs a object + /// Constructs a object. /// - /// The maximum amount to buffer when processing from live subscription. Going above will drop subscription. - /// The number of events to read per batch when reading history - /// Enables verbose logging on the subscription - /// Whether or not to resolve link events - /// The name of subscription. + /// The maximum amount of events to buffer when processing from a live subscription. Going above this amount will drop the subscription. + /// The number of events to read per batch when reading through history. + /// Enables verbose logging on the subscription. + /// Whether to resolve link events. + /// The name of the subscription. public CatchUpSubscriptionSettings(int maxLiveQueueSize, int readBatchSize, bool verboseLogging, bool resolveLinkTos, string subscriptionName = "") { Ensure.Positive(readBatchSize, "readBatchSize"); Ensure.Positive(maxLiveQueueSize, "maxLiveQueueSize"); - if (readBatchSize > Consts.MaxReadSize) throw new ArgumentException(string.Format("Read batch size should be less than {0}. For larger reads you should page.", Consts.MaxReadSize)); + if (readBatchSize > ClientApiConstants.MaxReadSize) throw new ArgumentException(string.Format("Read batch size should be less than {0}. For larger reads you should page.", ClientApiConstants.MaxReadSize)); MaxLiveQueueSize = maxLiveQueueSize; ReadBatchSize = readBatchSize; VerboseLogging = verboseLogging; diff --git a/src/EventStore.ClientAPI.NetCore/ClientApiConstants.cs b/src/EventStore.ClientAPI.NetCore/ClientApiConstants.cs new file mode 100644 index 0000000..86e13a5 --- /dev/null +++ b/src/EventStore.ClientAPI.NetCore/ClientApiConstants.cs @@ -0,0 +1,13 @@ +namespace EventStore.ClientAPI +{ + /// + /// Various constant values that may be useful when working with the ClientAPI. + /// + public static class ClientApiConstants + { + /// + /// The maximum number of events that can be read in a single operation. + /// + public static readonly int MaxReadSize = Consts.MaxReadSize; + } +} \ No newline at end of file diff --git a/src/EventStore.ClientAPI.NetCore/ClientOperations/ConnectToPersistentSubscriptionOperation.cs b/src/EventStore.ClientAPI.NetCore/ClientOperations/ConnectToPersistentSubscriptionOperation.cs index 1a6008d..dd4ea08 100644 --- a/src/EventStore.ClientAPI.NetCore/ClientOperations/ConnectToPersistentSubscriptionOperation.cs +++ b/src/EventStore.ClientAPI.NetCore/ClientOperations/ConnectToPersistentSubscriptionOperation.cs @@ -9,13 +9,13 @@ namespace EventStore.ClientAPI.ClientOperations { - internal class ConnectToPersistentSubscriptionOperation : SubscriptionOperation, IConnectToPersistentSubscriptions + internal class ConnectToPersistentSubscriptionOperation : SubscriptionOperation, IConnectToPersistentSubscriptions { private readonly string _groupName; private readonly int _bufferSize; private string _subscriptionId; - public ConnectToPersistentSubscriptionOperation(ILogger log, TaskCompletionSource source, string groupName, int bufferSize, string streamId, UserCredentials userCredentials, Func eventAppeared, Action subscriptionDropped, bool verboseLogging, Func getConnection) + public ConnectToPersistentSubscriptionOperation(ILogger log, TaskCompletionSource source, string groupName, int bufferSize, string streamId, UserCredentials userCredentials, Func eventAppeared, Action subscriptionDropped, bool verboseLogging, Func getConnection) : base(log, source, streamId, false, userCredentials, eventAppeared, subscriptionDropped, verboseLogging, getConnection) { _groupName = groupName; @@ -46,7 +46,7 @@ protected override bool InspectPackage(TcpPackage package, out InspectionResult if (package.Command == TcpCommand.PersistentSubscriptionStreamEventAppeared) { var dto = package.Data.Deserialize(); - EventAppeared(new ResolvedEvent(dto.Event)); + EventAppeared(new PersistentSubscriptionResolvedEvent(new ResolvedEvent(dto.Event), dto.RetryCount)); result = new InspectionResult(InspectionDecision.DoNothing, "StreamEventAppeared"); return true; } diff --git a/src/EventStore.ClientAPI.NetCore/ClientOperations/SubscriptionOperation.cs b/src/EventStore.ClientAPI.NetCore/ClientOperations/SubscriptionOperation.cs index 38bc5cd..2e6ea5b 100644 --- a/src/EventStore.ClientAPI.NetCore/ClientOperations/SubscriptionOperation.cs +++ b/src/EventStore.ClientAPI.NetCore/ClientOperations/SubscriptionOperation.cs @@ -10,14 +10,21 @@ namespace EventStore.ClientAPI.ClientOperations { - internal abstract class SubscriptionOperation : ISubscriptionOperation where T : EventStoreSubscription + internal interface IResolvedEvent + { + string OriginalStreamId{ get; } + long OriginalEventNumber { get; } + RecordedEvent OriginalEvent { get; } + Position? OriginalPosition { get; } + } + internal abstract class SubscriptionOperation : ISubscriptionOperation where T : EventStoreSubscription where TE: IResolvedEvent { private readonly ILogger _log; private readonly TaskCompletionSource _source; protected readonly string _streamId; protected readonly bool _resolveLinkTos; protected readonly UserCredentials _userCredentials; - protected readonly Func _eventAppeared; + protected readonly Func _eventAppeared; private readonly Action _subscriptionDropped; private readonly bool _verboseLogging; protected readonly Func _getConnection; @@ -33,7 +40,7 @@ protected SubscriptionOperation(ILogger log, string streamId, bool resolveLinkTos, UserCredentials userCredentials, - Func eventAppeared, + Func eventAppeared, Action subscriptionDropped, bool verboseLogging, Func getConnection) @@ -84,7 +91,6 @@ private TcpPackage CreateUnsubscriptionPackage() } protected abstract bool InspectPackage(TcpPackage package, out InspectionResult result); - public InspectionResult InspectPackage(TcpPackage package) { try @@ -97,13 +103,6 @@ public InspectionResult InspectPackage(TcpPackage package) switch (package.Command) { - case TcpCommand.StreamEventAppeared: - { - var dto = package.Data.Deserialize(); - EventAppeared(new ResolvedEvent(dto.Event)); - return new InspectionResult(InspectionDecision.DoNothing, "StreamEventAppeared"); - } - case TcpCommand.SubscriptionDropped: { var dto = package.Data.Deserialize(); @@ -240,7 +239,7 @@ protected void ConfirmSubscription(long lastCommitPosition, long? lastEventNumbe protected abstract T CreateSubscriptionObject(long lastCommitPosition, long? lastEventNumber); - protected void EventAppeared(ResolvedEvent e) + protected void EventAppeared(TE e) { if (_unsubscribed != 0) return; diff --git a/src/EventStore.ClientAPI.NetCore/ClientOperations/VolatileSubscriptionOperation.cs b/src/EventStore.ClientAPI.NetCore/ClientOperations/VolatileSubscriptionOperation.cs index 179e620..361733d 100644 --- a/src/EventStore.ClientAPI.NetCore/ClientOperations/VolatileSubscriptionOperation.cs +++ b/src/EventStore.ClientAPI.NetCore/ClientOperations/VolatileSubscriptionOperation.cs @@ -7,7 +7,7 @@ namespace EventStore.ClientAPI.ClientOperations { - internal class VolatileSubscriptionOperation : SubscriptionOperation + internal class VolatileSubscriptionOperation : SubscriptionOperation { public VolatileSubscriptionOperation(ILogger log, TaskCompletionSource source, string streamId, bool resolveLinkTos, UserCredentials userCredentials, Func eventAppeared, Action subscriptionDropped, bool verboseLogging, Func getConnection) : base(log, source, streamId, resolveLinkTos, userCredentials, eventAppeared, subscriptionDropped, verboseLogging, getConnection) diff --git a/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs b/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs index 9566aca..c503d3b 100644 --- a/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs +++ b/src/EventStore.ClientAPI.NetCore/ClusterSettings.cs @@ -43,34 +43,34 @@ public static ClusterSettingsBuilder Create() /// /// Prefer a randomly selected node. /// - public bool PreferRandomNode; + public NodePreference NodePreference; /// - /// Used if we're connecting with gossip seeds + /// Used if connecting with gossip seeds. /// - /// 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) + /// Endpoints for seeding gossip. + /// Maximum number of attempts to discover the cluster. + /// Timeout for cluster gossip. + /// 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; } /// - /// Used if we're discovering via DNS + /// Used if discovering via DNS. /// - /// The DNS name to use for discovering endpoints - /// 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) + /// The DNS name to use for discovering endpoints. + /// 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 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 56c0c11..5397411 100644 --- a/src/EventStore.ClientAPI.NetCore/ConnectionSettings.cs +++ b/src/EventStore.ClientAPI.NetCore/ConnectionSettings.cs @@ -6,7 +6,7 @@ namespace EventStore.ClientAPI { /// /// A object is an immutable representation of the settings for an - /// . A object can be built using + /// . You can build a object using /// a , either via the method, or via /// the constructor of . /// @@ -15,53 +15,57 @@ public sealed class ConnectionSettings private static readonly Lazy DefaultSettings = new Lazy(() => Create(), true); /// - /// The default + /// The default . /// public static ConnectionSettings Default { get { return DefaultSettings.Value; } } /// - /// Creates a new set of + /// Creates a new set of . /// - /// A that can be used to build up a + /// A you can use to build up a . public static ConnectionSettingsBuilder Create() { return new ConnectionSettingsBuilder(); } /// - /// The that this connection will use + /// The that this connection will use. /// public readonly ILogger Log; /// - /// Whether or not do excessive logging of internal logic. + /// Whether to use excessive logging of internal logic. /// public readonly bool VerboseLogging; /// - /// The maximum number of outstanding items allowed in the queue + /// The maximum number of outstanding items allowed in the queue. /// public readonly int MaxQueueSize; /// - /// The maximum number of allowed asynchronous operations to be in process + /// The maximum number of allowed asynchronous operations to be in process. /// public readonly int MaxConcurrentItems; /// - /// The maximum number of retry attempts + /// The maximum number of retry attempts. /// public readonly int MaxRetries; /// - /// The maximum number of times to allow for reconnection + /// The maximum number of times to allow for reconnection. /// public readonly int MaxReconnections; /// - /// Whether or not to require EventStore to refuse serving read or write request if it is not master + /// Whether to require EventStore to refuse serving read or write request if it is not master. /// public readonly bool RequireMaster; /// - /// The amount of time to delay before attempting to reconnect + /// The amount of time to delay before attempting to reconnect. /// public readonly TimeSpan ReconnectionDelay; /// - /// The amount of time before an operation is considered to have timed out + /// 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; /// @@ -73,7 +77,7 @@ public static ConnectionSettingsBuilder Create() /// public readonly UserCredentials DefaultUserCredentials; /// - /// Whether or not the connection is encrypted using SSL. + /// Whether the connection is encrypted using SSL. /// public readonly bool UseSslConnection; /// @@ -81,12 +85,12 @@ public static ConnectionSettingsBuilder Create() /// public readonly string TargetHost; /// - /// Whether or not to validate the server SSL certificate. + /// Whether to validate the server SSL certificate. /// public readonly bool ValidateServer; /// - /// Whether or not to raise an error if no response is received from the server for an operation. + /// Whether to raise an error if no response is received from the server for an operation. /// public readonly bool FailOnNoServerResponse; /// @@ -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 490b598..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() @@ -48,7 +49,7 @@ internal ConnectionSettingsBuilder() } /// - /// Configures the connection to output log messages to the given . + /// Configures the connection to output log messages to the given . You should implement this interface using another library such as NLog or log4net. /// /// The to use. /// @@ -89,7 +90,7 @@ public ConnectionSettingsBuilder UseFileLogger(string filename) } /// - /// Turns on verbose internal logic logging. + /// Turns on verbose internal logic logging. By contains default information about connection, disconnection and errors, but you can customize output. /// /// public ConnectionSettingsBuilder EnableVerboseLogging() @@ -99,7 +100,7 @@ public ConnectionSettingsBuilder EnableVerboseLogging() } /// - /// Sets the limit for number of outstanding operations + /// Sets the limit for number of outstanding operations. /// /// The new limit of outstanding operations /// @@ -112,7 +113,7 @@ public ConnectionSettingsBuilder LimitOperationsQueueTo(int limit) } /// - /// Limits the number of concurrent operations that this connection can have + /// Limits the number of concurrent operations that this connection can have. /// /// /// @@ -125,7 +126,7 @@ public ConnectionSettingsBuilder LimitConcurrentOperationsTo(int limit) } /// - /// Limits the number of operation attempts + /// Limits the number of operation attempts. /// /// /// @@ -138,7 +139,7 @@ public ConnectionSettingsBuilder LimitAttemptsForOperationTo(int limit) } /// - /// Limits the number of operation retries + /// Limits the number of operation retries. /// /// /// @@ -151,7 +152,7 @@ public ConnectionSettingsBuilder LimitRetriesForOperationTo(int limit) } /// - /// Allows infinite operation retry attempts + /// Allows infinite operation retry attempts. /// /// public ConnectionSettingsBuilder KeepRetrying() @@ -161,7 +162,7 @@ public ConnectionSettingsBuilder KeepRetrying() } /// - /// Limits the number of reconnections this connection can try to make + /// Limits the number of reconnections this connection can try to make. /// /// /// @@ -174,7 +175,7 @@ public ConnectionSettingsBuilder LimitReconnectionsTo(int limit) } /// - /// Allows infinite reconnection attempts + /// Allows infinite reconnection attempts. /// /// public ConnectionSettingsBuilder KeepReconnecting() @@ -184,7 +185,7 @@ public ConnectionSettingsBuilder KeepReconnecting() } /// - /// Requires all write and read requests to be served only by master (cluster version only) + /// Requires all write and read requests to be served only by master (cluster version only). /// /// public ConnectionSettingsBuilder PerformOnMasterOnly() @@ -194,7 +195,7 @@ public ConnectionSettingsBuilder PerformOnMasterOnly() } /// - /// Allow for writes to be forwarded and read requests served locally if node is not master (cluster version only) + /// Allow for writes to be forwarded and read requests served locally if node is not master (cluster version only). /// /// public ConnectionSettingsBuilder PerformOnAnyNode() @@ -204,7 +205,7 @@ public ConnectionSettingsBuilder PerformOnAnyNode() } /// - /// Sets the delay between reconnection attempts + /// Sets the delay between reconnection attempts. /// /// /// @@ -215,7 +216,18 @@ public ConnectionSettingsBuilder SetReconnectionDelayTo(TimeSpan reconnectionDel } /// - /// Sets the operation timeout duration + /// 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. /// /// /// @@ -237,7 +249,7 @@ public ConnectionSettingsBuilder SetTimeoutCheckPeriodTo(TimeSpan timeoutCheckPe } /// - /// Sets the default to be used for this connection. + /// Sets the default used for this connection. /// If user credentials are not given for an operation, these credentials will be used. /// /// @@ -264,7 +276,7 @@ public ConnectionSettingsBuilder UseSslConnection(string targetHost, bool valida } /// - /// Marks that no response from server should cause an error on the request + /// Marks that no response from server should cause an error on the request. /// /// public ConnectionSettingsBuilder FailOnNoServerResponse() @@ -274,7 +286,7 @@ public ConnectionSettingsBuilder FailOnNoServerResponse() } /// - /// Sets how often heartbeats should be expected on the connection (lower values detect broken sockets faster) + /// Sets how often heartbeats should be expected on the connection (lower values detect broken sockets faster). /// /// /// @@ -285,7 +297,7 @@ public ConnectionSettingsBuilder SetHeartbeatInterval(TimeSpan interval) } /// - /// Sets how long to wait without heartbeats before determining a connection to be dead (must be longer than heartbeat interval) + /// Sets how long to wait without heartbeats before determining a connection to be dead (must be longer than heartbeat interval). /// /// /// @@ -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; } @@ -378,10 +400,11 @@ public ConnectionSettingsBuilder SetClusterGossipPort(int clusterGossipPort) /// /// Sets gossip seed endpoints for the client. /// - /// Note that this should be the external HTTP endpoint of the server, as it is required - /// for the client to exchange gossip with the server. The standard port which should be - /// used here is 2113. - /// + /// + /// This should be the external HTTP endpoint of the server, as it is required + /// for the client to exchange gossip with the server. The standard port is 2113. + /// + /// /// If the server requires a specific Host header to be sent as part of the gossip /// request, use the overload of this method taking instead. /// @@ -438,6 +461,7 @@ public ConnectionSettings Build() _maxReconnections, _requireMaster, _reconnectionDelay, + _queueTimeout, _operationTimeout, _operationTimeoutCheckPeriod, _defaultUserCredentials, @@ -453,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/EventReadResult.cs b/src/EventStore.ClientAPI.NetCore/EventReadResult.cs index 7cca733..3d1c095 100644 --- a/src/EventStore.ClientAPI.NetCore/EventReadResult.cs +++ b/src/EventStore.ClientAPI.NetCore/EventReadResult.cs @@ -4,17 +4,17 @@ namespace EventStore.ClientAPI { /// - /// A Event Read Result is the result of a single event read operation to the event store. + /// A event read result is the result of a single event read operation to Event Store. /// public class EventReadResult { /// - /// The representing the status of this read attempt + /// The representing the status of this read attempt. /// public readonly EventReadStatus Status; /// - /// The name of the stream read + /// The name of the stream read. /// public readonly string Stream; @@ -24,7 +24,7 @@ public class EventReadResult public readonly long EventNumber; /// - /// The event read represented as + /// The event read represented as . /// public readonly ResolvedEvent? Event; 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 351f655..98fac26 100644 --- a/src/EventStore.ClientAPI.NetCore/EventStoreCatchUpSubscription.cs +++ b/src/EventStore.ClientAPI.NetCore/EventStoreCatchUpSubscription.cs @@ -136,9 +136,9 @@ internal Task StartAsync() } /// - /// Attempts to stop the subscription. + /// Attempts to stop the subscription blocking for completion of stop. /// - /// The amount of time within which the subscription should stop. + /// The maximum amount of time which the current thread will block waiting for the subscription to stop before throwing a TimeoutException. /// Thrown if the subscription fails to stop within it's timeout period. public void Stop(TimeSpan timeout) { @@ -430,7 +430,7 @@ private async Task ReadEventsInternalAsync(IEventStoreConnection connection, boo private async Task ReadEventsCallbackAsync(AllEventsSlice slice, long? lastCommitPosition) { bool shouldStopOrDone = ShouldStop || await ProcessEventsAsync(lastCommitPosition, slice).ConfigureAwait(false); - if (shouldStopOrDone && Verbose) + if(shouldStopOrDone && Verbose) { Log.Debug( "Catch-up Subscription {0} to {1}: finished reading events, nextReadPosition = {2}.", @@ -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; } @@ -534,7 +534,7 @@ internal EventStoreStreamCatchUpSubscription(IEventStoreConnection connection, protected override Task ReadEventsTillAsync(IEventStoreConnection connection, bool resolveLinkTos, UserCredentials userCredentials, long? lastCommitPosition, long? lastEventNumber) => ReadEventsInternalAsync(connection, resolveLinkTos, userCredentials, lastEventNumber); - + private async Task ReadEventsInternalAsync(IEventStoreConnection connection, bool resolveLinkTos, UserCredentials userCredentials, long? lastEventNumber) { @@ -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 58d7601..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 @@ -37,7 +36,7 @@ public static IEventStoreConnection Create(string connectionString, string conne } /// - /// Creates a new to single node using the overriding them with values provided via the connectionstring + /// Creates a new to single node using default provided via a connectionstring /// /// Optional name of connection (will be generated automatically, if not provided) /// Pre-populated settings builder, optional. If not specified, a new builder will be created. @@ -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/EventStorePersistentSubscription.cs b/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscription.cs index 001e1e8..701c0e2 100644 --- a/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscription.cs +++ b/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscription.cs @@ -1,10 +1,48 @@ using System; using System.Threading.Tasks; +using EventStore.ClientAPI.ClientOperations; using EventStore.ClientAPI.Internal; using EventStore.ClientAPI.SystemData; namespace EventStore.ClientAPI { + internal struct PersistentSubscriptionResolvedEvent : IResolvedEvent + { + /// + /// + /// + public readonly int? RetryCount; + /// + /// + /// + public readonly ResolvedEvent Event; + + /// + /// + /// + /// + /// + internal PersistentSubscriptionResolvedEvent(ResolvedEvent @event, int? retryCount) + { + Event = @event; + RetryCount = retryCount; + } + + string IResolvedEvent.OriginalStreamId => Event.OriginalStreamId; + + long IResolvedEvent.OriginalEventNumber => Event.OriginalEventNumber; + + RecordedEvent IResolvedEvent.OriginalEvent => Event.OriginalEvent; + + Position? IResolvedEvent.OriginalPosition => Event.OriginalPosition; + + /// + /// + /// + /// + /// + public static implicit operator ResolvedEvent(PersistentSubscriptionResolvedEvent x) => x.Event; + } /// /// Represents a persistent subscription connection. /// @@ -14,7 +52,7 @@ public class EventStorePersistentSubscription : EventStorePersistentSubscription internal EventStorePersistentSubscription( string subscriptionId, string streamId, - Func eventAppeared, + Func eventAppeared, Action subscriptionDropped, UserCredentials userCredentials, ILogger log, bool verboseLogging, ConnectionSettings settings, EventStoreConnectionLogicHandler handler, int bufferSize = 10, bool autoAck = true) @@ -27,7 +65,7 @@ internal EventStorePersistentSubscription( internal override Task StartSubscription( string subscriptionId, string streamId, int bufferSize, UserCredentials userCredentials, - Func onEventAppeared, + Func onEventAppeared, Action onSubscriptionDropped, ConnectionSettings settings) { var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); diff --git a/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs b/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs index 9297793..0a37376 100644 --- a/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs +++ b/src/EventStore.ClientAPI.NetCore/EventStorePersistentSubscriptionBase.cs @@ -21,7 +21,7 @@ public abstract class EventStorePersistentSubscriptionBase private readonly string _subscriptionId; private readonly string _streamId; - private readonly Func _eventAppeared; + private readonly Func _eventAppeared; private readonly Action _subscriptionDropped; private readonly UserCredentials _userCredentials; private readonly ILogger _log; @@ -30,7 +30,7 @@ public abstract class EventStorePersistentSubscriptionBase private readonly bool _autoAck; private PersistentEventStoreSubscription _subscription; - private readonly ConcurrentQueue _queue = new ConcurrentQueue(); + private readonly ConcurrentQueue _queue = new ConcurrentQueue(); private int _isProcessing; private DropData _dropData; @@ -40,7 +40,7 @@ public abstract class EventStorePersistentSubscriptionBase internal EventStorePersistentSubscriptionBase(string subscriptionId, string streamId, - Func eventAppeared, + Func eventAppeared, Action subscriptionDropped, UserCredentials userCredentials, ILogger log, @@ -77,7 +77,7 @@ internal Task Start() internal abstract Task StartSubscription( string subscriptionId, string streamId, int bufferSize, UserCredentials userCredentials, - Func onEventAppeared, + Func onEventAppeared, Action onSubscriptionDropped, ConnectionSettings settings); @@ -168,7 +168,7 @@ private void EnqueueSubscriptionDropNotification(SubscriptionDropReason reason, var dropData = new DropData(reason, error); if (Interlocked.CompareExchange(ref _dropData, dropData, null) == null) { - Enqueue(DropSubscriptionEvent); + Enqueue(new PersistentSubscriptionResolvedEvent(DropSubscriptionEvent, null)); } } @@ -177,13 +177,13 @@ private void OnSubscriptionDropped(EventStoreSubscription subscription, Subscrip EnqueueSubscriptionDropNotification(reason, exception); } - private Task OnEventAppeared(EventStoreSubscription subscription, ResolvedEvent resolvedEvent) + private Task OnEventAppeared(EventStoreSubscription subscription, PersistentSubscriptionResolvedEvent resolvedEvent) { Enqueue(resolvedEvent); return Task.CompletedTask; } - private void Enqueue(ResolvedEvent resolvedEvent) + private void Enqueue(PersistentSubscriptionResolvedEvent resolvedEvent) { _queue.Enqueue(resolvedEvent); if (Interlocked.CompareExchange(ref _isProcessing, 1, 0) == 0) @@ -197,14 +197,14 @@ private async void ProcessQueue() { if (_subscription == null) { - Thread.Sleep(1); + await Task.Delay(1).ConfigureAwait(false); } else - { - ResolvedEvent e; + { + PersistentSubscriptionResolvedEvent e; while (_queue.TryDequeue(out e)) { - if (e.Equals(DropSubscriptionEvent)) // drop subscription artificial ResolvedEvent + if (e.Event.Equals(DropSubscriptionEvent)) // drop subscription artificial ResolvedEvent { if (_dropData == null) throw new Exception("Drop reason not specified."); DropSubscription(_dropData.Reason, _dropData.Error); @@ -217,13 +217,13 @@ private async void ProcessQueue() } try { - await _eventAppeared(this, e).ConfigureAwait(false); + await _eventAppeared(this, e, e.RetryCount).ConfigureAwait(false); if (_autoAck) - _subscription.NotifyEventsProcessed(new[] { e.OriginalEvent.EventId }); + _subscription.NotifyEventsProcessed(new[] { e.Event.OriginalEvent.EventId }); if (_verbose) _log.Debug("Persistent Subscription to {0}: processed event ({1}, {2}, {3} @ {4}).", _streamId, - e.OriginalEvent.EventStreamId, e.OriginalEvent.EventNumber, e.OriginalEvent.EventType, e.OriginalEventNumber); + e.Event.OriginalEvent.EventStreamId, e.Event.OriginalEvent.EventNumber, e.Event.OriginalEvent.EventType, e.Event.OriginalEventNumber); } catch (Exception exc) { 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 9d1b95d..32f96d2 100644 --- a/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs +++ b/src/EventStore.ClientAPI.NetCore/GossipSeedClusterSettingsBuilder.cs @@ -13,14 +13,13 @@ 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. - /// - /// Note that this should be the external HTTP endpoint of the server, as it is required - /// for the client to exchange gossip with the server. The standard port which should be - /// used here is 2113. + /// TODO: This was a note. + /// This should be the external HTTP endpoint of the server, as it is required + /// for the client to exchange gossip with the server. The standard port is 2113. /// /// If the server requires a specific Host header to be sent as part of the gossip /// request, use the overload of this method taking instead. @@ -51,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. @@ -83,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; } @@ -101,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/IEventStoreConnection.cs b/src/EventStore.ClientAPI.NetCore/IEventStoreConnection.cs index e191a90..f6cbd47 100644 --- a/src/EventStore.ClientAPI.NetCore/IEventStoreConnection.cs +++ b/src/EventStore.ClientAPI.NetCore/IEventStoreConnection.cs @@ -7,40 +7,40 @@ namespace EventStore.ClientAPI { /// - /// Maintains a full duplex connection to the EventStore + /// Maintains a full duplex connection to Event Store. /// /// - /// An operates quite differently than say a SqlConnection. Normally + /// An operates differently than a SqlConnection. Normally /// when using an you want to keep the connection open for a much longer of time than /// when you use a SqlConnection. If you prefer the usage pattern of using(new Connection()) .. then you would likely /// want to create a FlyWeight on top of the . /// /// Another difference is that with the all operations are handled in a full async manner /// (even if you call the synchronous behaviors). Many threads can use an at the same - /// time or a single thread can make many asynchronous requests. To get the most performance out of the connection + /// time or a single thread can make many asynchronous requests. To get the best performance out of the connection /// it is generally recommended to use it in this way. /// public interface IEventStoreConnection : IDisposable { /// - /// Gets the name of this connection. A connection name can be used for disambiguation + /// Gets the name of this connection. A connection name is useful for disambiguation /// in log files. /// string ConnectionName { get; } /// - /// Connects the asynchronously to a destination + /// Connects the asynchronously to a destination. /// - /// A that can be waited upon. + /// A to wait upon. Task ConnectAsync(); /// - /// Closes this + /// Closes this . /// void Close(); /// - /// Deletes a stream from the Event Store asynchronously + /// Deletes a stream from Event Store asynchronously. /// /// The name of the stream to delete. /// The expected version that the streams should have when being deleted. @@ -49,111 +49,110 @@ public interface IEventStoreConnection : IDisposable Task DeleteStreamAsync(string stream, long expectedVersion, UserCredentials userCredentials = null); /// - /// Deletes a stream from the Event Store asynchronously + /// Deletes a stream from Event Store asynchronously. /// /// The name of the stream to delete. /// The expected version that the streams should have when being deleted. - /// Indicator for tombstoning vs soft-deleting the stream. Tombstoned streams can never be recreated. Soft-deleted streams - /// can be written to again, but the EventNumber sequence will not start from 0. + /// Indicator for tombstoning vs soft-deleting the stream. Tombstoned streams can never be recreated. Soft-deleted streams can be written to again, but the EventNumber sequence will not start from 0. /// The optional user credentials to perform operation with. /// A containing the results of the delete stream operation. Task DeleteStreamAsync(string stream, long expectedVersion, bool hardDelete, UserCredentials userCredentials = null); /// - /// Appends Events asynchronously to a stream. + /// Appends events asynchronously to a stream. /// /// /// When appending events to a stream the choice can - /// make a very large difference in the observed behavior. For example, if no stream exists + /// make a large difference in the observed behavior. For example, if no stream exists /// and ExpectedVersion.Any is used, a new stream will be implicitly created when appending. - /// + /// TODO: Link above and below? /// There are also differences in idempotency between different types of calls. - /// If you specify an ExpectedVersion aside from ExpectedVersion.Any the Event Store - /// will give you an idempotency guarantee. If using ExpectedVersion.Any the Event Store - /// will do its best to provide idempotency but does not guarantee idempotency + /// If you specify an ExpectedVersion aside from ExpectedVersion.Any, Event Store + /// will give you an idempotency guarantee. If using ExpectedVersion.Any, Event Store + /// will do its best to provide idempotency but does not guarantee idempotency. /// - /// The name of the stream to append events to - /// The of the stream to append to - /// The events to append to the stream + /// The name of the stream to append events to. + /// The of the stream to append to. + /// The events to append to the stream. /// A containing the results of the write operation. Task AppendToStreamAsync(string stream, long expectedVersion, params EventData[] events); /// - /// Appends Events asynchronously to a stream. + /// Appends events asynchronously to a stream. /// /// /// When appending events to a stream the choice can - /// make a very large difference in the observed behavior. For example, if no stream exists + /// make a large difference in the observed behavior. For example, if no stream exists /// and ExpectedVersion.Any is used, a new stream will be implicitly created when appending. - /// + /// /// There are also differences in idempotency between different types of calls. - /// If you specify an ExpectedVersion aside from ExpectedVersion.Any the Event Store - /// will give you an idempotency guarantee. If using ExpectedVersion.Any the Event Store - /// will do its best to provide idempotency but does not guarantee idempotency + /// If you specify an ExpectedVersion aside from ExpectedVersion.Any, Event Store + /// will give you an idempotency guarantee. If using ExpectedVersion.Any, Event Store + /// will do its best to provide idempotency but does not guarantee idempotency. /// - /// The name of the stream to append events to - /// The of the stream to append to + /// The name of the stream to append events to. + /// The of the stream to append to. /// The optional user credentials to perform operation with. - /// The events to append to the stream + /// The events to append to the stream. /// A containing the results of the write operation. Task AppendToStreamAsync(string stream, long expectedVersion, UserCredentials userCredentials, params EventData[] events); /// - /// Appends Events asynchronously to a stream. + /// Appends events asynchronously to a stream. /// /// /// When appending events to a stream the choice can - /// make a very large difference in the observed behavior. For example, if no stream exists + /// make a large difference in the observed behavior. For example, if no stream exists /// and ExpectedVersion.Any is used, a new stream will be implicitly created when appending. /// /// There are also differences in idempotency between different types of calls. - /// If you specify an ExpectedVersion aside from ExpectedVersion.Any the Event Store - /// will give you an idempotency guarantee. If using ExpectedVersion.Any the Event Store - /// will do its best to provide idempotency but does not guarantee idempotency + /// If you specify an ExpectedVersion aside from ExpectedVersion.Any, Event Store + /// will give you an idempotency guarantee. If using ExpectedVersion.Any, Event Store + /// will do its best to provide idempotency but does not guarantee idempotency. /// - /// The name of the stream to append events to - /// The of the stream to append to - /// The events to append to the stream + /// The name of the stream to append events to. + /// The of the stream to append to. + /// The events to append to the stream. /// The optional user credentials to perform operation with. ///A containing the results of the write operation. Task AppendToStreamAsync(string stream, long expectedVersion, IEnumerable events, UserCredentials userCredentials = null); /// - /// Appends Events asynchronously to a stream if the stream version matches the . + /// Appends events asynchronously to a stream if the stream version matches the . /// /// /// When appending events to a stream the choice can - /// make a very large difference in the observed behavior. For example, if no stream exists + /// make a large difference in the observed behavior. For example, if no stream exists /// and ExpectedVersion.Any is used, a new stream will be implicitly created when appending. /// /// There are also differences in idempotency between different types of calls. - /// If you specify an ExpectedVersion aside from ExpectedVersion.Any the Event Store - /// will give you an idempotency guarantee. If using ExpectedVersion.Any the Event Store - /// will do its best to provide idempotency but does not guarantee idempotency + /// If you specify an ExpectedVersion aside from ExpectedVersion.Any, Event Store + /// will give you an idempotency guarantee. If using ExpectedVersion.Any, Event Store + /// will do its best to provide idempotency but does not guarantee idempotency. /// - /// The name of the stream to append events to - /// The of the stream to append to - /// The events to append to the stream + /// The name of the stream to append events to. + /// The of the stream to append to. + /// The events to append to the stream. /// The optional user credentials to perform operation with. /// A describing if the operation succeeded and, if not, the reason for failure (which can be either stream version mismatch or trying to write to a deleted stream). Task ConditionalAppendToStreamAsync(string stream, long expectedVersion, IEnumerable events, UserCredentials userCredentials = null); /// - /// Starts a transaction in the event store on a given stream asynchronously + /// Starts an asynchronous transaction in Event Store on a given stream. /// /// /// A allows the calling of multiple writes with multiple - /// round trips over long periods of time between the caller and the event store. This method + /// round trips over long periods of time between the caller and Event Store. This method /// is only available through the TCP interface and no equivalent exists for the RESTful interface. /// - /// The stream to start a transaction on - /// The expected version of the stream at the time of starting the transaction + /// The stream to start a transaction on. + /// The expected version of the stream at the time of starting the transaction. /// The optional user credentials to perform operation with. /// A representing a multi-request transaction. Task StartTransactionAsync(string stream, long expectedVersion, UserCredentials userCredentials = null); /// - /// Continues transaction by provided transaction ID. + /// Continues specified transaction. /// /// /// A allows the calling of multiple writes with multiple @@ -176,43 +175,43 @@ public interface IEventStoreConnection : IDisposable Task ReadEventAsync(string stream, long eventNumber, bool resolveLinkTos, UserCredentials userCredentials = null); /// - /// Reads count Events from an Event Stream forwards (e.g. oldest to newest) starting from position start + /// Asynchronously reads count events from an event stream forwards (e.g. oldest to newest) starting from position start. /// - /// The stream to read from - /// The starting point to read from - /// The count of items to read - /// Whether to resolve LinkTo events automatically + /// The stream to read from. + /// The starting point to read from. + /// The count of items to read. + /// Whether to resolve LinkTo events automatically. /// The optional user credentials to perform operation with. /// A containing the results of the read operation. Task ReadStreamEventsForwardAsync(string stream, long start, int count, bool resolveLinkTos, UserCredentials userCredentials = null); /// - /// Reads count events from an Event Stream backwards (e.g. newest to oldest) from position asynchronously + /// Asynchronously reads count events from an event stream backwards (e.g. newest to oldest) from position. /// - /// The Event Stream to read from - /// The position to start reading from - /// The count to read from the position - /// Whether to resolve LinkTo events automatically + /// The stream to read from. + /// The position to start reading from. + /// The count to read from the position. + /// Whether to resolve LinkTo events automatically. /// The optional user credentials to perform operation with. /// A containing the results of the read operation. Task ReadStreamEventsBackwardAsync(string stream, long start, int count, bool resolveLinkTos, UserCredentials userCredentials = null); /// - /// Reads All Events in the node forward asynchronously (e.g. beginning to end) + /// Asynchronously reads all events in the node forward (e.g. beginning to end). /// - /// The position to start reading from - /// The maximum count to read - /// Whether to resolve LinkTo events automatically + /// The position to start reading from. + /// The maximum count to read. + /// Whether to resolve LinkTo events automatically. /// The optional user credentials to perform operation with. /// A containing the records read. Task ReadAllEventsForwardAsync(Position position, int maxCount, bool resolveLinkTos, UserCredentials userCredentials = null); /// - /// Reads All Events in the node backwards (e.g. end to beginning) + /// Asynchronously reads all events in the node backwards (e.g. end to beginning). /// - /// The position to start reading from - /// The maximum count to read - /// Whether to resolve Link events automatically + /// The position to start reading from. + /// The maximum count to read. + /// Whether to resolve Link events automatically. /// The optional user credentials to perform operation with. /// A containing the records read. Task ReadAllEventsBackwardAsync(Position position, int maxCount, bool resolveLinkTos, UserCredentials userCredentials = null); @@ -222,11 +221,11 @@ public interface IEventStoreConnection : IDisposable /// written to the stream while the subscription is active will be /// pushed to the client. /// - /// The stream to subscribe to - /// Whether to resolve Link events automatically - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation + /// The stream to subscribe to. + /// Whether to resolve Link events automatically. + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. /// A representing the subscription. Task SubscribeToStreamAsync( string stream, @@ -257,15 +256,15 @@ Task SubscribeToStreamAsync( /// is desired, use the event number of the last event processed which /// appeared on the subscription. /// - /// NOTE: Using here will result in missing + /// Using here will result in missing /// the first event in the stream. - /// Whether to resolve Link events automatically - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked when the subscription switches to newly-pushed events - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation - /// The batch size to use during the read phase - /// The name of subscription + /// Whether to resolve Link events automatically. + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked when the subscription switches to newly-pushed events. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. + /// The batch size to use during the read phase. + /// The name of subscription. /// An representing the subscription. [Obsolete("This method will be obsoleted in the next major version please switch to the overload with a settings object")] EventStoreStreamCatchUpSubscription SubscribeToStreamFrom( @@ -293,7 +292,7 @@ EventStoreStreamCatchUpSubscription SubscribeToStreamFrom( /// switches from the reading /// phase to the live subscription phase. /// - /// The stream to subscribe to + /// The stream to subscribe to. /// The event number from which to start. /// /// To receive all events in the stream, use . @@ -301,13 +300,13 @@ EventStoreStreamCatchUpSubscription SubscribeToStreamFrom( /// is desired, use the event number of the last event processed which /// appeared on the subscription. /// - /// NOTE: Using here will result in missing + /// Using here will result in missing /// the first event in the stream. - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked when the subscription switches to newly-pushed events - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation - /// The for the subscription + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked when the subscription switches to newly-pushed events. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. + /// The for the subscription. /// An representing the subscription. EventStoreStreamCatchUpSubscription SubscribeToStreamFrom( string stream, @@ -321,14 +320,14 @@ EventStoreStreamCatchUpSubscription SubscribeToStreamFrom( /// - /// Asynchronously subscribes to all events in the Event Store. New + /// Asynchronously subscribes to all events in Event Store. New /// events written to the stream while the subscription is active /// will be pushed to the client. /// - /// Whether to resolve Link events automatically - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation + /// Whether to resolve Link events automatically. + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. /// A representing the subscription. Task SubscribeToAllAsync( bool resolveLinkTos, @@ -337,18 +336,18 @@ Task SubscribeToAllAsync( UserCredentials userCredentials = null); /// - /// Subscribes to a persistent subscription(competing consumer) on event store + /// Subscribes to a persistent subscription (competing consumer) on an event store. /// - /// The subscription group to connect to - /// The stream to subscribe to - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation - /// The buffer size to use for the persistent subscription + /// The subscription group to connect to. + /// The stream to subscribe to. + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. + /// The buffer size to use for the persistent subscription. /// Whether the subscription should automatically acknowledge messages processed. /// If not set the receiver is required to explicitly acknowledge messages through the subscription. /// This will connect you to a persistent subscription group for a stream. The subscription group - /// must first be created with CreatePersistentSubscriptionAsync many connections + /// must first be created with CreatePersistentSubscriptionAsync. Many connections /// can connect to the same group and they will be treated as competing consumers within the group. /// If one connection dies work will be balanced across the rest of the consumers in the group. If /// you attempt to connect to a group that does not exist you will be given an exception. @@ -357,25 +356,25 @@ Task SubscribeToAllAsync( EventStorePersistentSubscriptionBase ConnectToPersistentSubscription( string stream, string groupName, - Func eventAppeared, + Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, bool autoAck = true); /// - /// Asynchronously subscribes to a persistent subscription(competing consumer) on event store + /// Asynchronously subscribes to a persistent subscription (competing consumer) on an event store. /// - /// The subscription group to connect to - /// The stream to subscribe to - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation - /// The buffer size to use for the persistent subscription + /// The subscription group to connect to. + /// The stream to subscribe to. + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. + /// The buffer size to use for the persistent subscription. /// Whether the subscription should automatically acknowledge messages processed. /// If not set the receiver is required to explicitly acknowledge messages through the subscription. /// This will connect you to a persistent subscription group for a stream. The subscription group - /// must first be created with CreatePersistentSubscriptionAsync many connections + /// must first be created with CreatePersistentSubscriptionAsync. Many connections /// can connect to the same group and they will be treated as competing consumers within the group. /// If one connection dies work will be balanced across the rest of the consumers in the group. If /// you attempt to connect to a group that does not exist you will be given an exception. @@ -384,7 +383,7 @@ EventStorePersistentSubscriptionBase ConnectToPersistentSubscription( Task ConnectToPersistentSubscriptionAsync( string stream, string groupName, - Func eventAppeared, + Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, @@ -419,7 +418,7 @@ EventStorePersistentSubscription ConnectToPersistentSubscriptionForAll( /// /// Subscribes to all events. Existing events from lastCheckpoint - /// onwards are read from the Event Store and presented to the user of + /// onwards are read from Event Store and presented to the user of /// as if they had been pushed. /// /// Once the end of the stream is read the subscription is @@ -437,15 +436,15 @@ EventStorePersistentSubscription ConnectToPersistentSubscriptionForAll( /// is desired, use the position representing the last event processed which /// appeared on the subscription. /// - /// NOTE: Using here will result in missing + /// Using here will result in missing /// the first event in the stream. - /// Whether to resolve Link events automatically - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked when the subscription switches to newly-pushed events - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation - /// The batch size to use during the read phase - /// The name of subscription + /// Whether to resolve Link events automatically. + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked when the subscription switches to newly-pushed events. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. + /// The batch size to use during the read phase. + /// The name of subscription. /// An representing the subscription. [Obsolete("This overload will be removed in the next major release please use the overload with a settings object")] EventStoreAllCatchUpSubscription SubscribeToAllFrom( @@ -460,7 +459,7 @@ EventStoreAllCatchUpSubscription SubscribeToAllFrom( /// /// Subscribes to a all events. Existing events from lastCheckpoint - /// onwards are read from the Event Store and presented to the user of + /// onwards are read from Event Store and presented to the user of /// as if they had been pushed. /// /// Once the end of the stream is read the subscription is @@ -478,13 +477,13 @@ EventStoreAllCatchUpSubscription SubscribeToAllFrom( /// is desired, use the position representing the last event processed which /// appeared on the subscription. /// - /// NOTE: Using here will result in missing + /// Using here will result in missing /// the first event in the stream. - /// A Task invoked and awaited when a new event is received over the subscription - /// An action invoked when the subscription switches to newly-pushed events - /// An action invoked if the subscription is dropped - /// User credentials to use for the operation - /// The for the subscription + /// A Task invoked and awaited when a new event is received over the subscription. + /// An action invoked when the subscription switches to newly-pushed events. + /// An action invoked if the subscription is dropped. + /// User credentials to use for the operation. + /// The for the subscription. /// An representing the subscription. EventStoreAllCatchUpSubscription SubscribeToAllFrom( Position? lastCheckpoint, @@ -508,10 +507,10 @@ EventStoreAllCatchUpSubscription SubscribeToAllFrom( */ /// - /// Asynchronously update a persistent subscription group on a stream + /// Asynchronously update a persistent subscription group on a stream. /// - /// The name of the stream to create the persistent subscription on - /// The name of the group to create + /// The name of the stream to create the persistent subscription on. + /// The name of the group to create. /// The for the subscription /// The credentials to be used for this operation. /// A that can be waited upon. @@ -519,22 +518,22 @@ EventStoreAllCatchUpSubscription SubscribeToAllFrom( /// - /// Asynchronously create a persistent subscription group on a stream + /// Asynchronously create a persistent subscription group on a stream. /// - /// The name of the stream to create the persistent subscription on - /// The name of the group to create - /// The for the subscription + /// The name of the stream to create the persistent subscription on. + /// The name of the group to create. + /// The for the subscription. /// The credentials to be used for this operation. /// A that can be waited upon. Task CreatePersistentSubscriptionAsync(string stream, string groupName, PersistentSubscriptionSettings settings, UserCredentials credentials); /// - /// Asynchronously delete a persistent subscription group on a stream + /// Asynchronously delete a persistent subscription group on a stream. /// - /// The name of the stream to delete the persistent subscription on - /// The name of the group to delete - /// User credentials to use for the operation + /// The name of the stream to delete the persistent subscription on. + /// The name of the group to delete. + /// User credentials to use for the operation. /// A that can be waited upon. Task DeletePersistentSubscriptionAsync(string stream, string groupName, UserCredentials userCredentials = null); @@ -554,7 +553,7 @@ EventStoreAllCatchUpSubscription SubscribeToAllFrom( /// The name of the stream for which to set metadata. /// The expected version for the write to the metadata stream. /// A representing the new metadata. - /// User credentials to use for the operation + /// User credentials to use for the operation. /// A containing the results of the write operation. Task SetStreamMetadataAsync(string stream, long expectedMetastreamVersion, StreamMetadata metadata, UserCredentials userCredentials = null); diff --git a/src/EventStore.ClientAPI.NetCore/IEventStoreConnectionExtensions.cs b/src/EventStore.ClientAPI.NetCore/IEventStoreConnectionExtensions.cs index f9dce42..af5f306 100644 --- a/src/EventStore.ClientAPI.NetCore/IEventStoreConnectionExtensions.cs +++ b/src/EventStore.ClientAPI.NetCore/IEventStoreConnectionExtensions.cs @@ -16,7 +16,7 @@ private static Func ToTask(Action eventAppeared(subscription, e); return Task.CompletedTask; }; - + /// /// Asynchronously subscribes to a single event stream. New events /// written to the stream while the subscription is active will be @@ -214,7 +214,44 @@ public static EventStorePersistentSubscriptionBase ConnectToPersistentSubscripti ); /// - /// Asynchronously subscribes to a persistent subscription(competing consumer) on event store + /// Subscribes to a persistent subscription(competing consumer) on event store + /// + /// The connection to subscribe to + /// The subscription group to connect to + /// The stream to subscribe to + /// An action invoked when a new event is received over the subscription + /// An action invoked if the subscription is dropped + /// User credentials to use for the operation + /// The buffer size to use for the persistent subscription + /// Whether the subscription should automatically acknowledge messages processed. + /// If not set the receiver is required to explicitly acknowledge messages through the subscription. + /// This will connect you to a persistent subscription group for a stream. The subscription group + /// must first be created with CreatePersistentSubscriptionAsync many connections + /// can connect to the same group and they will be treated as competing consumers within the group. + /// If one connection dies work will be balanced across the rest of the consumers in the group. If + /// you attempt to connect to a group that does not exist you will be given an exception. + /// + /// An representing the subscription + public static EventStorePersistentSubscriptionBase ConnectToPersistentSubscription( + this IEventStoreConnection target, + string stream, + string groupName, + Func eventAppeared, + Action subscriptionDropped = null, + UserCredentials userCredentials = null, + int bufferSize = 10, + bool autoAck = true) => + target.ConnectToPersistentSubscription( + stream, + groupName, + (s,e, i) => eventAppeared(s,e), + subscriptionDropped, + userCredentials, + bufferSize, + autoAck + ); + /// + /// Subscribes to a persistent subscription(competing consumer) on event store /// /// The connection to subscribe to /// The subscription group to connect to @@ -251,6 +288,44 @@ public static Task ConnectToPersistentSubs autoAck ); + /// + /// Subscribes to a persistent subscription(competing consumer) on event store + /// + /// The connection to subscribe to + /// The subscription group to connect to + /// The stream to subscribe to + /// An action invoked when a new event is received over the subscription + /// An action invoked if the subscription is dropped + /// User credentials to use for the operation + /// The buffer size to use for the persistent subscription + /// Whether the subscription should automatically acknowledge messages processed. + /// If not set the receiver is required to explicitly acknowledge messages through the subscription. + /// This will connect you to a persistent subscription group for a stream. The subscription group + /// must first be created with CreatePersistentSubscriptionAsync many connections + /// can connect to the same group and they will be treated as competing consumers within the group. + /// If one connection dies work will be balanced across the rest of the consumers in the group. If + /// you attempt to connect to a group that does not exist you will be given an exception. + /// + /// An representing the subscription + public static Task ConnectToPersistentSubscriptionAsync( + this IEventStoreConnection target, + string stream, + string groupName, + Func eventAppeared, + Action subscriptionDropped = null, + UserCredentials userCredentials = null, + int bufferSize = 10, + bool autoAck = true) => + target.ConnectToPersistentSubscriptionAsync( + stream, + groupName, + (s,e, i) => eventAppeared(s,e), + subscriptionDropped, + userCredentials, + bufferSize, + autoAck + ); + /// /// Subscribes to all events. Existing events from lastCheckpoint /// onwards are read from the Event Store and presented to the user of 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 46870bf..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,93 +152,95 @@ 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"); Ensure.Positive(count, "count"); - if(count > Consts.MaxReadSize) throw new ArgumentException(string.Format("Count should be less than {0}. For larger reads you should page.", Consts.MaxReadSize)); + if(count > 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 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"); - if (count > Consts.MaxReadSize) throw new ArgumentException(string.Format("Count should be less than {0}. For larger reads you should page.", Consts.MaxReadSize)); + if (count > 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 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 > Consts.MaxReadSize) throw new ArgumentException(string.Format("Count should be less than {0}. For larger reads you should page.", Consts.MaxReadSize)); + 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 > Consts.MaxReadSize) throw new ArgumentException(string.Format("Count should be less than {0}. For larger reads you should page.", Consts.MaxReadSize)); + 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)); } @@ -345,7 +350,7 @@ public EventStoreAllCatchUpSubscription SubscribeToAllFrom( public EventStorePersistentSubscriptionBase ConnectToPersistentSubscription( string stream, string groupName, - Func eventAppeared, + Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, @@ -366,7 +371,7 @@ public EventStorePersistentSubscriptionBase ConnectToPersistentSubscription( public Task ConnectToPersistentSubscriptionAsync( string stream, string groupName, - Func eventAppeared, + Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, @@ -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/Messages.cs b/src/EventStore.ClientAPI.NetCore/Internal/Messages.cs index 51f52e4..f42151b 100644 --- a/src/EventStore.ClientAPI.NetCore/Internal/Messages.cs +++ b/src/EventStore.ClientAPI.NetCore/Internal/Messages.cs @@ -136,13 +136,13 @@ internal class StartPersistentSubscriptionMessage : Message public readonly string StreamId; public readonly int BufferSize; public readonly UserCredentials UserCredentials; - public readonly Func EventAppeared; + public readonly Func EventAppeared; public readonly Action SubscriptionDropped; public readonly int MaxRetries; public readonly TimeSpan Timeout; - public StartPersistentSubscriptionMessage(TaskCompletionSource source, string subscriptionId, string streamId, int bufferSize, UserCredentials userCredentials, Func eventAppeared, Action subscriptionDropped, int maxRetries, TimeSpan timeout) + public StartPersistentSubscriptionMessage(TaskCompletionSource source, string subscriptionId, string streamId, int bufferSize, UserCredentials userCredentials, Func eventAppeared, Action subscriptionDropped, int maxRetries, TimeSpan timeout) { Ensure.NotNull(source, "source"); Ensure.NotNull(eventAppeared, "eventAppeared"); 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/Messages/ClientMessage.cs b/src/EventStore.ClientAPI.NetCore/Messages/ClientMessage.cs index fdf253f..8740549 100644 --- a/src/EventStore.ClientAPI.NetCore/Messages/ClientMessage.cs +++ b/src/EventStore.ClientAPI.NetCore/Messages/ClientMessage.cs @@ -1037,12 +1037,16 @@ public partial class PersistentSubscriptionStreamEventAppeared [ProtoMember(1, IsRequired = true, Name = @"event", DataFormat = DataFormat.Default)] public readonly ResolvedIndexedEvent Event; - private PersistentSubscriptionStreamEventAppeared() { } - - public PersistentSubscriptionStreamEventAppeared(ResolvedIndexedEvent @event) - { - Event = @event; - } + [ProtoMember(2, IsRequired = false, Name=@"retryCount", DataFormat = DataFormat.TwosComplement)] + public readonly int? RetryCount; + + private PersistentSubscriptionStreamEventAppeared() {} + + public PersistentSubscriptionStreamEventAppeared(ResolvedIndexedEvent @event, int? retryCount) + { + Event = @event; + RetryCount = retryCount; + } } [ProtoContract(Name = @"SubscribeToStream")] 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/PersistentSubscriptionSettings.cs b/src/EventStore.ClientAPI.NetCore/PersistentSubscriptionSettings.cs index 6275166..43ecb30 100644 --- a/src/EventStore.ClientAPI.NetCore/PersistentSubscriptionSettings.cs +++ b/src/EventStore.ClientAPI.NetCore/PersistentSubscriptionSettings.cs @@ -4,8 +4,8 @@ namespace EventStore.ClientAPI { /// - /// Represents the settings for a . This should not be used - /// directly, but instead created via a + /// Represents the settings for a . You should not use + /// this directly, but instead created via a . /// public class PersistentSubscriptionSettings { @@ -32,57 +32,57 @@ public static PersistentSubscriptionSettingsBuilder Create() /// - /// Whether or not the should resolve linkTo events to their linked events. + /// Whether the should resolve linkTo events to their linked events. /// public readonly bool ResolveLinkTos; /// - /// Where the subscription should start from (position) + /// Which event position in the stream the subscription should start from. /// public readonly long StartFrom; /// - /// Whether or not in depth latency statistics should be tracked on this subscription. + /// Whether to track latency statistics on this subscription. /// public readonly bool ExtraStatistics; /// - /// The amount of time after which a message should be considered to be timedout and retried. + /// The amount of time after which to consider a message as timedout and retried. /// public readonly TimeSpan MessageTimeout; /// - /// The maximum number of retries (due to timeout) before a message get considered to be parked + /// The maximum number of retries (due to timeout) before a message is considered to be parked. /// public int MaxRetryCount; /// - /// The size of the buffer listening to live messages as they happen + /// The size of the buffer (in-memory) listening to live messages as they happen before paging occurs. /// public int LiveBufferSize; /// - /// The number of events read at a time when paging in history + /// The number of events read at a time when paging through history. /// public int ReadBatchSize; /// - /// The number of events to cache when paging through history + /// The number of events to cache when paging through history. /// public int HistoryBufferSize; /// - /// The amount of time to try to checkpoint after + /// The amount of time to try to checkpoint after. /// public readonly TimeSpan CheckPointAfter; /// - /// The minimum number of messages to checkpoint + /// The minimum number of messages to write to a checkpoint. /// public readonly int MinCheckPointCount; /// - /// The maximum number of messages to checkpoint if this number is a reached a checkpoint will be forced. + /// The maximum number of messages not checkpointed before forcing a checkpoint. /// public readonly int MaxCheckPointCount; @@ -97,15 +97,15 @@ public static PersistentSubscriptionSettingsBuilder Create() public string NamedConsumerStrategy; /// - /// Constructs a new + /// Constructs a new . /// internal PersistentSubscriptionSettings(bool resolveLinkTos, long startFrom, bool extraStatistics, TimeSpan messageTimeout, int maxRetryCount, int liveBufferSize, int readBatchSize, int historyBufferSize, TimeSpan checkPointAfter, int minCheckPointCount, int maxCheckPointCount, int maxSubscriberCount, string namedConsumerStrategy) { - if (messageTimeout.TotalMilliseconds > Int32.MaxValue) throw new ArgumentException("milliseconds must be less or equal to than int32.MaxValue", nameof(messageTimeout)); - if (checkPointAfter.TotalMilliseconds > Int32.MaxValue) throw new ArgumentException("milliseconds must be less or equal to than int32.MaxValue", nameof(checkPointAfter)); + if(messageTimeout.TotalMilliseconds > Int32.MaxValue) throw new ArgumentException("milliseconds must be less or equal to than int32.MaxValue", nameof(messageTimeout)); + if(checkPointAfter.TotalMilliseconds > Int32.MaxValue) throw new ArgumentException("milliseconds must be less or equal to than int32.MaxValue", nameof(checkPointAfter)); MessageTimeout = messageTimeout; ResolveLinkTos = resolveLinkTos; StartFrom = startFrom; diff --git a/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsClient.cs b/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsClient.cs index 80e1887..e87ac51 100644 --- a/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsClient.cs +++ b/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsClient.cs @@ -147,6 +147,13 @@ public Task UpdateQuery(EndPoint endPoint, string name, string query, UserCreden return SendPut(endPoint.ToHttpUrl(httpSchema, "/projection/{0}/query?type=JS", name), query, userCredentials, HttpStatusCode.OK); } + public Task UpdateQuery(EndPoint endPoint, string name, string query, UserCredentials userCredentials = null, string httpSchema = EndpointExtensions.HTTP_SCHEMA, bool? emitEnabled = null) + { + var emit = emitEnabled.HasValue ? (emitEnabled.Value ? "1" : "0") : string.Empty; + var url = emitEnabled.HasValue ? "/projection/{0}/query?emit={1}" : "/projection/{0}/query"; + return SendPut(endPoint.ToHttpUrl(httpSchema, url, name, emit), query, userCredentials, HttpStatusCode.OK); + } + public Task Delete(EndPoint endPoint, string name, bool deleteEmittedStreams, UserCredentials userCredentials = null, string httpSchema = EndpointExtensions.HTTP_SCHEMA) { return SendDelete(endPoint.ToHttpUrl(httpSchema, "/projection/{0}?deleteEmittedStreams={1}", name, deleteEmittedStreams), userCredentials, HttpStatusCode.OK); diff --git a/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsManager.cs b/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsManager.cs index 582a79e..cb78f84 100644 --- a/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsManager.cs +++ b/src/EventStore.ClientAPI.NetCore/Projections/ProjectionsManager.cs @@ -286,6 +286,21 @@ public Task UpdateQueryAsync(string name, string query, UserCredentials userCred return _client.UpdateQuery(_httpEndPoint, name, query, userCredentials, _httpSchema); } + /// + /// Asynchronously updates the definition of a query. + /// + /// The name of the query. + /// The JavaScript source code of the query. + /// Credentials for the operation. + /// Allow projection to send events to other streams + public Task UpdateQueryAsync(string name, string query, bool? emitEnabled = null, UserCredentials userCredentials = null) + { + Ensure.NotNullOrEmpty(name, "name"); + Ensure.NotNullOrEmpty(query, "query"); + + return _client.UpdateQuery(_httpEndPoint, name, query, userCredentials, _httpSchema, emitEnabled); + } + /// /// Asynchronously deletes a projection /// 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/ResolvedEvent.cs b/src/EventStore.ClientAPI.NetCore/ResolvedEvent.cs index 2515c11..403b697 100644 --- a/src/EventStore.ClientAPI.NetCore/ResolvedEvent.cs +++ b/src/EventStore.ClientAPI.NetCore/ResolvedEvent.cs @@ -1,11 +1,12 @@ -using EventStore.ClientAPI.Messages; +using EventStore.ClientAPI.ClientOperations; +using EventStore.ClientAPI.Messages; namespace EventStore.ClientAPI { /// /// A structure representing a single event or an resolved link event. /// - public struct ResolvedEvent + public struct ResolvedEvent : IResolvedEvent { /// /// The event, or the resolved link event if this is @@ -23,7 +24,7 @@ public struct ResolvedEvent /// /// If this represents a link event, the Link /// will be the , otherwise it will be the - /// Event. + /// event. /// public RecordedEvent OriginalEvent { get { return Link ?? Event; } } @@ -61,5 +62,10 @@ internal ResolvedEvent(ClientMessage.ResolvedIndexedEvent evnt) Link = evnt.Link == null ? null : new RecordedEvent(evnt.Link); OriginalPosition = null; } + + Position? IResolvedEvent.OriginalPosition => OriginalPosition; + RecordedEvent IResolvedEvent.OriginalEvent => OriginalEvent; + long IResolvedEvent.OriginalEventNumber=> OriginalEventNumber; + string IResolvedEvent.OriginalStreamId => OriginalStreamId; } } \ No newline at end of file diff --git a/src/EventStore.ClientAPI.NetCore/StreamEventsSlice.cs b/src/EventStore.ClientAPI.NetCore/StreamEventsSlice.cs index 4018a98..99d7ec4 100644 --- a/src/EventStore.ClientAPI.NetCore/StreamEventsSlice.cs +++ b/src/EventStore.ClientAPI.NetCore/StreamEventsSlice.cs @@ -5,17 +5,17 @@ namespace EventStore.ClientAPI { /// - /// An Stream Events Slice represents the result of a single read operation to the event store. + /// A stream events slice represents the result of a single read operation to Event Store. /// public class StreamEventsSlice { /// - /// The representing the status of this read attempt + /// The representing the status of this read attempt. /// public readonly SliceReadStatus Status; /// - /// The name of the stream read + /// The name of the stream to read. /// public readonly string Stream; @@ -30,7 +30,7 @@ public class StreamEventsSlice public readonly ReadDirection ReadDirection; /// - /// The events read represented as + /// The events read represented as . /// public readonly ResolvedEvent[] Events; diff --git a/src/EventStore.ClientAPI.NetCore/StreamMetadata.cs b/src/EventStore.ClientAPI.NetCore/StreamMetadata.cs index b32ca3c..61ddf8d 100644 --- a/src/EventStore.ClientAPI.NetCore/StreamMetadata.cs +++ b/src/EventStore.ClientAPI.NetCore/StreamMetadata.cs @@ -48,7 +48,7 @@ public class StreamMetadata public IEnumerable> CustomMetadataAsRawJsons { get { return _customMetadata.Select(x => new KeyValuePair(x.Key, x.Value.ToString())); } - } + } private readonly IDictionary _customMetadata; @@ -70,7 +70,7 @@ internal StreamMetadata(long? maxCount, TimeSpan? maxAge, long? truncateBefore, TruncateBefore = truncateBefore; CacheControl = cacheControl; Acl = acl; - _customMetadata = customMetadata ?? Empty.CustomStreamMetadata; + _customMetadata = customMetadata ?? Empty.CustomStreamMetadata; } /// @@ -104,10 +104,10 @@ public StreamMetadataBuilder Copy() { if(Acl == null) return new StreamMetadataBuilder( - MaxCount, - MaxAge, - TruncateBefore, - CacheControl, + MaxCount, + MaxAge, + TruncateBefore, + CacheControl, customMetadata: _customMetadata); return new StreamMetadataBuilder( MaxCount, @@ -429,7 +429,7 @@ private static void Check(JsonToken type, JsonTextReader reader) { if (reader.TokenType != type) throw new Exception("Invalid JSON"); - } + } private static void Check(bool read, JsonTextReader reader) { diff --git a/src/EventStore.ClientAPI.NetCore/StreamMetadataBuilder.cs b/src/EventStore.ClientAPI.NetCore/StreamMetadataBuilder.cs index d224e38..7b38207 100644 --- a/src/EventStore.ClientAPI.NetCore/StreamMetadataBuilder.cs +++ b/src/EventStore.ClientAPI.NetCore/StreamMetadataBuilder.cs @@ -3,7 +3,6 @@ using System.Linq; using EventStore.ClientAPI.Common.Utils; using Newtonsoft.Json.Linq; - namespace EventStore.ClientAPI { /// 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/FakeEventStoreConnection.cs b/test/EventStore.ClientAPI.NetCore.Tests/FakeEventStoreConnection.cs index d3cc962..7141c9c 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/FakeEventStoreConnection.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/FakeEventStoreConnection.cs @@ -11,6 +11,7 @@ internal class FakeEventStoreConnection : IEventStoreConnection private Func> _readAllEventsForwardAsync; private Func> _readStreamEventsForwardAsync; private Func, Action, Task> _subscribeToStreamAsync; + private Func, Action, Task> _subscribeToAllAsync; public void Dispose() { @@ -140,14 +141,26 @@ public EventStoreStreamCatchUpSubscription SubscribeToStreamFrom( throw new NotImplementedException(); } + public void HandleSubscribeToAllAsync(Func, Action, Task> callback) + { + _subscribeToAllAsync = callback; + } + public Task SubscribeToAllAsync(bool resolveLinkTos, Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null) { - throw new NotImplementedException(); + return _subscribeToAllAsync(resolveLinkTos, eventAppeared, subscriptionDropped); } public EventStorePersistentSubscriptionBase ConnectToPersistentSubscription(string stream, string groupName, - Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, + Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, + bool autoAck = true) + { + throw new NotImplementedException(); + } + + public Task ConnectToPersistentSubscriptionAsync(string stream, string groupName, Func eventAppeared, + Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, bool autoAck = true) { throw new NotImplementedException(); @@ -259,10 +272,5 @@ public void OnConnected(ClientConnectionEventArgs e) var handler = Connected; if (handler != null) handler(this, e); } - - public Task ConnectToPersistentSubscriptionAsync(string stream, string groupName, Func eventAppeared, Action subscriptionDropped = null, UserCredentials userCredentials = null, int bufferSize = 10, bool autoAck = true) - { - throw new NotImplementedException(); - } } } \ No newline at end of file diff --git a/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions.cs b/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions.cs index 9ee7a1b..98e9e76 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions.cs @@ -16,7 +16,7 @@ protected override void When() _conn.ConnectToPersistentSubscription( "nonexisting2", "foo", - (sub, e) => + (sub, e, i) => { Console.Write("appeared"); return Task.CompletedTask; diff --git a/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions_async.cs b/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions_async.cs index c7d8c20..4fc69e9 100644 --- a/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions_async.cs +++ b/test/EventStore.ClientAPI.NetCore.Tests/connect_to_non_existing_persistent_subscription_with_permissions_async.cs @@ -16,7 +16,7 @@ protected override void When() _conn.ConnectToPersistentSubscriptionAsync( "nonexisting2", "foo", - (sub, e) => + (sub, e, i) => { Console.Write("appeared"); return Task.CompletedTask; diff --git a/test/EventStore.ClientAPI.NetCore.Tests/connect_to_persistent_subscription_with_retries.cs b/test/EventStore.ClientAPI.NetCore.Tests/connect_to_persistent_subscription_with_retries.cs new file mode 100644 index 0000000..b5ffeb7 --- /dev/null +++ b/test/EventStore.ClientAPI.NetCore.Tests/connect_to_persistent_subscription_with_retries.cs @@ -0,0 +1,65 @@ +using EventStore.ClientAPI; +using NUnit.Framework; +using System; +using System.Collections.Generic; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace EventStore.Core.Tests.ClientAPI +{ + [TestFixture, Category("LongRunning")] + public class connect_to_persistent_subscription_with_retries : SpecificationWithConnection + { + private readonly string _stream = Guid.NewGuid().ToString("N"); + private readonly PersistentSubscriptionSettings _settings = PersistentSubscriptionSettings.Create() + .DoNotResolveLinkTos() + .StartFromBeginning(); + + private readonly AutoResetEvent _resetEvent = new AutoResetEvent(false); + private readonly Guid _id = Guid.NewGuid(); + int? _retryCount; + private const string _group = "retries"; + + protected override void Given() + { + _conn.CreatePersistentSubscriptionAsync(_stream, _group, _settings, + DefaultData.AdminCredentials).Wait(); + _conn.ConnectToPersistentSubscription( + _stream, + _group, + HandleEvent, + (sub, reason, ex) => { }, + DefaultData.AdminCredentials,autoAck:false); + + } + + protected override void When() + { + _conn.AppendToStreamAsync(_stream, ExpectedVersion.Any, DefaultData.AdminCredentials, + new EventData(_id, "test", true, Encoding.UTF8.GetBytes("{'foo' : 'bar'}"), new byte[0])).Wait(); + } + + private Task HandleEvent(EventStorePersistentSubscriptionBase sub, ResolvedEvent resolvedEvent, int? retryCount) + { + if (retryCount > 4) + { + _retryCount = retryCount; + sub.Acknowledge(resolvedEvent); + _resetEvent.Set(); + } + else + { + sub.Fail(resolvedEvent, PersistentSubscriptionNakEventAction.Retry, "Not yet tried enough times"); + } + return Task.CompletedTask; + } + + [Test] + public void events_are_retried_until_success() + { + Assert.IsTrue(_resetEvent.WaitOne(TimeSpan.FromSeconds(10))); + Assert.AreEqual(5, _retryCount); + } + } +} 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; diff --git a/wercker.yml b/wercker.yml index 2709a44..a821e06 100644 --- a/wercker.yml +++ b/wercker.yml @@ -17,9 +17,9 @@ build: name: install and run eventstore code: | apt-get install wget tzdata curl iproute2 -y - wget https://eventstore.org/downloads/EventStore-OSS-Ubuntu-14.04-v4.0.3.tar.gz - tar xfv EventStore-OSS-Ubuntu-14.04-v4.0.3.tar.gz - pushd ./EventStore-OSS-Ubuntu-14.04-v4.0.3 + wget https://eventstore.org/downloads/EventStore-OSS-Ubuntu-14.04-v4.1.0.tar.gz + tar xfv EventStore-OSS-Ubuntu-14.04-v4.1.0.tar.gz + pushd ./EventStore-OSS-Ubuntu-14.04-v4.1.0 nohup ./eventstored --run-projections All --mem-db --log ~/logs & popd