blankensteiner commented on a change in pull request #71: URL: https://github.com/apache/pulsar-dotpulsar/pull/71#discussion_r643908350
########## File path: src/DotPulsar/SinglePartitionRouter.cs ########## @@ -0,0 +1,43 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar +{ + using Abstractions; + using HashDepot; + using System; + using System.Text; + + public sealed class SinglePartitionRouter : IMessageRouter + { + private int? _partitionIndex; + + public SinglePartitionRouter(int? partitionIndex = null) Review comment: Do we want this as public? ########## File path: tests/DotPulsar.Tests/Internal/PartitionedProducerProcessTests.cs ########## @@ -0,0 +1,102 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar.Tests.Internal +{ + using Abstractions; + using DotPulsar.Internal; + using DotPulsar.Internal.Abstractions; + using DotPulsar.Internal.Events; + using NSubstitute; + using System; + using System.Collections.Concurrent; + using System.Collections.Generic; + using System.Threading.Tasks; + using Xunit; + + public class PartitionedProducerProcessTests + { + [Fact] + public async Task TestPartitionedProducerStateManage_WhenSubProducersStateChange_ThenPartitionedProducerStateChangeCorrectly() Review comment: Let's fix the warnings from GitHub Actions ########## File path: src/DotPulsar/PulsarClient.cs ########## @@ -57,26 +58,84 @@ public sealed class PulsarClient : IPulsarClient public static IPulsarClientBuilder Builder() => new PulsarClientBuilder(); + public async Task<uint> GetNumberOfPartitions(string topic, CancellationToken cancellationToken) + { + var connection = await _connectionPool.FindConnectionForTopic(topic, cancellationToken).ConfigureAwait(false); + var commandPartitionedMetadata = new CommandPartitionedTopicMetadata() { Topic = topic }; + var response = await connection.Send(commandPartitionedMetadata, cancellationToken).ConfigureAwait(false); + + response.Expect(BaseCommand.Type.PartitionedMetadataResponse); + + if (response.PartitionMetadataResponse.Response == CommandPartitionedTopicMetadataResponse.LookupType.Failed) + response.PartitionMetadataResponse.Throw(); + + return response.PartitionMetadataResponse.Partitions; + } + /// <summary> /// Create a producer. /// </summary> public IProducer<TMessage> CreateProducer<TMessage>(ProducerOptions<TMessage> options) { ThrowIfDisposed(); + var partitionsCount = GetNumberOfPartitions(options.Topic, default).Result; Review comment: In regards to this and the ProducerProcess and PartitionedProducer, I think we need to rethink it. If we want to be really resilient, we don't just want to support resizing a partitioned topic on the fly, but even support the fact that a topic can change to and from a partitioned topic. This means that the "GetNumberOfPartitions" lookup is just part of the (re)connect logic. As such, it no longer makes sense to talk about a Producer and a PartitionedProducer, since they are the same. So the current 'Producer' could be renamed to 'SubProducer' and 'PartitionedProducer' to 'Producer'. A non-partitioned producer is then just a Producer with 1 and only 1 SubProducer. Let me know what you think about this. ########## File path: src/DotPulsar/Internal/Abstractions/Process.cs ########## @@ -68,11 +68,16 @@ public void Handle(IEvent e) case ChannelUnsubscribed _: ChannelState = ChannelState.Unsubscribed; break; + default:HandleExtend(e); Review comment: If you run the automatic code cleanup and fix the warnings, then small styling issues like these will be solved :) ########## File path: src/DotPulsar/RoundRobinPartitionRouter.cs ########## @@ -0,0 +1,36 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar +{ + using Abstractions; + using HashDepot; + using System.Text; + using System.Threading; + + public sealed class RoundRobinPartitionRouter : IMessageRouter Review comment: Since this is public we need some documentation ########## File path: src/DotPulsar/SinglePartitionRouter.cs ########## @@ -0,0 +1,43 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar +{ + using Abstractions; + using HashDepot; + using System; + using System.Text; + + public sealed class SinglePartitionRouter : IMessageRouter Review comment: Since this is public we need some documentation ########## File path: src/DotPulsar/Internal/PartitionedProducer.cs ########## @@ -0,0 +1,96 @@ +namespace DotPulsar.Internal +{ + using Abstractions; + using DotPulsar.Abstractions; + using Events; + using System; + using System.Collections.Concurrent; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + + public sealed class PartitionedProducer<TMessage> : IProducer<TMessage> + { + private readonly Guid _correlationId; + private readonly IRegisterEvent _eventRegister; + private readonly IStateChanged<ProducerState> _state; + private readonly PulsarClient _pulsarClient; + private readonly ProducerOptions<TMessage> _options; + private readonly ConcurrentDictionary<int, IProducer<TMessage>> _producers; + private readonly IMessageRouter _messageRouter; + private readonly CancellationTokenSource _cts = new(); + private readonly int _producersCount; + private int _isDisposed; + public Uri ServiceUrl { get; } + public string Topic { get; } + + public PartitionedProducer( + Guid correlationId, + Uri serviceUrl, + string topic, + IRegisterEvent registerEvent, + IStateChanged<ProducerState> state, + uint partitionsCount, + ProducerOptions<TMessage> options, + PulsarClient pulsarClient + ) + { + _correlationId = correlationId; + ServiceUrl = serviceUrl; + Topic = topic; + _eventRegister = registerEvent; + _state = state; + _isDisposed = 0; + _options = options; + _pulsarClient = pulsarClient; + _producersCount = (int) partitionsCount; + _messageRouter = options.MessageRouter; + + _producers = new ConcurrentDictionary<int, IProducer<TMessage>>(Environment.ProcessorCount, _producersCount); + CreateSubProducers(0, _producersCount); + } + + private void CreateSubProducers(int startIndex, int count) + { + for (int i = 0; i < count; i++) + { + var producer = _pulsarClient.NewProducer(Topic, _options, (uint)(i+startIndex), _correlationId); + _producers[i+startIndex] = producer; + } + } + + public bool IsFinalState() + => _state.IsFinalState(); + + public bool IsFinalState(ProducerState state) + => _state.IsFinalState(state); + + public async ValueTask<ProducerState> OnStateChangeTo(ProducerState state, CancellationToken cancellationToken = default) + => await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false); + + public async ValueTask<ProducerState> OnStateChangeFrom(ProducerState state, CancellationToken cancellationToken = default) + => await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false); + + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _isDisposed, 1) != 0) + return; + + _cts.Cancel(); + _cts.Dispose(); + + foreach (var producer in _producers.Values) + { + await producer.DisposeAsync().ConfigureAwait(false); + } + + _eventRegister.Register(new ProducerDisposed(_correlationId)); + } + + public async ValueTask<MessageId> Send(TMessage message, CancellationToken cancellationToken = default) + => await _producers[_messageRouter.ChoosePartition(null, _producersCount)].Send(message, cancellationToken); Review comment: Missing ConfigureAwait(false); ########## File path: src/DotPulsar/Internal/PartitionedProducer.cs ########## @@ -0,0 +1,96 @@ +namespace DotPulsar.Internal +{ + using Abstractions; + using DotPulsar.Abstractions; + using Events; + using System; + using System.Collections.Concurrent; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + + public sealed class PartitionedProducer<TMessage> : IProducer<TMessage> + { + private readonly Guid _correlationId; + private readonly IRegisterEvent _eventRegister; + private readonly IStateChanged<ProducerState> _state; + private readonly PulsarClient _pulsarClient; + private readonly ProducerOptions<TMessage> _options; + private readonly ConcurrentDictionary<int, IProducer<TMessage>> _producers; + private readonly IMessageRouter _messageRouter; + private readonly CancellationTokenSource _cts = new(); + private readonly int _producersCount; + private int _isDisposed; + public Uri ServiceUrl { get; } + public string Topic { get; } + + public PartitionedProducer( + Guid correlationId, + Uri serviceUrl, + string topic, + IRegisterEvent registerEvent, + IStateChanged<ProducerState> state, + uint partitionsCount, + ProducerOptions<TMessage> options, + PulsarClient pulsarClient + ) + { + _correlationId = correlationId; + ServiceUrl = serviceUrl; + Topic = topic; + _eventRegister = registerEvent; + _state = state; + _isDisposed = 0; + _options = options; + _pulsarClient = pulsarClient; + _producersCount = (int) partitionsCount; + _messageRouter = options.MessageRouter; + + _producers = new ConcurrentDictionary<int, IProducer<TMessage>>(Environment.ProcessorCount, _producersCount); + CreateSubProducers(0, _producersCount); + } + + private void CreateSubProducers(int startIndex, int count) + { + for (int i = 0; i < count; i++) + { + var producer = _pulsarClient.NewProducer(Topic, _options, (uint)(i+startIndex), _correlationId); + _producers[i+startIndex] = producer; + } + } + + public bool IsFinalState() + => _state.IsFinalState(); + + public bool IsFinalState(ProducerState state) + => _state.IsFinalState(state); + + public async ValueTask<ProducerState> OnStateChangeTo(ProducerState state, CancellationToken cancellationToken = default) + => await _state.StateChangedTo(state, cancellationToken).ConfigureAwait(false); + + public async ValueTask<ProducerState> OnStateChangeFrom(ProducerState state, CancellationToken cancellationToken = default) + => await _state.StateChangedFrom(state, cancellationToken).ConfigureAwait(false); + + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _isDisposed, 1) != 0) + return; + + _cts.Cancel(); + _cts.Dispose(); + + foreach (var producer in _producers.Values) + { + await producer.DisposeAsync().ConfigureAwait(false); + } + + _eventRegister.Register(new ProducerDisposed(_correlationId)); + } + + public async ValueTask<MessageId> Send(TMessage message, CancellationToken cancellationToken = default) + => await _producers[_messageRouter.ChoosePartition(null, _producersCount)].Send(message, cancellationToken); + + public async ValueTask<MessageId> Send(MessageMetadata metadata, TMessage message, CancellationToken cancellationToken = default) + => await _producers[_messageRouter.ChoosePartition(metadata, _producersCount)].Send(message, cancellationToken); Review comment: Missing ConfigureAwait(false); ########## File path: src/DotPulsar/Internal/PartitionedProducer.cs ########## @@ -0,0 +1,96 @@ +namespace DotPulsar.Internal +{ + using Abstractions; + using DotPulsar.Abstractions; + using Events; + using System; + using System.Collections.Concurrent; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + + public sealed class PartitionedProducer<TMessage> : IProducer<TMessage> + { + private readonly Guid _correlationId; + private readonly IRegisterEvent _eventRegister; + private readonly IStateChanged<ProducerState> _state; + private readonly PulsarClient _pulsarClient; + private readonly ProducerOptions<TMessage> _options; + private readonly ConcurrentDictionary<int, IProducer<TMessage>> _producers; + private readonly IMessageRouter _messageRouter; + private readonly CancellationTokenSource _cts = new(); + private readonly int _producersCount; + private int _isDisposed; + public Uri ServiceUrl { get; } + public string Topic { get; } + + public PartitionedProducer( + Guid correlationId, + Uri serviceUrl, + string topic, + IRegisterEvent registerEvent, + IStateChanged<ProducerState> state, + uint partitionsCount, + ProducerOptions<TMessage> options, + PulsarClient pulsarClient + ) + { + _correlationId = correlationId; + ServiceUrl = serviceUrl; + Topic = topic; + _eventRegister = registerEvent; + _state = state; + _isDisposed = 0; + _options = options; + _pulsarClient = pulsarClient; + _producersCount = (int) partitionsCount; + _messageRouter = options.MessageRouter; + + _producers = new ConcurrentDictionary<int, IProducer<TMessage>>(Environment.ProcessorCount, _producersCount); Review comment: Multiple tasks will be reading the dictionary, but only one will update it. We can therefore use another constructor since 'concurrencyLevel' is only 1 and not {Environment.ProcessorCount}. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
