feat(sentry): add AsyncHttpTransport for non-blocking error reporting

src/sentry/publish/sentry.php

@@ -128,7 +128,8 @@
         'timezone' => env('SENTRY_CRONS_TIMEZONE', date_default_timezone_get()),
     ],
 
+    'transport_channel_size' => (int) env('SENTRY_TRANSPORT_CHANNEL_SIZE', 65535),
+    'transport_concurrent_limit' => (int) env('SENTRY_TRANSPORT_CONCURRENT_LIMIT', 1000),
+
     'http_timeout' => (float) env('SENTRY_HTTP_TIMEOUT', 2.0),
-    'http_chanel_size' => (int) env('SENTRY_HTTP_CHANEL_SIZE', 65535),
-    'http_concurrent_limit' => (int) env('SENTRY_HTTP_CONCURRENT_LIMIT', 100),
 ];

src/sentry/src/ConfigProvider.php

@@ -47,7 +47,8 @@ public function __invoke(): array
             'dependencies' => [
                 \Sentry\ClientBuilder::class => Factory\ClientBuilderFactory::class,
                 \Sentry\State\HubInterface::class => Factory\HubFactory::class,
-                \Sentry\HttpClient\HttpClientInterface::class => HttpClient\HttpClientFactory::class,
+                // \Sentry\HttpClient\HttpClientInterface::class => HttpClient\HttpClientFactory::class,
+                \Sentry\Transport\TransportInterface::class => Transport\CoHttpTransport::class,
             ],
             'listeners' => [
                 Listener\AmqpExceptionListener::class,

src/sentry/src/Factory/ClientBuilderFactory.php

@@ -15,7 +15,7 @@
 use Hyperf\Contract\ConfigInterface;
 use Psr\Container\ContainerInterface;
 use Sentry\ClientBuilder;
-use Sentry\HttpClient\HttpClientInterface;
+use Sentry\Transport\TransportInterface;
 
 use function Hyperf\Support\env;
 use function Hyperf\Tappable\tap;
@@ -28,8 +28,10 @@ class ClientBuilderFactory
         'enable',
         'ignore_commands',
         'integrations',
-        'http_chanel_size',
-        'http_concurrent_limit',
+        'http_chanel_size', // deprecated, will be removed in v3.2
+        'http_concurrent_limit', // deprecated, will be removed in v3.2
+        'transport_channel_size',
+        'transport_concurrent_limit',
         'tracing',
     ];
 
@@ -66,11 +68,14 @@ public function __invoke(ContainerInterface $container)
             $options['environment'] = env('APP_ENV', 'production');
         }
 
+        // If the user has not set a transport we check if one is bound in the container
+        // and use that one. This allows us to use the CoHttpTransport as the default
+        // transport instead of the default Guzzle one.
         if (
-            ! ($options['http_client'] ?? null) instanceof HttpClientInterface
-            && $container->has(HttpClientInterface::class)
+            ! ($options['transport'] ?? null) instanceof TransportInterface
+            && $container->has(TransportInterface::class)
         ) {
-            $options['http_client'] = $container->get(HttpClientInterface::class);
+            $options['transport'] = $container->get(TransportInterface::class);
         }
 
         return tap(

src/sentry/src/HttpClient/HttpClient.php

@@ -24,6 +24,9 @@
 use function Hyperf\Support\msleep;
 use function Hyperf\Tappable\tap;
 
+/**
+ * @deprecated since v3.1, will be removed in v3.2
+ */
 class HttpClient extends \Sentry\HttpClient\HttpClient
 {
     protected ?Channel $chan = null;

src/sentry/src/HttpClient/HttpClientFactory.php

@@ -15,6 +15,9 @@
 use Hyperf\Contract\ConfigInterface;
 use Psr\Container\ContainerInterface;
 
+/**
+ * @deprecated since v3.1, will be removed in v3.2
+ */
 class HttpClientFactory
 {
     public function __invoke(ContainerInterface $container)

src/sentry/src/Transport/CoHttpTransport.php

@@ -0,0 +1,154 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of friendsofhyperf/components.
+ *
+ * @link     https://github.com/friendsofhyperf/components
+ * @document https://github.com/friendsofhyperf/components/blob/main/README.md
+ * @contact  huangdijia@gmail.com
+ */
+
+namespace FriendsOfHyperf\Sentry\Transport;
+
+use Hyperf\Coordinator\Constants;
+use Hyperf\Coordinator\CoordinatorManager;
+use Hyperf\Coroutine\Concurrent;
+use Hyperf\Engine\Channel;
+use Hyperf\Engine\Coroutine;
+use Psr\Container\ContainerInterface;
+use Sentry\ClientBuilder;
+use Sentry\Event;
+use Sentry\Serializer\PayloadSerializer;
+use Sentry\Transport\Result;
+use Sentry\Transport\ResultStatus;
+use Sentry\Transport\TransportInterface;
+use Throwable;
+
+use function Hyperf\Support\msleep;
+
+class CoHttpTransport implements TransportInterface
+{
+    protected ?Channel $chan = null;
+
+    protected bool $workerExited = false;
+
+    protected ?Coroutine $workerWatcher = null;
+
+    protected ?Concurrent $concurrent = null;
+
+    protected ?ClientBuilder $clientBuilder = null;
+
+    protected int $channelSize = 65535;
+
+    public function __construct(
+        protected ContainerInterface $container,
+    ) {
+        $config = $this->container->get('config');
+        $channelSize = (int) $config->get('sentry.transport_channel_size', 65535);
+        if ($channelSize > 0) {
+            $this->channelSize = $channelSize;
+        }
+        $concurrentLimit = (int) $config->get('sentry.transport_concurrent_limit', 1000);
+        if ($concurrentLimit > 0) {
+            $this->concurrent = new Concurrent($concurrentLimit);
+        }
+    }
+
+    public function send(Event $event): Result
+    {
+        $this->loop();
+
+        $chan = $this->chan;
+        $chan?->push($event);
+
+        return new Result(ResultStatus::success(), $event);
+    }
+
+    public function close(?int $timeout = null): Result
+    {
+        return new Result(ResultStatus::success());
+    }
+
+    protected function loop(): void
+    {
+        if ($this->workerExited) {
+            return;
+        }
+
+        if ($this->chan !== null) {
+            return;
+        }
+
+        $this->chan = new Channel($this->channelSize);
+
+        Coroutine::create(function () {
+            while (true) {
+                $transport = $this->makeHttpTransport();
+                $logger = $this->clientBuilder?->getLogger();
+
+                while (true) {
+                    /** @var Event|false|null $event */
+                    $event = $this->chan?->pop();
+
+                    if (! $event) {
+                        break 2;
+                    }
+
+                    try {
+                        $callable = static fn () => $transport->send($event);
+                        if ($this->concurrent !== null) {
+                            $this->concurrent->create($callable);
+                        } else {
+                            Coroutine::create($callable);
+                        }
+                    } catch (Throwable $e) {
+                        $logger?->error('Failed to send event to Sentry: ' . $e->getMessage(), ['exception' => $e]);
+                        $transport->close();
+
+                        break;
+                    } finally {
+                        // Prevent memory leak
+                        $event = null;
+                    }
+                }
+            }
+
+            $this->closeChannel();
+        });
+
+        $this->workerWatcher ??= Coroutine::create(function () {
+            if (CoordinatorManager::until(Constants::WORKER_EXIT)->yield()) {
+                $this->workerExited = true;
+
+                while (! $this->chan?->isEmpty()) {
+                    msleep(100);
+                }
+
+                $this->closeChannel();
+            }
+        });
+    }
+
+    protected function makeHttpTransport(): TransportInterface
+    {
+        $this->clientBuilder ??= $this->container->get(ClientBuilder::class);
+
+        return new \Sentry\Transport\HttpTransport(
+            $this->clientBuilder->getOptions(),
+            $this->clientBuilder->getHttpClient(),
+            new PayloadSerializer($this->clientBuilder->getOptions()),
+            $this->clientBuilder->getLogger()
+        );
+    }
+
+    protected function closeChannel(): void
+    {
+        $chan = $this->chan;
+        $chan?->close();
+
+        if ($this->chan === $chan) {
+            $this->chan = null;
+        }
+    }
+}