[1.0.0] Consolidate server queue writing to one spot (Part 1)

src/FreeDSx/Ldap/Container/HandlerContainerProvider.php

@@ -71,43 +71,39 @@ private function makeFactoryMap(Container $container): ProtocolHandlerFactoryMap
         return new ProtocolHandlerFactoryMap([
             HandlerId::Abandon->value => static fn(): ServerProtocolHandlerInterface
                 => new ServerAbandonHandler(),
-            HandlerId::Cancel->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
-                => new ServerCancelHandler($context->queue),
-            HandlerId::WhoAmI->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
-                => new ServerWhoAmIHandler($context->queue),
+            HandlerId::Cancel->value => static fn(): ServerProtocolHandlerInterface
+                => new ServerCancelHandler(),
+            HandlerId::WhoAmI->value => static fn(): ServerProtocolHandlerInterface
+                => new ServerWhoAmIHandler(),
             HandlerId::PasswordModify->value => fn(HandlerContext $context): ServerProtocolHandlerInterface
                 => $this->makePasswordModifyHandler($container, $context),
-            HandlerId::PasswordPolicyForward->value => fn(HandlerContext $context): ServerProtocolHandlerInterface
-                => $this->makeForwardHandler($container, $context),
+            HandlerId::PasswordPolicyForward->value => fn(): ServerProtocolHandlerInterface
+                => $this->makeForwardHandler($container),
             HandlerId::StartTls->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
                 => new ServerStartTlsHandler(
                     options: $container->get(ServerOptions::class),
-                    queue: $context->queue,
+                    connection: $context->queue,
                     eventLogger: $context->eventLogger,
                 ),
-            HandlerId::UnsupportedExtended->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
-                => new ServerUnsupportedExtendedHandler($context->queue),
-            HandlerId::RootDse->value => fn(HandlerContext $context): ServerProtocolHandlerInterface
-                => $this->makeRootDseHandler($container, $context),
-            HandlerId::Subschema->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
-                => new ServerSubschemaHandler(
-                    options: $container->get(ServerOptions::class),
-                    queue: $context->queue,
-                ),
-            HandlerId::Monitor->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
+            HandlerId::UnsupportedExtended->value => static fn(): ServerProtocolHandlerInterface
+                => new ServerUnsupportedExtendedHandler(),
+            HandlerId::RootDse->value => fn(): ServerProtocolHandlerInterface
+                => $this->makeRootDseHandler($container),
+            HandlerId::Subschema->value => static fn(): ServerProtocolHandlerInterface
+                => new ServerSubschemaHandler($container->get(ServerOptions::class)),
+            HandlerId::Monitor->value => static fn(): ServerProtocolHandlerInterface
                 => new ServerMonitorHandler(
                     options: $container->get(ServerOptions::class),
-                    queue: $context->queue,
                     snapshots: $container->get(MetricsSnapshotProvider::class),
                 ),
             HandlerId::Paging->value => fn(HandlerContext $context, ?SearchLimits $limits): ServerProtocolHandlerInterface
                 => $this->makePagingHandler($container, $context, $limits),
             HandlerId::Sync->value => fn(HandlerContext $context, ?SearchLimits $limits): ServerProtocolHandlerInterface
                 => $this->makeSyncHandler($container, $context, $limits),
             HandlerId::Search->value => fn(HandlerContext $context, ?SearchLimits $limits): ServerProtocolHandlerInterface
-                => $this->makeSearchHandler($container, $context, $limits),
-            HandlerId::Unbind->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
-                => new ServerUnbindHandler($context->queue),
+                => $this->makeSearchHandler($container, $limits),
+            HandlerId::Unbind->value => static fn(): ServerProtocolHandlerInterface
+                => new ServerUnbindHandler(),
             HandlerId::Dispatch->value => fn(HandlerContext $context): ServerProtocolHandlerInterface
                 => $this->makeDispatchHandler($container, $context),
         ]);
@@ -120,7 +116,6 @@ private function makePasswordModifyHandler(
         $policyComponentFactory = $container->get(PasswordPolicyComponentFactory::class);
 
         return new ServerPasswordModifyHandler(
-            queue: $context->queue,
             service: new PasswordModifyService(
                 targetResolver: $container->get(PasswordModifyTargetResolver::class),
                 accessControl: $container->get(ServerOptions::class)->getAccessControl(),
@@ -135,33 +130,27 @@ private function makePasswordModifyHandler(
         );
     }
 
-    private function makeForwardHandler(
-        Container $container,
-        HandlerContext $context,
-    ): ServerProtocolHandlerInterface {
+    private function makeForwardHandler(Container $container): ServerProtocolHandlerInterface
+    {
         $backend = $container->get(HandlerFactoryInterface::class)->makeBackend();
         if (!$backend instanceof WritableLdapBackendInterface) {
-            return new ServerUnsupportedExtendedHandler($context->queue);
+            return new ServerUnsupportedExtendedHandler();
         }
 
         return new ServerPasswordPolicyForwardHandler(
-            queue: $context->queue,
             backend: $backend,
             policyResolver: $container->get(PasswordPolicyComponentFactory::class)->makeResolver(),
             engine: $container->get(PasswordPolicyEngine::class),
         );
     }
 
-    private function makeRootDseHandler(
-        Container $container,
-        HandlerContext $context,
-    ): ServerRootDseHandler {
+    private function makeRootDseHandler(Container $container): ServerRootDseHandler
+    {
         $handlerFactory = $container->get(HandlerFactoryInterface::class);
         $backend = $handlerFactory->makeBackend();
 
         return new ServerRootDseHandler(
             options: $container->get(ServerOptions::class),
-            queue: $context->queue,
             backend: $backend,
             rootDseHandler: $handlerFactory->makeRootDseHandler(),
             supportsSync: $this->syncJournalFor($container, $backend) !== null,
@@ -223,7 +212,6 @@ private function makeDispatchHandler(
         );
 
         return new ServerDispatchHandler(
-            queue: $context->queue,
             backend: $handlerFactory->makeBackend(),
             writeDispatcher: $policyWriteHandler !== null
                 ? $handlerFactory->makeWriteDispatcher($policyWriteHandler)
@@ -235,13 +223,11 @@ private function makeDispatchHandler(
 
     private function makeSearchHandler(
         Container $container,
-        HandlerContext $context,
         ?SearchLimits $searchLimits,
     ): ServerSearchHandler {
         $options = $container->get(ServerOptions::class);
 
         return new ServerSearchHandler(
-            queue: $context->queue,
             backend: $container->get(HandlerFactoryInterface::class)->makeBackend(),
             filterEvaluator: $options->getFilterEvaluator(),
             accessControl: $options->getAccessControl(),
@@ -258,7 +244,6 @@ private function makePagingHandler(
         $options = $container->get(ServerOptions::class);
 
         return new ServerPagingHandler(
-            queue: $context->queue,
             backend: $container->get(HandlerFactoryInterface::class)->makeBackend(),
             filterEvaluator: $options->getFilterEvaluator(),
             accessControl: $options->getAccessControl(),

src/FreeDSx/Ldap/Protocol/Queue/ConnectionControl.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * This file is part of the FreeDSx LDAP package.
+ *
+ * (c) Chad Sikorra <Chad.Sikorra@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace FreeDSx\Ldap\Protocol\Queue;
+
+/**
+ * A narrow connection-lifecycle seam handlers use for post-write socket actions without the queue.
+ *
+ * @internal
+ * @author Chad Sikorra <Chad.Sikorra@gmail.com>
+ */
+interface ConnectionControl
+{
+    public function isEncrypted(): bool;
+
+    public function encrypt(): static;
+
+    public function close(): void;
+}

src/FreeDSx/Ldap/Protocol/Queue/Response/Cancellation.php

@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * This file is part of the FreeDSx LDAP package.
+ *
+ * (c) Chad Sikorra <Chad.Sikorra@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace FreeDSx\Ldap\Protocol\Queue\Response;
+
+use FreeDSx\Ldap\Operation\Request\AbandonRequest;
+use FreeDSx\Ldap\Operation\Request\CancelRequest;
+use FreeDSx\Ldap\Protocol\LdapMessageRequest;
+
+/**
+ * The control channel between the response writer (which polls the queue) and a streaming producer
+ * (which reads the offered abandon/cancel signal after each yield).
+ *
+ * @internal
+ * @author Chad Sikorra <Chad.Sikorra@gmail.com>
+ */
+final class Cancellation
+{
+    private ?LdapMessageRequest $signal = null;
+
+    /**
+     * The writer offers each poll result; nulls are ignored so a real signal is never overwritten.
+     */
+    public function offer(?LdapMessageRequest $signal): void
+    {
+        if ($signal !== null) {
+            $this->signal = $signal;
+        }
+    }
+
+    public function isSignalled(): bool
+    {
+        return $this->signal !== null;
+    }
+
+    public function isAbandoned(): bool
+    {
+        return $this->signal?->getRequest() instanceof AbandonRequest;
+    }
+
+    public function isCanceled(): bool
+    {
+        return $this->signal?->getRequest() instanceof CancelRequest;
+    }
+
+    public function signal(): ?LdapMessageRequest
+    {
+        return $this->signal;
+    }
+}

src/FreeDSx/Ldap/Protocol/Queue/Response/QueueWriterConfig.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * This file is part of the FreeDSx LDAP package.
+ *
+ * (c) Chad Sikorra <Chad.Sikorra@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace FreeDSx\Ldap\Protocol\Queue\Response;
+
+/**
+ * Tells the response writer how to drive a stream; defaults to the fully-batched, unpolled fast path.
+ *
+ * @internal
+ * @author Chad Sikorra <Chad.Sikorra@gmail.com>
+ */
+final readonly class QueueWriterConfig
+{
+    /**
+     * @param int $signalInterval poll for a cancel/abandon signal every N messages; 0 disables it.
+     * @param bool $flushPerMessage flush the socket after each message instead of the default buffering.
+     */
+    public function __construct(
+        public int $signalInterval = 0,
+        public bool $flushPerMessage = false,
+    ) {}
+
+    public function mustFlushOrSignal(): bool
+    {
+        return $this->signalInterval > 0 || $this->flushPerMessage;
+    }
+}