add tests

baseline.xml

@@ -217,9 +217,6 @@
     <MissingParamType>
       <code><![CDATA[$message]]></code>
     </MissingParamType>
-    <ReservedWord>
-      <code><![CDATA[ProfileVisited&ProfileCreated $event]]></code>
-    </ReservedWord>
   </file>
   <file src="tests/Unit/Schema/DoctrineSchemaSubscriberTest.php">
     <DeprecatedClass>
@@ -332,6 +329,16 @@
     </InternalMethod>
   </file>
   <file src="tests/Unit/Subscription/Engine/DefaultSubscriptionEngineTest.php">
+    <PossiblyNullArgument>
+      <code><![CDATA[$subscriber->throwForBeginBatch]]></code>
+      <code><![CDATA[$subscriber->throwForBeginBatch]]></code>
+      <code><![CDATA[$subscriber->throwForCommitBatch]]></code>
+      <code><![CDATA[$subscriber->throwForCommitBatch]]></code>
+      <code><![CDATA[$subscriber->throwForMessage]]></code>
+      <code><![CDATA[$subscriber->throwForMessage]]></code>
+      <code><![CDATA[$subscriber->throwForMessage]]></code>
+      <code><![CDATA[$subscriber->throwForMessage]]></code>
+    </PossiblyNullArgument>
     <PossiblyUndefinedArrayOffset>
       <code><![CDATA[$update1]]></code>
     </PossiblyUndefinedArrayOffset>

src/Metadata/Subscriber/AttributeSubscriberMetadataFactory.php

@@ -40,7 +40,7 @@ public function metadata(string $subscriber): SubscriberMetadata
         $methods = $reflector->getMethods();
 
         $subscribeMethods = [];
-        $createMethod = null;
+        $setupMethod = null;
         $teardownMethod = null;
 
         foreach ($methods as $method) {
@@ -54,15 +54,15 @@ public function metadata(string $subscriber): SubscriberMetadata
             }
 
             if ($method->getAttributes(Setup::class)) {
-                if ($createMethod !== null) {
+                if ($setupMethod !== null) {
                     throw new DuplicateSetupMethod(
                         $subscriber,
-                        $createMethod,
+                        $setupMethod,
                         $method->getName(),
                     );
                 }
 
-                $createMethod = $method->getName();
+                $setupMethod = $method->getName();
             }
 
             if (!$method->getAttributes(Teardown::class)) {
@@ -85,7 +85,7 @@ public function metadata(string $subscriber): SubscriberMetadata
             $subscriberInfo->group,
             $subscriberInfo->runMode,
             $subscribeMethods,
-            $createMethod,
+            $setupMethod,
             $teardownMethod,
         );
 

src/Subscription/Engine/DefaultSubscriptionEngine.php

@@ -1018,8 +1018,12 @@ private function ensureCommitBatch(Subscription $subscription, int $index): Erro
             return null;
         }
 
+        $subscriber = $this->batching[$subscription->id()];
+
+        unset($this->batching[$subscription->id()]);
+
         try {
-            $this->batching[$subscription->id()]->commitBatch();
+            $subscriber->commitBatch();
             $subscription->changePosition($index);
         } catch (Throwable $e) {
             $this->logger?->error(sprintf(

tests/Unit/Fixture/BatchingSubscriber.php

@@ -0,0 +1,78 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Unit\Fixture;
+
+use Patchlevel\EventSourcing\Attribute\Subscribe;
+use Patchlevel\EventSourcing\Attribute\Subscriber;
+use Patchlevel\EventSourcing\Message\Message;
+use Patchlevel\EventSourcing\Subscription\RunMode;
+use Patchlevel\EventSourcing\Subscription\Subscriber\BatchableSubscriber;
+use Throwable;
+
+use function count;
+
+#[Subscriber(BatchingSubscriber::ID, RunMode::FromBeginning)]
+final class BatchingSubscriber implements BatchableSubscriber
+{
+    public const ID = 'test';
+
+    /** @var list<Message> */
+    public array $receivedMessages = [];
+
+    public int $beginBatchCalled = 0;
+    public int $commitBatchCalled = 0;
+    public int $rollbackBatchCalled = 0;
+
+    public function __construct(
+        public readonly Throwable|null $throwForMessage = null,
+        public readonly Throwable|null $throwForBeginBatch = null,
+        public readonly Throwable|null $throwForCommitBatch = null,
+        public readonly Throwable|null $throwForRollbackBatch = null,
+        public readonly int $forceCommitAfterMessages = 1_000,
+    ) {
+    }
+
+    #[Subscribe(ProfileVisited::class)]
+    public function handle(Message $message): void
+    {
+        $this->receivedMessages[] = $message;
+
+        if ($this->throwForMessage !== null) {
+            throw $this->throwForMessage;
+        }
+    }
+
+    public function beginBatch(): void
+    {
+        $this->beginBatchCalled++;
+
+        if ($this->throwForBeginBatch !== null) {
+            throw $this->throwForBeginBatch;
+        }
+    }
+
+    public function commitBatch(): void
+    {
+        $this->commitBatchCalled++;
+
+        if ($this->throwForCommitBatch !== null) {
+            throw $this->throwForCommitBatch;
+        }
+    }
+
+    public function rollbackBatch(): void
+    {
+        $this->rollbackBatchCalled++;
+
+        if ($this->throwForRollbackBatch !== null) {
+            throw $this->throwForRollbackBatch;
+        }
+    }
+
+    public function forceCommit(): bool
+    {
+        return $this->forceCommitAfterMessages <= count($this->receivedMessages);
+    }
+}