feat: add disconnect() method to release resources deterministically

src/Contract/PheanstalkClientInterface.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Pheanstalk\Contract;
+
+interface PheanstalkClientInterface
+{
+    /**
+     * Closes the current connection and releases all associated resources.
+     *
+     * This method ensures that after the call, no resources are held by the client,
+     * allowing garbage collection to safely reclaim memory. If no connection is open,
+     * this method does nothing (noop). Future command dispatches will automatically
+     * establish a new connection if needed.
+     *
+     * Note: If the client has any reserved jobs at the time of disconnection, beanstalkd
+     * will automatically release them back into the ready queue.
+     */
+    public function disconnect(): void;
+}

src/Contract/PheanstalkManagerInterface.php

@@ -11,7 +11,7 @@
 use Pheanstalk\Values\TubeName;
 use Pheanstalk\Values\TubeStats;
 
-interface PheanstalkManagerInterface
+interface PheanstalkManagerInterface extends PheanstalkClientInterface
 {
     /**
      * Kicks buried or delayed jobs into a 'ready' state.

src/Contract/PheanstalkPublisherInterface.php

@@ -6,7 +6,7 @@
 
 use Pheanstalk\Values\TubeName;
 
-interface PheanstalkPublisherInterface
+interface PheanstalkPublisherInterface extends PheanstalkClientInterface
 {
     public const int DEFAULT_DELAY = 0; // no delay
     public const int DEFAULT_PRIORITY = 1024; // most urgent: 0, least urgent: 4294967295

src/Contract/PheanstalkSubscriberInterface.php

@@ -8,7 +8,7 @@
 use Pheanstalk\Values\TubeList;
 use Pheanstalk\Values\TubeName;
 
-interface PheanstalkSubscriberInterface
+interface PheanstalkSubscriberInterface extends PheanstalkClientInterface
 {
     /**
      * Permanently deletes a job.

src/Pheanstalk.php

@@ -118,6 +118,11 @@ public function stats(): ServerStats
         return $this->manager->stats();
     }
 
+    public function disconnect(): void
+    {
+        $this->manager->disconnect();
+    }
+
     public function bury(JobIdInterface $job, int $priority = self::DEFAULT_PRIORITY): void
     {
         $this->subscriber->bury($job, $priority);

src/PheanstalkManager.php

@@ -114,4 +114,9 @@ public function stats(): ServerStats
         $command = new Command\StatsCommand();
         return $command->interpret($this->dispatch($command));
     }
+
+    public function disconnect(): void
+    {
+        $this->connection->disconnect();
+    }
 }

src/PheanstalkPublisher.php

@@ -40,4 +40,9 @@ public function put(
         $command = new PutCommand($data, $priority, $delay, $timeToRelease);
         return $command->interpret($this->connection->dispatchCommand($command));
     }
+
+    public function disconnect(): void
+    {
+        $this->connection->disconnect();
+    }
 }

src/PheanstalkSubscriber.php

@@ -102,4 +102,9 @@ public function reserveJob(JobIdInterface $job): Job
         $command = new ReserveJobCommand($job);
         return $command->interpret($this->dispatch($command));
     }
+
+    public function disconnect(): void
+    {
+        $this->connection->disconnect();
+    }
 }

tests/Unit/ConnectionTest.php

@@ -29,15 +29,34 @@
     public function testDisconnect(): void
     {
         $socket = $this->getMockBuilder(SocketInterface::class)->getMock();
         $socket->expects(self::once())->method('disconnect');
         $factory = $this->getMockBuilder(SocketFactoryInterface::class)->getMock();
         $factory->expects(self::atLeastOnce())->method('create')->willReturn($socket);
 
         $connection = new Connection($factory);
         $connection->connect();
         $connection->disconnect();
     }
 
+    public function testReconnectAfterDispatch(): void
+    {
+        $socket = $this->getMockBuilder(SocketInterface::class)->getMock();
+        $socket->expects(self::once())->method('disconnect');
+        $socket->expects(self::once())
+            ->method('getLine')
+            ->willReturn(ResponseType::Using->value);
+
+        $factory = $this->getMockBuilder(SocketFactoryInterface::class)->getMock();
+        $factory->expects(self::exactly(2))
+            ->method('create')
+            ->willReturn($socket);
+
+        $connection = new Connection($factory);
+        $connection->connect();
+        $connection->disconnect();
+        $connection->dispatchCommand(new UseCommand(new TubeName('foo')));
+    }
+
     private function getCommand(): CommandInterface
     {
         return new UseCommand(new TubeName('tube5'));
@@ -51,11 +70,11 @@
         $socket = $this->getMockBuilder(SocketInterface::class)
             ->getMock();
 
         $socket->expects(self::once())
             ->method('getLine')
             ->willReturn($line);
 
         $socket->expects(self::once())->method('write');
 
         return $this->createConnectionBySocket($socket);
     }
@@ -102,7 +121,7 @@
     {
         $socket = $this->getMockBuilder(SocketInterface::class)
             ->getMock();
         $socket->expects(self::exactly(2))
             ->method('write')
             ->willReturnCallback(static function () {
                 static $firstRun = true;
@@ -112,7 +131,7 @@
                 }
             });
 
         $socket->expects(self::once())->method('disconnect');
 
         $connection = $this->createConnectionBySocket($socket);
 
@@ -121,7 +140,7 @@
             self::fail('Expected ConnectionException was not thrown');
         } catch (ConnectionException) {
         }
         $socket->expects(self::exactly(1))
             ->method('getLine')
             ->willReturn(ResponseType::Released->value);
         $connection->dispatchCommand($this->getCommand());