mnadel · May 25, 2020 17:47 · Nov 8, 2015 · Nov 7, 2015 · Nov 7, 2015 · Nov 7, 2015
diff --git a/ThrottledTplDataflowPipelne.cs b/ThrottledTplDataflowPipelne.cs
@@ -22,9 +22,7 @@ class Pipeline<T>
             }
         });
 
-        _buffer.LinkTo(actionBlock, new DataflowLinkOptions {
-            PropagateCompletion = true
-        });
+        _buffer.LinkTo(actionBlock);
     }
 
     void Post(T message)

diff --git a/ThrottledTplDataflowPipelne.cs b/ThrottledTplDataflowPipelne.cs
@@ -22,7 +22,7 @@ class Pipeline<T>
             }
         });
 
-        _buffer.LinkTo (actionBlock, new DataflowLinkOptions {
+        _buffer.LinkTo(actionBlock, new DataflowLinkOptions {
             PropagateCompletion = true
         });
     }

diff --git a/ThrottledTplDataflowPipelne.cs b/ThrottledTplDataflowPipelne.cs
@@ -0,0 +1,35 @@
+using System.Threading;
+using System.Threading.Tasks.Dataflow;
+
+class Pipeline<T>
+{
+    private readonly SemaphoreSlim _semaphore;
+    private readonly BufferBlock<T> _buffer;
+
+    Pipeline(Action<T> action, int capacity)
+    {
+        _semaphore = new SemaphoreSlim(capacity, capacity);
+        _buffer = new BufferBlock<T>();
+
+        var actionBlock = new ActionBlock<T>(t => {
+            try
+            {
+                action(t);
+            }
+            finally
+            {
+                _semaphore.Release();
+            }
+        });
+
+        _buffer.LinkTo (actionBlock, new DataflowLinkOptions {
+            PropagateCompletion = true
+        });
+    }
+
+    void Post(T message)
+    {
+        _semaphore.Wait();
+        _buffer.Post(message);
+    }
+}
No results found