Update: delete stream transaction

Author: cclilshy

README.md

@@ -71,5 +71,3 @@ Provides standard coroutine architecture and tools for rapid development or pack
 `Email` jingnigg@gmail.com
 
 `WeChat` jingnigg
-
----

README.zh_cn.md

@@ -78,5 +78,3 @@ workbunny团队体统的webman协程扩展, 为Webman提供了协程支持
 `电邮` jingnigg@gmail.com
 
 `微信` jingnigg
-
----

example/suspension.php

@@ -0,0 +1,18 @@
+<?php declare(strict_types=1);
+
+include __DIR__ . '/../vendor/autoload.php';
+
+\Co\async(static function () {
+    $suspension = \Co\getSuspension();
+
+    \Co\async(function () use ($suspension) {
+        \Co\sleep(1);
+        $suspension->resume();
+    });
+
+    $suspension->suspend();
+
+    echo 'Coroutine 1', \PHP_EOL;
+});
+
+\Co\wait();

src/Coroutine/Coroutine.php

@@ -151,7 +151,7 @@ public function async(Closure $closure): Promise
         return promise(function (Closure $resolve, Closure $reject, Promise $promise) use ($closure) {
             $suspension = new Suspension(function () use ($closure, $resolve, $reject) {
                 try {
-                    $resolve($closure($resolve, $reject));
+                    $resolve($closure());
                 } catch (EscapeException $exception) {
                     throw $exception;
                 } catch (Throwable $exception) {

src/Stream.php

@@ -13,15 +13,12 @@
 namespace Ripple;
 
 use Closure;
-use Exception;
 use Revolt\EventLoop;
 use Ripple\Coroutine\Coroutine;
 use Ripple\Coroutine\Suspension;
 use Ripple\Stream\Exception\ConnectionCloseException;
-use Ripple\Stream\Exception\ConnectionException;
 use Ripple\Stream\Exception\ConnectionTimeoutException;
 use Ripple\Stream\Stream as StreamBase;
-use Ripple\Stream\Transaction;
 use Ripple\Utils\Format;
 use Ripple\Utils\Output;
 use Throwable;
@@ -79,22 +76,12 @@ class Stream extends StreamBase
      */
     protected int $index = 0;
 
-    /**
-     * @var Transaction
-     */
-    protected Transaction $transaction;
-
     /**
      * @param mixed $resource
      */
     public function __construct(mixed $resource)
     {
         parent::__construct($resource);
-
-        $this->onClose(function () {
-            $this->cancelReadable();
-            $this->cancelWriteable();
-        });
     }
 
     /**
@@ -140,57 +127,6 @@ public function setBlocking(bool $bool): bool
         return stream_set_blocking($this->stream, $bool);
     }
 
-    /**
-     * @param Closure $closure
-     *
-     * @return void
-     * @throws Throwable
-     */
-    public function transaction(Closure $closure): void
-    {
-        if (isset($this->transaction) && $this->transaction->getPromise()->getStatus() === Promise::PENDING) {
-            throw new Exception('Transaction has been completed');
-        }
-
-        $this->setTransaction(new Transaction($this));
-        call_user_func_array($closure, [$this->getTransaction()]);
-    }
-
-    /**
-     * @return Transaction|null
-     */
-    public function getTransaction(): Transaction|null
-    {
-        if (isset($this->transaction)) {
-            return $this->transaction;
-        }
-        return null;
-    }
-
-    /**
-     * @param Transaction $transaction
-     *
-     * @return void
-     */
-    protected function setTransaction(Transaction $transaction): void
-    {
-        if (isset($this->transaction)) {
-            $this->completeTransaction();
-            unset($this->transaction);
-        }
-        $this->transaction = $transaction;
-    }
-
-    /**
-     * @return void
-     */
-    public function completeTransaction(): void
-    {
-        if (isset($this->transaction)) {
-            $this->transaction->complete();
-        }
-    }
-
     /**
      * Wait for readable events. This method is only valid when there are no readable events to listen for.
      * After enabling this method, it is forbidden to use the onReadable method elsewhere unless you know what you are doing.
@@ -208,7 +144,7 @@ public function waitForReadable(int $timeout = 0): bool
     {
         $suspension = getSuspension();
         if (!isset($this->onReadable)) {
-            $this->onReadable(fn () => Coroutine::resume($suspension, true));
+            $this->onReadable(static fn () => Coroutine::resume($suspension, true));
             $suspension instanceof Suspension && $suspension->promise->finally(fn () => $this->cancelReadable());
         }
 
@@ -268,15 +204,6 @@ public function close(): void
         $this->cancelReadable();
         $this->cancelWriteable();
 
-        if (isset($this->transaction)) {
-            $this->failTransaction(new ConnectionException(
-                'Stream has been closed',
-                ConnectionException::CONNECTION_CLOSED,
-                null,
-                $this
-            ));
-        }
-
         foreach ($this->onCloseCallbacks as $callback) {
             try {
                 call_user_func($callback);
@@ -294,18 +221,6 @@ public function isClosed(): bool
         return !is_resource($this->stream);
     }
 
-    /**
-     * @param Throwable $exception
-     *
-     * @return void
-     */
-    public function failTransaction(Throwable $exception): void
-    {
-        if (isset($this->transaction)) {
-            $this->transaction->fail($exception);
-        }
-    }
-
     /**
      * @param string $key
      *
@@ -335,7 +250,7 @@ public function waitForWriteable(int $timeout = 0): bool
     {
         $suspension = getSuspension();
         if (!isset($this->onWriteable)) {
-            $this->onWriteable(fn () => Coroutine::resume($suspension, true));
+            $this->onWriteable(static fn () => Coroutine::resume($suspension, true));
             $suspension instanceof Suspension && $suspension->promise->finally(fn () => $this->cancelWriteable());
         }