Fix stream dead cycle

RayTale · Mar 8, 2021 · 240dcca · 240dcca
1 parent b90f727
commit 240dcca
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 1 deletion.
diff --git a/src/Stream/Vertex.Stream.InMemory/Consumer/ConsumerManager.cs b/src/Stream/Vertex.Stream.InMemory/Consumer/ConsumerManager.cs
@@ -144,6 +144,7 @@ private async Task DistributedStart()
                                         this.consumerRunners.TryAdd(key, runner);
                                         await runner.Run();
                                     }
+                                    break;
                                 }
                                 else if (expectMillisecondDelay > 0)
                                 {

diff --git a/src/Stream/Vertex.Stream.Kafka/Consumer/ConsumerManager.cs b/src/Stream/Vertex.Stream.Kafka/Consumer/ConsumerManager.cs
@@ -144,6 +144,7 @@ private async Task DistributedStart()
                                         this.consumerRunners.TryAdd(key, runner);
                                         await runner.Run();
                                     }
+                                    break;
                                 }
                                 else if (expectMillisecondDelay > 0)
                                 {

diff --git a/src/Stream/Vertex.Stream.RabbitMQ/Consumer/ConsumerManager.cs b/src/Stream/Vertex.Stream.RabbitMQ/Consumer/ConsumerManager.cs
@@ -139,8 +139,8 @@ private async Task DistributedStart()
                                         var runner = new ConsumerRunner(this.provider, queue);
                                         this.consumerRunners.TryAdd(key, runner);
                                         await runner.Run();
-                                        break;
                                     }
+                                    break;
                                 }
                                 else if (expectMillisecondDelay > 0)
                                 {