change partition id to a constructor parameter

dotnetcore · Jan 29, 2024 · 717581d · 717581d
1 parent e502c7f
commit 717581d
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 11 deletions.
diff --git a/src/Mocha.Core/Buffer/Memory/MemoryBuferQueue.cs b/src/Mocha.Core/Buffer/Memory/MemoryBuferQueue.cs
@@ -22,7 +22,7 @@ public MemoryBufferQueue(string topicName, int partitionNumber)
         _partitions = new MemoryBufferPartition<T>[partitionNumber];
         for (var i = 0; i < partitionNumber; i++)
         {
-            _partitions[i] = new MemoryBufferPartition<T>();
+            _partitions[i] = new MemoryBufferPartition<T>(i);
         }
 
         _producer = new MemoryBufferProducer<T>(topicName, _partitions);

diff --git a/src/Mocha.Core/Buffer/Memory/MemoryBufferPartition.cs b/src/Mocha.Core/Buffer/Memory/MemoryBufferPartition.cs
@@ -14,8 +14,6 @@ internal sealed class MemoryBufferPartition<T>
     // internal for test
     internal readonly int _segmentLength;
 
-    private static int _idIncreacement;
-
     private volatile MemoryBufferSegment<T> _head;
     private volatile MemoryBufferSegment<T> _tail;
 
@@ -25,10 +23,10 @@ internal sealed class MemoryBufferPartition<T>
 
     private readonly object _createSegmentLock;
 
-    public MemoryBufferPartition(int segmentLength = 1024)
+    public MemoryBufferPartition(int id, int segmentLength = 1024)
     {
         _segmentLength = segmentLength;
-        PartitionId = _idIncreacement++;
+        PartitionId = id;
         _head = _tail = new MemoryBufferSegment<T>(_segmentLength, default);
         _consumerReaders = new ConcurrentDictionary<string, Reader>();
         _consumers = new HashSet<MemoryBufferConsumer<T>>();

diff --git a/tests/Mocha.Core.Tests/Buffer/Memory/MemoryBufferPartitionTests.cs b/tests/Mocha.Core.Tests/Buffer/Memory/MemoryBufferPartitionTests.cs
@@ -11,7 +11,7 @@ public class MemoryBufferPartitionTests
     [Fact]
     public void Enqueue_And_TryPull()
     {
-        var partition = new MemoryBufferPartition<int>(2);
+        var partition = new MemoryBufferPartition<int>(0, 2);
 
         for (var i = 0; i < 12; i++)
         {
@@ -45,7 +45,7 @@ public void Enqueue_And_TryPull()
     [Fact]
     public void Repeatable_Pull_If_Not_Commit()
     {
-        var partition = new MemoryBufferPartition<int>(2);
+        var partition = new MemoryBufferPartition<int>(0, 2);
 
         for (var i = 0; i < 11; i++)
         {
@@ -82,7 +82,7 @@ public void Repeatable_Pull_If_Not_Commit()
     [Fact]
     public void Segment_Will_Be_Recycled_If_All_Consumers_Consumed_Single_Group()
     {
-        var partition = new MemoryBufferPartition<int>(3);
+        var partition = new MemoryBufferPartition<int>(0, 3);
 
         for (var i = 0; i < 9; i++)
         {
@@ -116,7 +116,7 @@ public void Segment_Will_Be_Recycled_If_All_Consumers_Consumed_Single_Group()
     [Fact]
     public void Segment_Will_Be_Recycled_If_All_Consumers_Consumed_MultipleGroup()
     {
-        var partition = new MemoryBufferPartition<int>(3);
+        var partition = new MemoryBufferPartition<int>(0, 3);
 
         for (var i = 0; i < 9; i++)
         {
@@ -148,7 +148,7 @@ public void Segment_Will_Be_Recycled_If_All_Consumers_Consumed_MultipleGroup()
     [Fact]
     public void Segment_Will_Not_Be_Recycled_If_Not_All_Consumers_Consumed_MultipleGroup()
     {
-        var partition = new MemoryBufferPartition<int>(3);
+        var partition = new MemoryBufferPartition<int>(0, 3);
 
         for (var i = 0; i < 6; i++)
         {

diff --git a/tests/Mocha.Core.Tests/Buffer/Memory/MemoryBufferQueueTests.cs b/tests/Mocha.Core.Tests/Buffer/Memory/MemoryBufferQueueTests.cs
@@ -9,7 +9,7 @@ namespace Mocha.Core.Tests.Buffer.Memory;
 
 public class MemoryBufferQueueTests
 {
-    private static int MemoryBufferPartitionSegmentLength => new MemoryBufferPartition<int>()._segmentLength;
+    private static int MemoryBufferPartitionSegmentLength => new MemoryBufferPartition<int>(0)._segmentLength;
 
     [Fact]
     public async Task Produce_And_Consume()