stream support queue

reactphp-x · Jan 13, 2025 · a9f2da5 · a9f2da5
1 parent 2828fa3
commit a9f2da5
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/src/Bandwidth.php b/src/Bandwidth.php
@@ -66,13 +66,17 @@ public function file(string $path, $p = 0, $length = -1)
     public function stream($stream)
     {
         $_stream = new \React\Stream\ThroughStream();
-        $stream->on('data', function ($data) use ($_stream) {
-            $this->concurrent->concurrent(function() use ($_stream, $data){
+
+        $concurrent = $this->queue ? $this->concurrent : new Concurrent(1);
+
+        $stream->on('data', function ($data) use ($_stream, $concurrent) {
+            $concurrent->concurrent(function() use ($_stream, $data){
                 return $this->bucket->removeTokens(1024 * strlen($data))->then(function () use ($_stream, $data) {
                     $_stream->write($data);
                 });
             });
         });
+
         return $_stream;
     }