This is an automated email from the ASF dual-hosted git repository. blankensteiner pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/pulsar-dotpulsar.git
commit f316f0a920bb22fd2229237724c5a4e1ef497a8b Author: SeĢrgio Silveira <[email protected]> AuthorDate: Tue Mar 31 00:48:53 2020 +0200 reverted some auto changes as suggested. fixed awaitor name to awaiter. more new and advanced contrucsts like switch expressions, ternary operations and coalesce operators. adjusted some rules to not get executed. --- DotPulsar.sln.DotSettings | 18 +++++++--- src/DotPulsar/Internal/AsyncLock.cs | 1 + src/DotPulsar/Internal/{Awaitor.cs => Awaiter.cs} | 4 +-- .../Internal/CancelableCompletionSource.cs | 2 +- src/DotPulsar/Internal/ChannelManager.cs | 6 +--- src/DotPulsar/Internal/ConnectionPool.cs | 8 ++--- src/DotPulsar/Internal/Constants.cs | 5 +-- src/DotPulsar/Internal/Consumer.cs | 3 ++ src/DotPulsar/Internal/ConsumerChannelFactory.cs | 1 + src/DotPulsar/Internal/DefaultExceptionHandler.cs | 42 ++++++++++------------ src/DotPulsar/Internal/Executor.cs | 31 ++++++++-------- .../Internal/Extensions/CommandExtensions.cs | 14 +++----- .../Extensions/ReadOnlySequenceExtensions.cs | 24 +++++-------- src/DotPulsar/Internal/MessagePackage.cs | 2 +- src/DotPulsar/Internal/NotReadyChannel.cs | 2 +- src/DotPulsar/Internal/ProducerBuilder.cs | 6 ++-- src/DotPulsar/Internal/PulsarClientBuilder.cs | 4 +-- src/DotPulsar/Internal/ReaderBuilder.cs | 6 ++-- src/DotPulsar/Internal/ReaderChannelFactory.cs | 1 + src/DotPulsar/Internal/RequestResponseHandler.cs | 4 +-- src/DotPulsar/Internal/Serializer.cs | 10 ++---- src/DotPulsar/Internal/StateTaskCollection.cs | 32 ++++++++--------- 22 files changed, 106 insertions(+), 120 deletions(-) diff --git a/DotPulsar.sln.DotSettings b/DotPulsar.sln.DotSettings index 16b30d6..7bfa62b 100644 --- a/DotPulsar.sln.DotSettings +++ b/DotPulsar.sln.DotSettings @@ -30,13 +30,14 @@ <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/EMPTY_BLOCK_STYLE/@EntryValue">TOGETHER_SAME_LINE</s:String> <s:Int64 x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_BLANK_LINES_IN_CODE/@EntryValue">1</s:Int64> <s:Int64 x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_BLANK_LINES_IN_DECLARATIONS/@EntryValue">1</s:Int64> - <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_EMBEDDED_ARRANGEMENT/@EntryValue">False</s:Boolean> + <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_EMBEDDED_ARRANGEMENT/@EntryValue">True</s:Boolean> + <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_EMBEDDED_BLOCK_ARRANGEMENT/@EntryValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_EXPR_MEMBER_ARRANGEMENT/@EntryValue">False</s:Boolean> - <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_INITIALIZER_ARRANGEMENT/@EntryValue">False</s:Boolean> + <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_INITIALIZER_ARRANGEMENT/@EntryValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_SWITCH_EXPRESSION_ARRANGEMENT/@EntryValue">False</s:Boolean> <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/LINE_FEED_AT_FILE_END/@EntryValue">True</s:Boolean> <s:Int64 x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/MAX_ARRAY_INITIALIZER_ELEMENTS_ON_LINE/@EntryValue">4</s:Int64> - <s:Int64 x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/MAX_INITIALIZER_ELEMENTS_ON_LINE/@EntryValue">1</s:Int64> + <s:Int64 x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/MAX_INITIALIZER_ELEMENTS_ON_LINE/@EntryValue">4</s:Int64> <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSORHOLDER_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String> <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_EXPR_ACCESSOR_ON_SINGLE_LINE/@EntryValue">ALWAYS</s:String> <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_EXPR_METHOD_ON_SINGLE_LINE/@EntryValue">NEVER</s:String> @@ -44,13 +45,20 @@ <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_FIELD_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String> <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_SIMPLE_CASE_STATEMENT_ON_SAME_LINE/@EntryValue">ALWAYS</s:String> <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_SIMPLE_EMBEDDED_STATEMENT_ON_SAME_LINE/@EntryValue">NEVER</s:String> - <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_SIMPLE_INITIALIZER_ON_SINGLE_LINE/@EntryValue">False</s:Boolean> + <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_SIMPLE_INITIALIZER_ON_SINGLE_LINE/@EntryValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_SINGLE_LINE_ARRAY_INITIALIZER_BRACES/@EntryValue">True</s:Boolean> + <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_ARRAY_INITIALIZER_STYLE/@EntryValue">CHOP_IF_LONG</s:String> <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_BEFORE_ARROW_WITH_EXPRESSIONS/@EntryValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_BEFORE_EXTENDS_COLON/@EntryValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_BEFORE_FIRST_TYPE_PARAMETER_CONSTRAINT/@EntryValue">True</s:Boolean> - <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_OBJECT_AND_COLLECTION_INITIALIZER_STYLE/@EntryValue">CHOP_ALWAYS</s:String> + <s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_OBJECT_AND_COLLECTION_INITIALIZER_STYLE/@EntryValue">CHOP_IF_LONG</s:String> <s:Boolean x:Key="/Default/CodeStyle/CSharpUsing/AddImportsToDeepestScope/@EntryValue">True</s:Boolean> <s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=SSL/@EntryIndexedValue">SSL</s:String> + <s:String x:Key="/Default/CustomTools/CustomToolsData/@EntryValue"></s:String> + <s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean> + <s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean> + <s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpUseContinuousIndentInsideBracesMigration/@EntryIndexedValue">True</s:Boolean> + <s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean> + <s:Boolean x:Key="/Default/UserDictionary/Words/=awaiters/@EntryIndexedValue">True</s:Boolean> <s:Boolean x:Key="/Default/UserDictionary/Words/=Xunit/@EntryIndexedValue">True</s:Boolean> </wpf:ResourceDictionary> diff --git a/src/DotPulsar/Internal/AsyncLock.cs b/src/DotPulsar/Internal/AsyncLock.cs index f2732f2..b95f422 100644 --- a/src/DotPulsar/Internal/AsyncLock.cs +++ b/src/DotPulsar/Internal/AsyncLock.cs @@ -74,6 +74,7 @@ namespace DotPulsar.Internal } await _semaphoreSlim.WaitAsync().ConfigureAwait(false); //Wait for possible lock-holder to finish + _semaphoreSlim.Release(); _semaphoreSlim.Dispose(); } diff --git a/src/DotPulsar/Internal/Awaitor.cs b/src/DotPulsar/Internal/Awaiter.cs similarity index 94% rename from src/DotPulsar/Internal/Awaitor.cs rename to src/DotPulsar/Internal/Awaiter.cs index 130e467..9f7a8c4 100644 --- a/src/DotPulsar/Internal/Awaitor.cs +++ b/src/DotPulsar/Internal/Awaiter.cs @@ -18,11 +18,11 @@ namespace DotPulsar.Internal using System.Collections.Concurrent; using System.Threading.Tasks; - public sealed class Awaitor<T, TResult> : IDisposable + public sealed class Awaiter<T, TResult> : IDisposable { private readonly ConcurrentDictionary<T, TaskCompletionSource<TResult>> _items; - public Awaitor() + public Awaiter() => _items = new ConcurrentDictionary<T, TaskCompletionSource<TResult>>(); public Task<TResult> CreateTask(T item) diff --git a/src/DotPulsar/Internal/CancelableCompletionSource.cs b/src/DotPulsar/Internal/CancelableCompletionSource.cs index c099aa3..ff66a9a 100644 --- a/src/DotPulsar/Internal/CancelableCompletionSource.cs +++ b/src/DotPulsar/Internal/CancelableCompletionSource.cs @@ -27,7 +27,7 @@ namespace DotPulsar.Internal => _source = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously); public void SetupCancellation(Action callback, CancellationToken token) - => _registration = token.Register(() => callback()); + => _registration = token.Register(callback); public void SetResult(T result) => _ = _source.TrySetResult(result); diff --git a/src/DotPulsar/Internal/ChannelManager.cs b/src/DotPulsar/Internal/ChannelManager.cs index 9667d53..9eaf603 100644 --- a/src/DotPulsar/Internal/ChannelManager.cs +++ b/src/DotPulsar/Internal/ChannelManager.cs @@ -103,11 +103,7 @@ namespace DotPulsar.Internal _ = response.ContinueWith(result => { if (result.Result.CommandType == BaseCommand.Type.Success) - { - var channel = _consumerChannels.Remove(consumerId); - - channel?.Unsubscribed(); - } + _consumerChannels.Remove(consumerId)?.Unsubscribed(); }, TaskContinuationOptions.OnlyOnRanToCompletion); } diff --git a/src/DotPulsar/Internal/ConnectionPool.cs b/src/DotPulsar/Internal/ConnectionPool.cs index 03773c4..cf53678 100644 --- a/src/DotPulsar/Internal/ConnectionPool.cs +++ b/src/DotPulsar/Internal/ConnectionPool.cs @@ -75,6 +75,7 @@ namespace DotPulsar.Internal { var connection = await GetConnection(serviceUrl, cancellationToken).ConfigureAwait(false); var response = await connection.Send(lookup, cancellationToken).ConfigureAwait(false); + response.Expect(BaseCommand.Type.LookupResponse); if (response.LookupTopicResponse.Response == CommandLookupTopicResponse.LookupType.Failed) @@ -87,10 +88,8 @@ namespace DotPulsar.Internal if (response.LookupTopicResponse.Response == CommandLookupTopicResponse.LookupType.Redirect || !response.LookupTopicResponse.Authoritative) continue; - if (_serviceUrl.IsLoopback) // LookupType is 'Connect', ServiceUrl is local and response is authoritative. Assume the Pulsar server is a standalone docker. - return connection; - else - return await GetConnection(serviceUrl, cancellationToken).ConfigureAwait(false); + // LookupType is 'Connect', ServiceUrl is local and response is authoritative. Assume the Pulsar server is a standalone docker. + return _serviceUrl.IsLoopback ? connection : await GetConnection(serviceUrl, cancellationToken).ConfigureAwait(false); } } @@ -164,6 +163,7 @@ namespace DotPulsar.Internal var connection = _connections[serviceUrl]; if (connection is null) continue; + if (!await connection.HasChannels(cancellationToken).ConfigureAwait(false)) await DisposeConnection(serviceUrl).ConfigureAwait(false); } diff --git a/src/DotPulsar/Internal/Constants.cs b/src/DotPulsar/Internal/Constants.cs index 4ff4758..ea0c44c 100644 --- a/src/DotPulsar/Internal/Constants.cs +++ b/src/DotPulsar/Internal/Constants.cs @@ -27,10 +27,7 @@ namespace DotPulsar.Internal PulsarSslScheme = "pulsar+ssl"; DefaultPulsarPort = 6650; DefaultPulsarSSLPort = 6651; - MagicNumber = new byte[] - { - 0x0e, 0x01 - }; + MagicNumber = new byte[] { 0x0e, 0x01 }; MetadataSizeOffset = 6; MetadataOffset = 10; } diff --git a/src/DotPulsar/Internal/Consumer.cs b/src/DotPulsar/Internal/Consumer.cs index 5fae580..4f895f3 100644 --- a/src/DotPulsar/Internal/Consumer.cs +++ b/src/DotPulsar/Internal/Consumer.cs @@ -97,6 +97,7 @@ namespace DotPulsar.Internal public async ValueTask Unsubscribe(CancellationToken cancellationToken) { ThrowIfDisposed(); + _ = await _executor.Execute(() => _channel.Send(new CommandUnsubscribe(), cancellationToken), cancellationToken).ConfigureAwait(false); } @@ -108,6 +109,7 @@ namespace DotPulsar.Internal { MessageId = messageId.Data }; + _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false); } @@ -117,6 +119,7 @@ namespace DotPulsar.Internal var response = await _executor.Execute(() => _channel.Send(new CommandGetLastMessageId(), cancellationToken), cancellationToken) .ConfigureAwait(false); + return new MessageId(response.LastMessageId); } diff --git a/src/DotPulsar/Internal/ConsumerChannelFactory.cs b/src/DotPulsar/Internal/ConsumerChannelFactory.cs index 183c0e3..b15734b 100644 --- a/src/DotPulsar/Internal/ConsumerChannelFactory.cs +++ b/src/DotPulsar/Internal/ConsumerChannelFactory.cs @@ -66,6 +66,7 @@ namespace DotPulsar.Internal var messageQueue = new AsyncQueue<MessagePackage>(); var channel = new Channel(_correlationId, _eventRegister, messageQueue); var response = await connection.Send(_subscribe, channel, cancellationToken).ConfigureAwait(false); + return new ConsumerChannel(response.ConsumerId, _messagePrefetchCount, messageQueue, connection, _batchHandler); } } diff --git a/src/DotPulsar/Internal/DefaultExceptionHandler.cs b/src/DotPulsar/Internal/DefaultExceptionHandler.cs index b0a44c3..6e4821e 100644 --- a/src/DotPulsar/Internal/DefaultExceptionHandler.cs +++ b/src/DotPulsar/Internal/DefaultExceptionHandler.cs @@ -40,29 +40,25 @@ namespace DotPulsar.Internal } private FaultAction DetermineFaultAction(Exception exception, CancellationToken cancellationToken) - { - switch (exception) + => exception switch { - case TooManyRequestsException _: return FaultAction.Retry; - case ChannelNotReadyException _: return FaultAction.Retry; - case ServiceNotReadyException _: return FaultAction.Retry; - case ConnectionDisposedException _: return FaultAction.Retry; - case AsyncLockDisposedException _: return FaultAction.Retry; - case PulsarStreamDisposedException _: return FaultAction.Retry; - case AsyncQueueDisposedException _: return FaultAction.Retry; - case OperationCanceledException _: return cancellationToken.IsCancellationRequested ? FaultAction.Rethrow : FaultAction.Retry; - case DotPulsarException _: return FaultAction.Rethrow; - case SocketException socketException: - return socketException.SocketErrorCode switch - { - SocketError.HostNotFound => FaultAction.Rethrow, - SocketError.HostUnreachable => FaultAction.Rethrow, - SocketError.NetworkUnreachable => FaultAction.Rethrow, - _ => FaultAction.Retry - }; - } - - return FaultAction.Rethrow; - } + TooManyRequestsException _ => FaultAction.Retry, + ChannelNotReadyException _ => FaultAction.Retry, + ServiceNotReadyException _ => FaultAction.Retry, + ConnectionDisposedException _ => FaultAction.Retry, + AsyncLockDisposedException _ => FaultAction.Retry, + PulsarStreamDisposedException _ => FaultAction.Retry, + AsyncQueueDisposedException _ => FaultAction.Retry, + OperationCanceledException _ => cancellationToken.IsCancellationRequested ? FaultAction.Rethrow : FaultAction.Retry, + DotPulsarException _ => FaultAction.Rethrow, + SocketException socketException => socketException.SocketErrorCode switch + { + SocketError.HostNotFound => FaultAction.Rethrow, + SocketError.HostUnreachable => FaultAction.Rethrow, + SocketError.NetworkUnreachable => FaultAction.Rethrow, + _ => FaultAction.Retry + }, + _ => FaultAction.Rethrow + }; } } diff --git a/src/DotPulsar/Internal/Executor.cs b/src/DotPulsar/Internal/Executor.cs index 6463ce5..64e7a76 100644 --- a/src/DotPulsar/Internal/Executor.cs +++ b/src/DotPulsar/Internal/Executor.cs @@ -43,9 +43,9 @@ namespace DotPulsar.Internal action(); return; } - catch (Exception exception) + catch (Exception ex) { - if (await Handle(exception, cancellationToken).ConfigureAwait(false)) + if (await Handle(ex, cancellationToken).ConfigureAwait(false)) throw; } @@ -62,9 +62,9 @@ namespace DotPulsar.Internal await func().ConfigureAwait(false); return; } - catch (Exception exception) + catch (Exception ex) { - if (await Handle(exception, cancellationToken).ConfigureAwait(false)) + if (await Handle(ex, cancellationToken).ConfigureAwait(false)) throw; } @@ -81,9 +81,9 @@ namespace DotPulsar.Internal await func().ConfigureAwait(false); return; } - catch (Exception exception) + catch (Exception ex) { - if (await Handle(exception, cancellationToken).ConfigureAwait(false)) + if (await Handle(ex, cancellationToken).ConfigureAwait(false)) throw; } @@ -99,9 +99,9 @@ namespace DotPulsar.Internal { return func(); } - catch (Exception exception) + catch (Exception ex) { - if (await Handle(exception, cancellationToken).ConfigureAwait(false)) + if (await Handle(ex, cancellationToken).ConfigureAwait(false)) throw; } @@ -117,9 +117,9 @@ namespace DotPulsar.Internal { return await func().ConfigureAwait(false); } - catch (Exception exception) + catch (Exception ex) { - if (await Handle(exception, cancellationToken).ConfigureAwait(false)) + if (await Handle(ex, cancellationToken).ConfigureAwait(false)) throw; } @@ -135,9 +135,9 @@ namespace DotPulsar.Internal { return await func().ConfigureAwait(false); } - catch (Exception exception) + catch (Exception ex) { - if (await Handle(exception, cancellationToken).ConfigureAwait(false)) + if (await Handle(ex, cancellationToken).ConfigureAwait(false)) throw; } @@ -157,10 +157,9 @@ namespace DotPulsar.Internal if (context.Result != FaultAction.Retry) _eventRegister.Register(new ExecutorFaulted(_correlationId)); - if (context.Result == FaultAction.ThrowException) - throw context.Exception; - - return context.Result == FaultAction.Rethrow; + return context.Result == FaultAction.ThrowException + ? throw context.Exception + : context.Result == FaultAction.Rethrow; } } } diff --git a/src/DotPulsar/Internal/Extensions/CommandExtensions.cs b/src/DotPulsar/Internal/Extensions/CommandExtensions.cs index 8cf27ab..9e595a4 100644 --- a/src/DotPulsar/Internal/Extensions/CommandExtensions.cs +++ b/src/DotPulsar/Internal/Extensions/CommandExtensions.cs @@ -25,16 +25,12 @@ namespace DotPulsar.Internal.Extensions if (command.CommandType == type) return; - switch (command.CommandType) - { - case BaseCommand.Type.Error: - command.Error.Throw(); - return; - case BaseCommand.Type.SendError: - command.SendError.Throw(); - return; - } + if (command.CommandType == BaseCommand.Type.Error) + command.Error.Throw(); + if (command.CommandType == BaseCommand.Type.SendError) + command.SendError.Throw(); + throw new UnexpectedResponseException($"Expected '{type}' but got '{command.CommandType}'"); } diff --git a/src/DotPulsar/Internal/Extensions/ReadOnlySequenceExtensions.cs b/src/DotPulsar/Internal/Extensions/ReadOnlySequenceExtensions.cs index 86cb043..ea5fa12 100644 --- a/src/DotPulsar/Internal/Extensions/ReadOnlySequenceExtensions.cs +++ b/src/DotPulsar/Internal/Extensions/ReadOnlySequenceExtensions.cs @@ -70,28 +70,20 @@ namespace DotPulsar.Internal.Extensions switch (read) { case 0: - if (reverse) - union.B0 = span[i]; - else - union.B3 = span[i]; + if (reverse) union.B0 = span[i]; + else union.B3 = span[i]; continue; case 1: - if (reverse) - union.B1 = span[i]; - else - union.B2 = span[i]; + if (reverse) union.B1 = span[i]; + else union.B2 = span[i]; continue; case 2: - if (reverse) - union.B2 = span[i]; - else - union.B1 = span[i]; + if (reverse) union.B2 = span[i]; + else union.B1 = span[i]; continue; case 3: - if (reverse) - union.B3 = span[i]; - else - union.B0 = span[i]; + if (reverse) union.B3 = span[i]; + else union.B0 = span[i]; break; } } diff --git a/src/DotPulsar/Internal/MessagePackage.cs b/src/DotPulsar/Internal/MessagePackage.cs index c2d14d7..e8b249b 100644 --- a/src/DotPulsar/Internal/MessagePackage.cs +++ b/src/DotPulsar/Internal/MessagePackage.cs @@ -17,7 +17,7 @@ namespace DotPulsar.Internal using System.Buffers; using PulsarApi; - public struct MessagePackage + public readonly struct MessagePackage { public MessagePackage(MessageIdData messageId, ReadOnlySequence<byte> data) { diff --git a/src/DotPulsar/Internal/NotReadyChannel.cs b/src/DotPulsar/Internal/NotReadyChannel.cs index 0c4b728..e7c2f3d 100644 --- a/src/DotPulsar/Internal/NotReadyChannel.cs +++ b/src/DotPulsar/Internal/NotReadyChannel.cs @@ -48,7 +48,7 @@ namespace DotPulsar.Internal public Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command, CancellationToken cancellationToken) => throw GetException(); - private Exception GetException() + private static Exception GetException() => new ChannelNotReadyException(); } } diff --git a/src/DotPulsar/Internal/ProducerBuilder.cs b/src/DotPulsar/Internal/ProducerBuilder.cs index 40f4599..8916cd0 100644 --- a/src/DotPulsar/Internal/ProducerBuilder.cs +++ b/src/DotPulsar/Internal/ProducerBuilder.cs @@ -53,11 +53,13 @@ namespace DotPulsar.Internal if (string.IsNullOrEmpty(_topic)) throw new ConfigurationException("ProducerOptions.Topic may not be null or empty"); - return _pulsarClient.CreateProducer(new ProducerOptions(_topic!) + var options = new ProducerOptions(_topic!) { InitialSequenceId = _initialSequenceId, ProducerName = _producerName - }); + }; + + return _pulsarClient.CreateProducer(options); } } } diff --git a/src/DotPulsar/Internal/PulsarClientBuilder.cs b/src/DotPulsar/Internal/PulsarClientBuilder.cs index bd1b166..387e87d 100644 --- a/src/DotPulsar/Internal/PulsarClientBuilder.cs +++ b/src/DotPulsar/Internal/PulsarClientBuilder.cs @@ -145,11 +145,9 @@ namespace DotPulsar.Internal { throw new InvalidSchemeException($"Invalid scheme '{scheme}'. Expected '{Constants.PulsarScheme}' or '{Constants.PulsarSslScheme}'"); } - - var connector = new Connector(_clientCertificates, _trustedCertificateAuthority, _verifyCertificateAuthority, _verifyCertificateName); + var connector = new Connector(_clientCertificates, _trustedCertificateAuthority, _verifyCertificateAuthority, _verifyCertificateName); var connectionPool = new ConnectionPool(_commandConnect, _serviceUrl, connector, _encryptionPolicy.Value, _closeInactiveConnectionsInterval); - var processManager = new ProcessManager(connectionPool); var exceptionHandlerPipeline = new ExceptionHandlerPipeline(new List<IHandleException>(_exceptionHandlers) diff --git a/src/DotPulsar/Internal/ReaderBuilder.cs b/src/DotPulsar/Internal/ReaderBuilder.cs index 15b7c64..41ff928 100644 --- a/src/DotPulsar/Internal/ReaderBuilder.cs +++ b/src/DotPulsar/Internal/ReaderBuilder.cs @@ -71,12 +71,14 @@ namespace DotPulsar.Internal if (string.IsNullOrEmpty(_topic)) throw new ConfigurationException("Topic may not be null or empty"); - return _pulsarClient.CreateReader(new ReaderOptions(_startMessageId, _topic!) + var options = new ReaderOptions(_startMessageId, _topic!) { MessagePrefetchCount = _messagePrefetchCount, ReadCompacted = _readCompacted, ReaderName = _readerName - }); + }; + + return _pulsarClient.CreateReader(options); } } } diff --git a/src/DotPulsar/Internal/ReaderChannelFactory.cs b/src/DotPulsar/Internal/ReaderChannelFactory.cs index b22aa76..447119a 100644 --- a/src/DotPulsar/Internal/ReaderChannelFactory.cs +++ b/src/DotPulsar/Internal/ReaderChannelFactory.cs @@ -65,6 +65,7 @@ namespace DotPulsar.Internal var messageQueue = new AsyncQueue<MessagePackage>(); var channel = new Channel(_correlationId, _eventRegister, messageQueue); var response = await connection.Send(_subscribe, channel, cancellationToken).ConfigureAwait(false); + return new ConsumerChannel(response.ConsumerId, _messagePrefetchCount, messageQueue, connection, _batchHandler); } } diff --git a/src/DotPulsar/Internal/RequestResponseHandler.cs b/src/DotPulsar/Internal/RequestResponseHandler.cs index 5b7ce56..3b57a44 100644 --- a/src/DotPulsar/Internal/RequestResponseHandler.cs +++ b/src/DotPulsar/Internal/RequestResponseHandler.cs @@ -22,12 +22,12 @@ namespace DotPulsar.Internal { private const string ConnectResponseIdentifier = "Connected"; - private readonly Awaitor<string, BaseCommand> _responses; + private readonly Awaiter<string, BaseCommand> _responses; private ulong _requestId; public RequestResponseHandler() { - _responses = new Awaitor<string, BaseCommand>(); + _responses = new Awaiter<string, BaseCommand>(); _requestId = 1; } diff --git a/src/DotPulsar/Internal/Serializer.cs b/src/DotPulsar/Internal/Serializer.cs index a7fc650..ca5f108 100644 --- a/src/DotPulsar/Internal/Serializer.cs +++ b/src/DotPulsar/Internal/Serializer.cs @@ -65,14 +65,8 @@ namespace DotPulsar.Internal var union = new UIntUnion(integer); return BitConverter.IsLittleEndian - ? new[] - { - union.B3, union.B2, union.B1, union.B0 - } - : new[] - { - union.B0, union.B1, union.B2, union.B3 - }; + ? new[] { union.B3, union.B2, union.B1, union.B0 } + : new[] { union.B0, union.B1, union.B2, union.B3 }; } private static byte[] Serialize<T>(T item) diff --git a/src/DotPulsar/Internal/StateTaskCollection.cs b/src/DotPulsar/Internal/StateTaskCollection.cs index 7606f02..745777a 100644 --- a/src/DotPulsar/Internal/StateTaskCollection.cs +++ b/src/DotPulsar/Internal/StateTaskCollection.cs @@ -21,12 +21,12 @@ namespace DotPulsar.Internal public sealed class StateTaskCollection<TState> where TState : notnull { private readonly object _lock; - private readonly LinkedList<StateTask<TState>> _awaitors; + private readonly LinkedList<StateTask<TState>> _awaiters; public StateTaskCollection() { _lock = new object(); - _awaitors = new LinkedList<StateTask<TState>>(); + _awaiters = new LinkedList<StateTask<TState>>(); } public Task<TState> CreateTaskFor(TState state, StateChanged changed, CancellationToken cancellationToken) @@ -35,7 +35,7 @@ namespace DotPulsar.Internal lock (_lock) { - node = _awaitors.AddFirst(new StateTask<TState>(state, changed)); + node = _awaiters.AddFirst(new StateTask<TState>(state, changed)); } node.Value.CancelableCompletionSource.SetupCancellation(() => TaskWasCanceled(node), cancellationToken); @@ -47,20 +47,20 @@ namespace DotPulsar.Internal { lock (_lock) { - var awaitor = _awaitors.First; + var awaiter = _awaiters.First; - while (awaitor != null) + while (awaiter != null) { - var next = awaitor.Next; + var next = awaiter.Next; - if (awaitor.Value.IsAwaiting(state)) + if (awaiter.Value.IsAwaiting(state)) { - _awaitors.Remove(awaitor); - awaitor.Value.CancelableCompletionSource.SetResult(state); - awaitor.Value.CancelableCompletionSource.Dispose(); + _awaiters.Remove(awaiter); + awaiter.Value.CancelableCompletionSource.SetResult(state); + awaiter.Value.CancelableCompletionSource.Dispose(); } - awaitor = next; + awaiter = next; } } } @@ -69,13 +69,13 @@ namespace DotPulsar.Internal { lock (_lock) { - foreach (var awaitor in _awaitors) + foreach (var awaiter in _awaiters) { - awaitor.CancelableCompletionSource.SetResult(state); - awaitor.CancelableCompletionSource.Dispose(); + awaiter.CancelableCompletionSource.SetResult(state); + awaiter.CancelableCompletionSource.Dispose(); } - _awaitors.Clear(); + _awaiters.Clear(); } } @@ -85,7 +85,7 @@ namespace DotPulsar.Internal { try { - _awaitors.Remove(node); + _awaiters.Remove(node); node.Value.Dispose(); } catch
