diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs index 60a5b758de6..f14582dbff5 100644 --- a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs +++ b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs @@ -84,7 +84,9 @@ public ReplayFilterSettings(Config config) case "repair-by-discard-old": mode = ReplayFilterMode.RepairByDiscardOld; break; case "fail": mode = ReplayFilterMode.Fail; break; case "warn": mode = ReplayFilterMode.Warn; break; - default: throw new ConfigurationException($"Invalid replay-filter.mode [{replayModeString}], supported values [off, repair-by-discard-old, fail, warn]"); + default: + throw new ConfigurationException( + $"Invalid replay-filter.mode [{replayModeString}], supported values [off, repair-by-discard-old, fail, warn]"); } Mode = mode; @@ -236,14 +238,16 @@ public abstract class BatchingSqlJournalSetup /// /// The default serializer used when not type override matching is found /// - [Obsolete(message: "This property should never be used for writes, use the default `System.Object` serializer instead")] + [Obsolete( + message: + "This property should never be used for writes, use the default `System.Object` serializer instead")] public string DefaultSerializer { get; } /// /// The fully qualified name of the type that should be used as timestamp provider. /// public string TimestampProviderTypeName { get; } - + /// /// Isolation level of transactions used during read query execution. /// @@ -278,19 +282,21 @@ protected BatchingSqlJournalSetup(Config config, QueryConfiguration namingConven throw ConfigurationException.NullOrEmptyConfig(); var connectionString = config.GetString("connection-string", null); +#if NETSTANDARD if (string.IsNullOrWhiteSpace(connectionString)) { connectionString = System.Configuration.ConfigurationManager .ConnectionStrings[config.GetString("connection-string-name", "DefaultConnection")]? .ConnectionString; } +#endif if (string.IsNullOrWhiteSpace(connectionString)) throw new ConfigurationException("No connection string for Sql Event Journal was specified"); ReadIsolationLevel = namingConventions.ReadIsolationLevel; WriteIsolationLevel = namingConventions.WriteIsolationLevel; - + // backward compatibility var level = config.GetString("isolation-level"); if (level is { }) @@ -357,8 +363,9 @@ protected BatchingSqlJournalSetup( replayFilterSettings: replayFilterSettings, namingConventions: namingConventions, defaultSerializer: defaultSerializer) - { } - + { + } + /// /// Initializes a new instance of the class. /// @@ -630,8 +637,9 @@ public RequestChunk(int chunkId, IJournalRequest[] requests) protected BatchingSqlJournal(BatchingSqlJournalSetup setup) { Setup = setup; - CanPublish = Persistence.Instance.Apply(Context.System).Settings.Internal.PublishPluginCommands; - TimestampProvider = TimestampProviderProvider.GetTimestampProvider(setup.TimestampProviderTypeName, Context); + CanPublish = Persistence.Instance.Apply(Context.System).Settings.Internal.PublishPluginCommands; + TimestampProvider = + TimestampProviderProvider.GetTimestampProvider(setup.TimestampProviderTypeName, Context); _remainingOperations = Setup.MaxConcurrentOperations; _buffers = new[] @@ -648,7 +656,7 @@ protected BatchingSqlJournal(BatchingSqlJournalSetup setup) maxFailures: Setup.CircuitBreakerSettings.MaxFailures, callTimeout: Setup.CircuitBreakerSettings.CallTimeout, resetTimeout: Setup.CircuitBreakerSettings.ResetTimeout); - + _writeIsolationLevel = Setup.WriteIsolationLevel; _readIsolationLevel = Setup.ReadIsolationLevel; @@ -808,11 +816,12 @@ private void FailChunkExecution(ChunkExecutionFailure message) var cause = message.Cause is AggregateException aggregateException ? aggregateException.Flatten().InnerExceptions.OfType().FirstOrDefault() - ?? aggregateException.Flatten().InnerExceptions[0] + ?? aggregateException.Flatten().InnerExceptions[0] : message.Cause; - Log.Error(cause, "An error occurred during event batch processing. ChunkId: [{0}], batched requests: [{1}]", message.ChunkId, message.Requests.Length); - + Log.Error(cause, "An error occurred during event batch processing. ChunkId: [{0}], batched requests: [{1}]", + message.ChunkId, message.Requests.Length); + foreach (var request in message.Requests) { switch (request) @@ -826,7 +835,7 @@ private void FailChunkExecution(ChunkExecutionFailure message) aRef.Tell(new WriteMessagesFailed(cause, atomicWriteCount)); foreach (var envelope in req.Messages) { - if (!(envelope is AtomicWrite write)) + if (!(envelope is AtomicWrite write)) continue; var writes = (IImmutableList)write.Payload; @@ -835,19 +844,23 @@ private void FailChunkExecution(ChunkExecutionFailure message) if (cause is DbException) { // database-related exceptions should result in failure - aRef.Tell(new WriteMessageFailure(unadapted, cause, actorInstanceId), unadapted.Sender); + aRef.Tell(new WriteMessageFailure(unadapted, cause, actorInstanceId), + unadapted.Sender); } else { - aRef.Tell(new WriteMessageRejected(unadapted, cause, actorInstanceId), unadapted.Sender); + aRef.Tell(new WriteMessageRejected(unadapted, cause, actorInstanceId), + unadapted.Sender); } } } + break; } case DeleteMessagesTo delete: - delete.PersistentActor.Tell(new DeleteMessagesFailure(cause, delete.ToSequenceNr), ActorRefs.NoSender); + delete.PersistentActor.Tell(new DeleteMessagesFailure(cause, delete.ToSequenceNr), + ActorRefs.NoSender); break; case ReplayMessages replay: @@ -869,7 +882,8 @@ private void FailChunkExecution(ChunkExecutionFailure message) break; default: - Log.Error(cause, $"Batching failure not reported to original sender. Unknown batched persistence journal request type [{request.GetType()}]."); + Log.Error(cause, + $"Batching failure not reported to original sender. Unknown batched persistence journal request type [{request.GetType()}]."); break; } } @@ -916,19 +930,26 @@ protected virtual void OnBufferOverflow(IJournalMessage request) { case WriteMessages msg: var atomicWriteCount = msg.Messages.OfType().Count(); - msg.PersistentActor.Tell(new WriteMessagesFailed(JournalBufferOverflowException.Instance, atomicWriteCount), ActorRefs.NoSender); + msg.PersistentActor.Tell( + new WriteMessagesFailed(JournalBufferOverflowException.Instance, atomicWriteCount), + ActorRefs.NoSender); break; case ReplayMessages msg: - msg.PersistentActor.Tell(new ReplayMessagesFailure(JournalBufferOverflowException.Instance), ActorRefs.NoSender); + msg.PersistentActor.Tell(new ReplayMessagesFailure(JournalBufferOverflowException.Instance), + ActorRefs.NoSender); break; case DeleteMessagesTo msg: - msg.PersistentActor.Tell(new DeleteMessagesFailure(JournalBufferOverflowException.Instance, msg.ToSequenceNr), ActorRefs.NoSender); + msg.PersistentActor.Tell( + new DeleteMessagesFailure(JournalBufferOverflowException.Instance, msg.ToSequenceNr), + ActorRefs.NoSender); break; case ReplayTaggedMessages msg: - msg.ReplyTo.Tell(new ReplayMessagesFailure(JournalBufferOverflowException.Instance), ActorRefs.NoSender); + msg.ReplyTo.Tell(new ReplayMessagesFailure(JournalBufferOverflowException.Instance), + ActorRefs.NoSender); break; case ReplayAllEvents msg: - msg.ReplyTo.Tell(new EventReplayFailure(JournalBufferOverflowException.Instance), ActorRefs.NoSender); + msg.ReplyTo.Tell(new EventReplayFailure(JournalBufferOverflowException.Instance), + ActorRefs.NoSender); break; } } @@ -956,7 +977,9 @@ private async Task ExecuteChunk(RequestChunk chunk, IActorContext // In the grand scheme of thing, using a transaction in an all read batch operation // should not hurt performance by much, because it is done only once at the start. - using (var tx = connection.BeginTransaction(isWriteOperation ? _writeIsolationLevel : _readIsolationLevel)) + using (var tx = connection.BeginTransaction(isWriteOperation + ? _writeIsolationLevel + : _readIsolationLevel)) using (var command = (TCommand)connection.CreateCommand()) { command.CommandTimeout = (int)Setup.ConnectionTimeout.TotalSeconds; @@ -971,7 +994,7 @@ private async Task ExecuteChunk(RequestChunk chunk, IActorContext switch (req) { case WriteMessages msg: - writeResults.Enqueue(await HandleWriteMessages(msg, command)); + writeResults.Enqueue(await HandleWriteMessages(msg, command)); break; case DeleteMessagesTo msg: await HandleDeleteMessagesTo(msg, command); @@ -993,6 +1016,7 @@ private async Task ExecuteChunk(RequestChunk chunk, IActorContext break; } } + tx.Commit(); } catch (Exception e1) @@ -1005,6 +1029,7 @@ private async Task ExecuteChunk(RequestChunk chunk, IActorContext { throw new AggregateException(e2, e1); } + throw; } finally @@ -1091,7 +1116,8 @@ protected virtual async Task ReadHighestSequenceNr(TCommand command) return highestSequenceNr; } - protected virtual async Task HandleSelectCurrentPersistenceIds(SelectCurrentPersistenceIds message, TCommand command) + protected virtual async Task HandleSelectCurrentPersistenceIds(SelectCurrentPersistenceIds message, + TCommand command) { long highestOrderingNumber = await ReadHighestSequenceNr(command); @@ -1279,10 +1305,12 @@ private async Task HandleWriteMessages(WriteMessages req, T tagBuilder.Append(tag).Append(';'); } } + persistent = persistent.WithPayload(tagged.Payload); } - WriteEvent(command, persistent.WithTimestamp(TimestampProvider.GenerateTimestamp(persistent)), tagBuilder.ToString()); + WriteEvent(command, persistent.WithTimestamp(TimestampProvider.GenerateTimestamp(persistent)), + tagBuilder.ToString()); await command.ExecuteNonQueryAsync(); @@ -1364,7 +1392,8 @@ protected virtual IPersistentRepresentation ReadEvent(DbDataReader reader) deserialized = _serialization.Deserialize((byte[])payload, serializerId, manifest); } - return new Persistent(deserialized, sequenceNr, persistenceId, manifest, isDeleted, ActorRefs.NoSender, null, timestamp); + return new Persistent(deserialized, sequenceNr, persistenceId, manifest, isDeleted, ActorRefs.NoSender, + null, timestamp); } /// @@ -1390,7 +1419,7 @@ protected void AddParameter(TCommand command, string paramName, DbType dbType, o param.DbType = dbType; PreAddParameterToCommand(command, param); - + command.Parameters.Add(param); } @@ -1399,8 +1428,10 @@ protected void AddParameter(TCommand command, string paramName, DbType dbType, o /// /// used to define a parameter in. /// Parameter to customize - protected virtual void PreAddParameterToCommand(TCommand command, DbParameter param) { } - + protected virtual void PreAddParameterToCommand(TCommand command, DbParameter param) + { + } + /// /// Select the buffer that has the smallest id on its first item, retrieve a maximum Setup.MaxBatchSize /// items from it, and return it as a chunk that needs to be batched @@ -1418,28 +1449,31 @@ protected virtual void PreAddParameterToCommand(TCommand command, DbParameter pa // We don't batch delete and writes in the same batch, reason being a database // can be deadlocked if write and delete happens in the same transaction var writeType = currentBuffer.Peek().request.GetType(); - while(currentBuffer.Count > 0 && currentBuffer.Peek().request.GetType() == writeType) + while (currentBuffer.Count > 0 && currentBuffer.Peek().request.GetType() == writeType) { operations.Add(currentBuffer.Dequeue().request); if (operations.Count == Setup.MaxBatchSize) break; } + return (new RequestChunk(chunkId, operations.ToArray()), true); } - - while(currentBuffer.Count > 0) + + while (currentBuffer.Count > 0) { operations.Add(currentBuffer.Dequeue().request); if (operations.Count == Setup.MaxBatchSize) break; } + return (new RequestChunk(chunkId, operations.ToArray()), false); } private void CompleteBatch(BatchComplete msg) { _remainingOperations++; - Log.Debug("Completed batch (chunkId: {0}) of {1} operations in {2} milliseconds", msg.ChunkId, msg.OperationCount, msg.TimeSpent.TotalMilliseconds); + Log.Debug("Completed batch (chunkId: {0}) of {1} operations in {2} milliseconds", msg.ChunkId, + msg.OperationCount, msg.TimeSpent.TotalMilliseconds); TryProcess(); } @@ -1453,7 +1487,7 @@ private class WriteMessagesResult public WriteMessagesResult( WriteMessages request, - IEnumerable tags, + IEnumerable tags, IEnumerable persistenceIds) { _request = request; @@ -1516,4 +1550,4 @@ protected JournalBufferOverflowException(SerializationInfo info, StreamingContex { } } -} +} \ No newline at end of file diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/SqlJournal.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/SqlJournal.cs index 7394eb56be5..2115c3ed539 100644 --- a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/SqlJournal.cs +++ b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/SqlJournal.cs @@ -76,11 +76,13 @@ protected override bool ReceivePluginInternal(object message) { case SelectCurrentPersistenceIds msg: SelectAllPersistenceIdsAsync(msg.Offset) - .PipeTo(msg.ReplyTo, success: h => new CurrentPersistenceIds(h.Ids, h.LastOrdering), failure: e => new Status.Failure(e)); + .PipeTo(msg.ReplyTo, success: h => new CurrentPersistenceIds(h.Ids, h.LastOrdering), + failure: e => new Status.Failure(e)); return true; case ReplayTaggedMessages replay: ReplayTaggedMessagesAsync(replay) - .PipeTo(replay.ReplyTo, success: h => new RecoverySuccess(h), failure: e => new ReplayMessagesFailure(e)); + .PipeTo(replay.ReplyTo, success: h => new RecoverySuccess(h), + failure: e => new ReplayMessagesFailure(e)); return true; case ReplayAllEvents replay: ReplayAllEventsAsync(replay) @@ -122,11 +124,13 @@ protected override async Task> WriteMessagesAsync(IEnu else eventToTags.Add(p, ImmutableHashSet.Empty); if (IsTagId(p.PersistenceId)) - throw new InvalidOperationException($"Persistence Id {p.PersistenceId} must not start with {QueryExecutor.Configuration.TagsColumnName}"); + throw new InvalidOperationException( + $"Persistence Id {p.PersistenceId} must not start with {QueryExecutor.Configuration.TagsColumnName}"); } var batch = new WriteJournalBatch(eventToTags); - using(var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) await QueryExecutor.InsertBatchAsync(connection, cancellationToken.Token, batch); } }).ToArray(); @@ -149,15 +153,20 @@ protected virtual async Task ReplayTaggedMessagesAsync(ReplayTaggedMessage using (var connection = CreateDbConnection()) { await connection.OpenAsync(); - using(var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { return await QueryExecutor - .SelectByTagAsync(connection, cancellationToken.Token, replay.Tag, replay.FromOffset, replay.ToOffset, replay.Max, replayedTagged => { - foreach(var adapted in AdaptFromJournal(replayedTagged.Persistent)) - { - replay.ReplyTo.Tell(new ReplayedTaggedMessage(adapted, replayedTagged.Tag, replayedTagged.Offset), ActorRefs.NoSender); - } - }); + .SelectByTagAsync(connection, cancellationToken.Token, replay.Tag, replay.FromOffset, + replay.ToOffset, replay.Max, replayedTagged => + { + foreach (var adapted in AdaptFromJournal(replayedTagged.Persistent)) + { + replay.ReplyTo.Tell( + new ReplayedTaggedMessage(adapted, replayedTagged.Tag, replayedTagged.Offset), + ActorRefs.NoSender); + } + }); } } } @@ -167,34 +176,41 @@ protected virtual async Task ReplayAllEventsAsync(ReplayAllEvents replay) using (var connection = CreateDbConnection()) { await connection.OpenAsync(); - using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { return await QueryExecutor .SelectAllEventsAsync( connection, - cancellationToken.Token, - replay.FromOffset, + cancellationToken.Token, + replay.FromOffset, replay.ToOffset, - replay.Max, - replayedEvent => { + replay.Max, + replayedEvent => + { foreach (var adapted in AdaptFromJournal(replayedEvent.Persistent)) { - replay.ReplyTo.Tell(new ReplayedEvent(adapted, replayedEvent.Offset), ActorRefs.NoSender); + replay.ReplyTo.Tell(new ReplayedEvent(adapted, replayedEvent.Offset), + ActorRefs.NoSender); } }); } } } - protected virtual async Task<(IEnumerable Ids, long LastOrdering)> SelectAllPersistenceIdsAsync(long offset) + protected virtual async Task<(IEnumerable Ids, long LastOrdering)> SelectAllPersistenceIdsAsync( + long offset) { using (var connection = CreateDbConnection()) { await connection.OpenAsync(); - using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { - var lastOrdering = await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token); - var ids = await QueryExecutor.SelectAllPersistenceIdsAsync(connection, cancellationToken.Token, offset); + var lastOrdering = + await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token); + var ids = await QueryExecutor.SelectAllPersistenceIdsAsync(connection, cancellationToken.Token, + offset); return (ids, lastOrdering); } } @@ -210,15 +226,18 @@ protected virtual async Task ReplayAllEventsAsync(ReplayAllEvents replay) /// TBD /// TBD /// TBD - public override async Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, long toSequenceNr, long max, + public override async Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, + long toSequenceNr, long max, Action recoveryCallback) { using (var connection = CreateDbConnection()) { await connection.OpenAsync(); - using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { - await QueryExecutor.SelectByPersistenceIdAsync(connection, cancellationToken.Token, persistenceId, fromSequenceNr, toSequenceNr, max, recoveryCallback); + await QueryExecutor.SelectByPersistenceIdAsync(connection, cancellationToken.Token, persistenceId, + fromSequenceNr, toSequenceNr, max, recoveryCallback); } } } @@ -257,7 +276,7 @@ protected bool WaitingForInitialization(object message) return true; case Status.Failure fail: Log.Error(fail.Cause, "Failure during {0} initialization.", Self); - + // trigger a restart so we have some hope of succeeding in the future even if initialization failed throw new ApplicationException("Failed to initialize SQL Journal.", fail.Cause); default: @@ -268,7 +287,7 @@ protected bool WaitingForInitialization(object message) private async Task Initialize() { - if (!Settings.AutoInitialize) + if (!Settings.AutoInitialize) return new Status.Success(NotUsed.Instance); try @@ -276,7 +295,8 @@ private async Task Initialize() using (var connection = CreateDbConnection()) { await connection.OpenAsync(); - using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { await QueryExecutor.CreateTablesAsync(connection, cancellationToken.Token); } @@ -286,6 +306,7 @@ private async Task Initialize() { return new Status.Failure(e); } + return new Status.Success(NotUsed.Instance); } @@ -328,9 +349,11 @@ protected override async Task DeleteMessagesToAsync(string persistenceId, long t using (var connection = CreateDbConnection()) { await connection.OpenAsync(); - using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { - await QueryExecutor.DeleteBatchAsync(connection, cancellationToken.Token, persistenceId, toSequenceNr); + await QueryExecutor.DeleteBatchAsync(connection, cancellationToken.Token, persistenceId, + toSequenceNr); } } } @@ -346,9 +369,11 @@ public override async Task ReadHighestSequenceNrAsync(string persistenceId using (var connection = CreateDbConnection()) { await connection.OpenAsync(); - using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var cancellationToken = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { - return await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token, persistenceId); + return await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token, + persistenceId); } } } @@ -361,10 +386,13 @@ protected virtual string GetConnectionString() { var connectionString = Settings.ConnectionString; +#if NETSTANDARD if (string.IsNullOrEmpty(connectionString)) { - connectionString = System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString; + connectionString = + System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString; } +#endif return connectionString; } @@ -372,4 +400,4 @@ protected virtual string GetConnectionString() protected ITimestampProvider GetTimestampProvider(string typeName) => TimestampProviderProvider.GetTimestampProvider(typeName, Context); } -} +} \ No newline at end of file diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs index 0ce43491611..98c0b8d2890 100644 --- a/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs +++ b/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs @@ -24,13 +24,16 @@ namespace Akka.Persistence.Sql.Common.Snapshot public abstract class SqlSnapshotStore : SnapshotStore, IWithUnboundedStash { #region messages - + private sealed class Initialized { public static readonly Initialized Instance = new Initialized(); - private Initialized() { } + + private Initialized() + { + } } - + #endregion /// @@ -57,6 +60,7 @@ protected SqlSnapshotStore(Config config) /// TBD /// protected ILoggingAdapter Log => _log ?? (_log ?? Context.GetLogger()); + private ILoggingAdapter _log; /// @@ -114,7 +118,8 @@ private async Task Initialize() try { using (var connection = CreateDbConnection()) - using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var nestedCancellationTokenSource = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { await connection.OpenAsync(nestedCancellationTokenSource.Token); await QueryExecutor.CreateTableAsync(connection, nestedCancellationTokenSource.Token); @@ -129,7 +134,7 @@ private async Task Initialize() private bool WaitingForInitialization(object message) { - switch(message) + switch (message) { case Initialized _: UnbecomeStacked(); @@ -137,7 +142,7 @@ private bool WaitingForInitialization(object message) return true; case Status.Failure msg: Log.Error(msg.Cause, "Error during snapshot store initialization"); - + // trigger a restart so we have some hope of succeeding in the future even if initialization failed throw new ApplicationException("Failed to initialize SQL SnapshotStore.", msg.Cause); default: @@ -154,10 +159,13 @@ protected virtual string GetConnectionString() { var connectionString = Settings.ConnectionString; +#if NETSTANDARD if (string.IsNullOrEmpty(connectionString)) { - connectionString = System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString; + connectionString = + System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString; } +#endif return connectionString; } @@ -168,13 +176,16 @@ protected virtual string GetConnectionString() /// TBD /// TBD /// TBD - protected override async Task LoadAsync(string persistenceId, SnapshotSelectionCriteria criteria) + protected override async Task LoadAsync(string persistenceId, + SnapshotSelectionCriteria criteria) { using (var connection = CreateDbConnection()) - using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var nestedCancellationTokenSource = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { await connection.OpenAsync(nestedCancellationTokenSource.Token); - return await QueryExecutor.SelectSnapshotAsync(connection, nestedCancellationTokenSource.Token, persistenceId, criteria.MaxSequenceNr, criteria.MaxTimeStamp); + return await QueryExecutor.SelectSnapshotAsync(connection, nestedCancellationTokenSource.Token, + persistenceId, criteria.MaxSequenceNr, criteria.MaxTimeStamp); } } @@ -187,7 +198,8 @@ protected override async Task LoadAsync(string persistenceId, protected override async Task SaveAsync(SnapshotMetadata metadata, object snapshot) { using (var connection = CreateDbConnection()) - using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var nestedCancellationTokenSource = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { await connection.OpenAsync(nestedCancellationTokenSource.Token); await QueryExecutor.InsertAsync(connection, nestedCancellationTokenSource.Token, snapshot, metadata); @@ -202,11 +214,13 @@ protected override async Task SaveAsync(SnapshotMetadata metadata, object snapsh protected override async Task DeleteAsync(SnapshotMetadata metadata) { using (var connection = CreateDbConnection()) - using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var nestedCancellationTokenSource = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { await connection.OpenAsync(nestedCancellationTokenSource.Token); DateTime? timestamp = metadata.Timestamp != DateTime.MinValue ? metadata.Timestamp : default(DateTime?); - await QueryExecutor.DeleteAsync(connection, nestedCancellationTokenSource.Token, metadata.PersistenceId, metadata.SequenceNr, timestamp); + await QueryExecutor.DeleteAsync(connection, nestedCancellationTokenSource.Token, metadata.PersistenceId, + metadata.SequenceNr, timestamp); } } @@ -219,11 +233,13 @@ protected override async Task DeleteAsync(SnapshotMetadata metadata) protected override async Task DeleteAsync(string persistenceId, SnapshotSelectionCriteria criteria) { using (var connection = CreateDbConnection()) - using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) + using (var nestedCancellationTokenSource = + CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token)) { await connection.OpenAsync(nestedCancellationTokenSource.Token); - await QueryExecutor.DeleteBatchAsync(connection, nestedCancellationTokenSource.Token, persistenceId, criteria.MaxSequenceNr, criteria.MaxTimeStamp); + await QueryExecutor.DeleteBatchAsync(connection, nestedCancellationTokenSource.Token, persistenceId, + criteria.MaxSequenceNr, criteria.MaxTimeStamp); } } } -} +} \ No newline at end of file diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt index b16d008ad18..31a7a108941 100644 --- a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt +++ b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt @@ -2297,26 +2297,12 @@ namespace Akka.Configuration public static Akka.Configuration.Config FromResource(string resourceName, object instanceInAssembly) { } public static Akka.Configuration.Config FromResource(string resourceName) { } public static Akka.Configuration.Config FromResource(string resourceName, System.Reflection.Assembly assembly) { } - public static Akka.Configuration.Config Load() { } public static Akka.Configuration.Config ParseString(string hocon, System.Func includeCallback) { } public static Akka.Configuration.Config ParseString(string hocon) { } } } namespace Akka.Configuration.Hocon { - public class AkkaConfigurationSection : System.Configuration.ConfigurationSection - { - public AkkaConfigurationSection() { } - public Akka.Configuration.Config AkkaConfig { get; } - [System.Configuration.ConfigurationPropertyAttribute("hocon", IsRequired=true)] - public Akka.Configuration.Hocon.HoconConfigurationElement Hocon { get; set; } - } - public abstract class CDataConfigurationElement : System.Configuration.ConfigurationElement - { - protected const string ContentPropertyName = "content"; - protected CDataConfigurationElement() { } - protected override void DeserializeElement(System.Xml.XmlReader reader, bool serializeCollectionKey) { } - } public class HoconArray : System.Collections.Generic.List, Akka.Configuration.Hocon.IHoconElement { public HoconArray() { } @@ -2326,12 +2312,6 @@ namespace Akka.Configuration.Hocon public bool IsString() { } public override string ToString() { } } - public class HoconConfigurationElement : Akka.Configuration.Hocon.CDataConfigurationElement - { - public HoconConfigurationElement() { } - [System.Configuration.ConfigurationPropertyAttribute("content", IsKey=true, IsRequired=true)] - public string Content { get; set; } - } public class HoconLiteral : Akka.Configuration.Hocon.IHoconElement { public HoconLiteral() { } diff --git a/src/core/Akka.Cluster.Tests/ProviderSelectionSpec.cs b/src/core/Akka.Cluster.Tests/ProviderSelectionSpec.cs index b522fbd7702..9765126f508 100644 --- a/src/core/Akka.Cluster.Tests/ProviderSelectionSpec.cs +++ b/src/core/Akka.Cluster.Tests/ProviderSelectionSpec.cs @@ -16,7 +16,7 @@ namespace Akka.Cluster.Tests public class ProviderSelectionSpec { public ActorSystemSetup Setup { get; } = ActorSystemSetup.Create(); - public Config LocalConfig { get; } = ConfigurationFactory.Load(); + public Config LocalConfig { get; } = ConfigurationFactory.Default(); public Settings SettingsWith(string key) { diff --git a/src/core/Akka.Discovery.Tests/DiscoveryConfigurationSpec.cs b/src/core/Akka.Discovery.Tests/DiscoveryConfigurationSpec.cs index f57e6247c31..148767ace16 100644 --- a/src/core/Akka.Discovery.Tests/DiscoveryConfigurationSpec.cs +++ b/src/core/Akka.Discovery.Tests/DiscoveryConfigurationSpec.cs @@ -47,7 +47,7 @@ public void ServiceDiscovery_should_select_implementation_from_config_by_config_ akka-mock-inside {{ class = ""{className}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); var discovery = Discovery.Get(sys).Default; discovery.Should().BeAssignableTo(); @@ -70,7 +70,7 @@ class = ""{className1}"" mock2 {{ class = ""{className2}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); Discovery.Get(sys).Default.Should().BeAssignableTo(); Discovery.Get(sys).LoadServiceDiscovery("mock2").Should().BeAssignableTo(); @@ -93,7 +93,7 @@ class = ""{className1}"" mock2 {{ class = ""{className2}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); Discovery.Get(sys).LoadServiceDiscovery("mock2").Should().BeSameAs(Discovery.Get(sys).LoadServiceDiscovery("mock2")); Discovery.Get(sys).Default.Should().BeSameAs(Discovery.Get(sys).LoadServiceDiscovery("mock1")); @@ -112,7 +112,7 @@ public void ServiceDiscovery_should_throw_a_specific_discovery_method_exception( mock1 {{ class = ""{className}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); Invoking(() => _ = Discovery.Get(sys).Default) .Should().Throw() @@ -130,7 +130,7 @@ public void ServiceDiscovery_should_throw_an_argument_exception_for_not_existing ConfigurationFactory.ParseString($@" akka.discovery {{ method = ""{className}"" - }}").WithFallback(ConfigurationFactory.Load())); + }}")); Invoking(() => _ = Discovery.Get(sys).Default).Should() .Throw() @@ -148,7 +148,7 @@ public void OneParameterTest() akka-mock-inside {{ class = ""{typeof(TestDiscoveryWithOneParam).TypeQualifiedName()}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); var discovery = Discovery.Get(sys).Default; discovery.Should().BeAssignableTo(); @@ -165,7 +165,7 @@ public void TwoParametersTest() akka-mock-inside {{ class = ""{typeof(TestDiscoveryWithTwoParam).TypeQualifiedName()}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); var discovery = Discovery.Get(sys).Default; discovery.Should().BeAssignableTo(); @@ -182,7 +182,7 @@ public void IllegalParametersTest() akka-mock-inside {{ class = ""{typeof(TestDiscoveryWithIllegalParam).TypeQualifiedName()}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); Invoking(() => _ = Discovery.Get(sys).Default).Should() .Throw() @@ -200,7 +200,7 @@ public void IllegalConstructorTest() akka-mock-inside {{ class = ""{typeof(TestDiscoveryWithIllegalCtor).TypeQualifiedName()}"" }} - }}").WithFallback(ConfigurationFactory.Load())); + }}")); Invoking(() => _ = Discovery.Get(sys).Default).Should() .Throw() diff --git a/src/core/Akka.FSharp.Tests/ApiTests.fs b/src/core/Akka.FSharp.Tests/ApiTests.fs index 5f19361ef64..a63c7a5ff2c 100644 --- a/src/core/Akka.FSharp.Tests/ApiTests.fs +++ b/src/core/Akka.FSharp.Tests/ApiTests.fs @@ -96,19 +96,19 @@ type TestActorWithArgs(arg1, arg2, arg3) = [] let ``can spawn simple actor from expression`` () = - let system = Configuration.load() |> System.create "test" + let system = Configuration.defaultConfig() |> System.create "test" let actor = spawnObj system "test-actor" <@ fun () -> TestActor() @> () [] let ``can spawn actor with constant args from expression`` () = - let system = Configuration.load() |> System.create "test" + let system = Configuration.defaultConfig() |> System.create "test" let actor = spawnObj system "test-actor" <@ fun () -> TestActorWithArgs(box 1, box true, box "yo") @> () [] let ``can spawn actor with captured args from expression`` () = - let system = Configuration.load() |> System.create "test" + let system = Configuration.defaultConfig() |> System.create "test" let arg1 = 1 let arg2 = true let arg3 = "yo" @@ -117,7 +117,7 @@ let ``can spawn actor with captured args from expression`` () = [] let ``cannot spawn actor with simple expr args from expression`` () = - let system = Configuration.load() |> System.create "test" + let system = Configuration.defaultConfig() |> System.create "test" // this formulation is supported in FsApi's expression evaluator, however the checks in Props.Create // do not support this, so we test that we can evaluate this but not actually run it, as a proof of concept Assert.Throws(fun () -> diff --git a/src/core/Akka.FSharp/FsApi.fs b/src/core/Akka.FSharp/FsApi.fs index b19a30e012c..80ff1d6d3b7 100644 --- a/src/core/Akka.FSharp/FsApi.fs +++ b/src/core/Akka.FSharp/FsApi.fs @@ -428,9 +428,6 @@ module Configuration = /// Returns default Akka configuration. let defaultConfig = Akka.Configuration.ConfigurationFactory.Default - - /// Loads Akka configuration from the project's .config file. - let load = Akka.Configuration.ConfigurationFactory.Load module internal OptionHelper = diff --git a/src/core/Akka.Streams.Tests/Dsl/StreamRefsSpec.cs b/src/core/Akka.Streams.Tests/Dsl/StreamRefsSpec.cs index 4907c4f7c81..699bb09b5dc 100644 --- a/src/core/Akka.Streams.Tests/Dsl/StreamRefsSpec.cs +++ b/src/core/Akka.Streams.Tests/Dsl/StreamRefsSpec.cs @@ -206,7 +206,7 @@ public static Config Config() port = {address.Port} hostname = ""{address.Address}"" }} - }}").WithFallback(ConfigurationFactory.Load()); + }}"); } public StreamRefsSpec(ITestOutputHelper output) : this(Config(), output: output) diff --git a/src/core/Akka/Actor/ActorSystem.cs b/src/core/Akka/Actor/ActorSystem.cs index f07635ff42b..b33528da42d 100644 --- a/src/core/Akka/Actor/ActorSystem.cs +++ b/src/core/Akka/Actor/ActorSystem.cs @@ -259,8 +259,11 @@ public static ActorSystem Create(string name, BootstrapSetup setup) public static ActorSystem Create(string name, ActorSystemSetup setup) { var bootstrapSetup = setup.Get(); - var appConfig = bootstrapSetup.FlatSelect(_ => _.Config).GetOrElse(ConfigurationFactory.Load()); - + #if NETSTANDARD + var appConfig = bootstrapSetup.FlatSelect(s => s.Config).GetOrElse(ConfigurationFactory.Load()); + #else + var appConfig = bootstrapSetup.FlatSelect(s => s.Config).GetOrElse(ConfigurationFactory.Default()); + #endif return CreateAndStartSystem(name, appConfig, setup); } diff --git a/src/core/Akka/Configuration/ConfigurationFactory.cs b/src/core/Akka/Configuration/ConfigurationFactory.cs index 2c4d829f5c8..2c31dab5a14 100644 --- a/src/core/Akka/Configuration/ConfigurationFactory.cs +++ b/src/core/Akka/Configuration/ConfigurationFactory.cs @@ -36,9 +36,9 @@ public static Config Empty /// A string that contains configuration options to use. /// callback used to resolve includes /// The configuration defined in the supplied HOCON string. - public static Config ParseString(string hocon, Func includeCallback) + public static Config ParseString(string hocon, Func includeCallback) { - HoconRoot res = Parser.Parse(hocon, includeCallback); + var res = Parser.Parse(hocon, includeCallback); return new Config(res); } @@ -54,6 +54,7 @@ public static Config ParseString(string hocon) return ParseString(hocon, null); } +#if NETSTANDARD /// /// Loads a configuration defined in the current application's /// configuration file, e.g. app.config or web.config @@ -61,9 +62,13 @@ public static Config ParseString(string hocon) /// The configuration defined in the configuration file. public static Config Load() { - var section = (AkkaConfigurationSection)System.Configuration.ConfigurationManager.GetSection("akka") ?? new AkkaConfigurationSection(); + + var section = + (AkkaConfigurationSection)System.Configuration.ConfigurationManager.GetSection("akka") ?? new AkkaConfigurationSection(); return section.AkkaConfig; + } +#endif /// /// Retrieves the default configuration that Akka.NET uses @@ -123,7 +128,7 @@ public static Config FromResource(string resourceName) /// The configuration defined in the assembly that contains the given resource. public static Config FromResource(string resourceName, Assembly assembly) { - using(Stream stream = assembly.GetManifestResourceStream(resourceName)) + using (Stream stream = assembly.GetManifestResourceStream(resourceName)) { Debug.Assert(stream != null, "stream != null"); using (var reader = new StreamReader(stream)) @@ -147,5 +152,4 @@ public static Config FromObject(object source) return config; } } -} - +} \ No newline at end of file diff --git a/src/core/Akka/Configuration/Hocon/AkkaConfigurationSection.cs b/src/core/Akka/Configuration/Hocon/AkkaConfigurationSection.cs index 300fc19c1ff..57a7114fe00 100644 --- a/src/core/Akka/Configuration/Hocon/AkkaConfigurationSection.cs +++ b/src/core/Akka/Configuration/Hocon/AkkaConfigurationSection.cs @@ -9,6 +9,7 @@ namespace Akka.Configuration.Hocon { +#if NETSTANDARD /// /// This class represents a custom akka node within a configuration file. /// @@ -65,4 +66,5 @@ public HoconConfigurationElement Hocon set { base[ConfigurationPropertyName] = value; } } } -} +#endif +} \ No newline at end of file diff --git a/src/core/Akka/Configuration/Hocon/CDataConfigurationElement.cs b/src/core/Akka/Configuration/Hocon/CDataConfigurationElement.cs index 0660af40633..cdc2da2ae3e 100644 --- a/src/core/Akka/Configuration/Hocon/CDataConfigurationElement.cs +++ b/src/core/Akka/Configuration/Hocon/CDataConfigurationElement.cs @@ -10,6 +10,7 @@ namespace Akka.Configuration.Hocon { + #if NETSTANDARD /// /// This class represents the base implementation for retrieving text from /// an XML CDATA node within a configuration file. @@ -62,4 +63,5 @@ protected override void DeserializeElement(XmlReader reader, bool serializeColle reader.ReadEndElement(); } } +#endif } diff --git a/src/core/Akka/Configuration/Hocon/HoconConfigurationElement.cs b/src/core/Akka/Configuration/Hocon/HoconConfigurationElement.cs index e552327d6fa..8f6df480813 100644 --- a/src/core/Akka/Configuration/Hocon/HoconConfigurationElement.cs +++ b/src/core/Akka/Configuration/Hocon/HoconConfigurationElement.cs @@ -9,6 +9,7 @@ namespace Akka.Configuration.Hocon { +#if NETSTANDARD /// /// This class represents a custom HOCON (Human-Optimized Config Object Notation) /// node within a configuration file. @@ -40,4 +41,5 @@ public string Content set { base[ContentPropertyName] = value; } } } -} +#endif +} \ No newline at end of file