Fix

Sharp-Pulsar · Jan 31, 2024 · 687ad12 · 687ad12
1 parent 36e4d98
commit 687ad12
Show file tree

Hide file tree

Showing 5 changed files with 17 additions and 8 deletions.
diff --git a/src/SharpPulsar/Client/Internal/SocketClientActor.cs b/src/SharpPulsar/Client/Internal/SocketClientActor.cs
@@ -185,10 +185,10 @@ public SocketClientActor(IActorRef client, ClientConfigurationData conf, DnsEndP
             {
                 try
                 {
-                    while (_start)
-                    {
-                        _logger.Info("Running on thread: " + Thread.CurrentThread.ManagedThreadId);
+                    _logger.Info("Running on thread: " + Thread.CurrentThread.ManagedThreadId);
 
+                    while (_start)
+                    {                        
                         var result = await _pipeReader.ReadAsync().ConfigureAwait(false);
 
                         var buffer = result.Buffer;

diff --git a/src/SharpPulsar/Consumer/ConsumerActor.cs b/src/SharpPulsar/Consumer/ConsumerActor.cs
@@ -949,6 +949,8 @@ private void Ready()
                     Sender.Tell(new AskResponse(Unwrap(ex)));
                 }
             });
+            Receive<bool>(c => { });
+            Receive<string>(s => { });
         }
 
         private async ValueTask Acknowledge(IAcknowledge ack)

diff --git a/src/SharpPulsar/Consumer/MultiTopicsConsumer.cs b/src/SharpPulsar/Consumer/MultiTopicsConsumer.cs
@@ -550,6 +550,8 @@ internal void Ready()
                     Sender.Tell(new AskResponse(Unwrap(ex)));
                 }
             });
+            Receive<bool>(c => { });
+            Receive<string>(s => { });
         }
 
         // Check topics are valid.

diff --git a/src/SharpPulsar/Extension/ArrayDeque.cs b/src/SharpPulsar/Extension/ArrayDeque.cs
@@ -342,7 +342,8 @@ public T RemoveFirst()
         {
 
             if (Count == 0)
-                throw new InvalidOperationException("There are no items to remove.");
+                //throw new InvalidOperationException("There are no items to remove.");
+                return default(T);  
 
             _size--;
 
@@ -358,7 +359,8 @@ public T RemoveLast()
         {
 
             if (Count == 0)
-                throw new InvalidOperationException("There are no items to remove.");
+                //throw new InvalidOperationException("There are no items to remove.");
+                return default(T);
 
             _size--;
 

diff --git a/src/SharpPulsar/Tracker/UnAckedMessageTracker.cs b/src/SharpPulsar/Tracker/UnAckedMessageTracker.cs
@@ -69,8 +69,8 @@ public UnAckedMessageTracker(IActorRef consumer, IActorRef unack, ConsumerConfig
             }
             else
             {
-                MessageIdPartitionMap = null;
-                TimePartitions = null;
+                MessageIdPartitionMap = new();
+                TimePartitions = new();
             }
             BecomeReady();
 
@@ -117,14 +117,16 @@ private void BecomeReady()
             {
                 RedeliverMessages();
             });
+            Receive<bool>(c => { });
+            Receive<string>(s => { });
         }
         internal virtual void RedeliverMessages()
         {
             MessageIds.Clear();
             try
             {
                 var headPartition = TimePartitions.RemoveFirst();
-                if (headPartition.Count > 0)
+                if (headPartition?.Count > 0)
                 {
                     _log.Warning($"[{Consumer.Path.Name}] {headPartition.Count} messages will be re-delivered");
                     headPartition.ForEach(async messageId =>
@@ -171,6 +173,7 @@ internal async ValueTask AddChunkedMessageIdsAndRemoveFromSequenceMap(IMessageId
                 Unack.Tell(new UnAckedChunckedMessageIdSequenceMapCmd(UnAckedCommand.Remove, new List<IMessageId> { messageId }));
             }
         }
+
         internal virtual void Clear()
         {
             try