RobertIndie commented on a change in pull request #71:
URL: https://github.com/apache/pulsar-dotpulsar/pull/71#discussion_r655154397
##########
File path: src/DotPulsar/Internal/ProducerProcess.cs
##########
@@ -15,52 +15,141 @@
namespace DotPulsar.Internal
{
using Abstractions;
+ using Events;
using System;
+ using System.Threading;
using System.Threading.Tasks;
public sealed class ProducerProcess : Process
{
private readonly IStateManager<ProducerState> _stateManager;
- private readonly IEstablishNewChannel _producer;
+ private readonly IEstablishNewChannel? _producer;
+
+ // The following variables are only used when this is the process for
parent producer.
+ private readonly IRegisterEvent _processManager;
+ private int _partitionsCount;
+ private int _connectedProducersCount;
+ private int _initialProducersCount;
+
+ // The following variables are only used for sub producer
+ private readonly Guid? _partitionedProducerId;
public ProducerProcess(
Guid correlationId,
IStateManager<ProducerState> stateManager,
- IEstablishNewChannel producer) : base(correlationId)
+ IEstablishNewChannel? producer,
+ IRegisterEvent processManager,
+ Guid? partitionedProducerId = null) : base(correlationId)
{
_stateManager = stateManager;
_producer = producer;
+ _processManager = processManager;
+ _partitionedProducerId = partitionedProducerId;
}
public override async ValueTask DisposeAsync()
{
- _stateManager.SetState(ProducerState.Closed);
+ SetState(ProducerState.Closed);
CancellationTokenSource.Cancel();
- await _producer.DisposeAsync().ConfigureAwait(false);
+
+ if (_producer != null)
+ await _producer.DisposeAsync().ConfigureAwait(false);
+ }
+
+ protected override void HandleExtend(IEvent e)
Review comment:
What confuses me is the UpdatePartitions part. I think Producer class
need to be responsible for sub producers creation. The producer process does
not create or delete sub producers. If we want to implement in this way, we may
need to share the collection of sub producers between Producer and
ParoducerProcess. But it seems more complex.
--
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]
