<?php

/*
 * This file is part of the Predis package.
 *
 * (c) 2009-2020 Daniele Alessandri
 * (c) 2021-2026 Till Krüss
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Transaction;

use Exception;
use PHPUnit\Framework\MockObject\MockObject;
use Predis\Client;
use Predis\ClientInterface;
use Predis\Cluster\RedisStrategy;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\Connection\Cluster\ClusterInterface;
use Predis\Connection\Cluster\RedisCluster;
use Predis\Connection\ConnectionException;
use Predis\Connection\FactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Connection\Parameters;
use Predis\Connection\StreamConnection;
use Predis\Response;
use Predis\Retry\Retry;
use Predis\Retry\Strategy\ExponentialBackoff;
use Predis\TimeoutException;
use Predis\Transaction\Exception\TransactionException;
use PredisTestCase;
use RuntimeException;
use Throwable;
use TypeError;

/**
 * @group realm-transaction
 */
class MultiExecTest extends PredisTestCase
{
    /**
     * @group disconnected
     */
    public function testThrowsExceptionOnUnsupportedMultiExecInCommandFactory(): void
    {
        $this->expectException('Predis\NotSupportedException');
        $this->expectExceptionMessage('MULTI, EXEC and DISCARD are not supported by the current command factory.');

        $commands = $this->getMockBuilder('Predis\Command\FactoryInterface')->getMock();
        $commands
            ->expects($this->once())
            ->method('supports')
            ->with('MULTI', 'EXEC', 'DISCARD')
            ->willReturn(false);

        $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
        $client = new Client($connection, ['commands' => $commands]);

        new MultiExec($client);
    }

    /**
     * @group disconnected
     */
    public function testThrowsExceptionOnUnsupportedWatchInCommandFactory(): void
    {
        $this->expectException('Predis\NotSupportedException');
        $this->expectExceptionMessage('WATCH is not supported by the current command factory.');

        $commands = $this->getMockBuilder('Predis\Command\FactoryInterface')->getMock();
        $commands
            ->expects($this->exactly(2))
            ->method('supports')
            ->withConsecutive(
                ['MULTI', 'EXEC', 'DISCARD'],
                ['WATCH']
            )
            ->willReturnOnConsecutiveCalls(
                true,
                false
            );

        $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
        $client = new Client($connection, ['commands' => $commands]);

        $tx = new MultiExec($client, ['options' => 'cas']);
        $tx->watch('foo');
    }

    /**
     * @group disconnected
     */
    public function testThrowsExceptionOnUnsupportedUnwatchInCommandFactory(): void
    {
        $this->expectException('Predis\NotSupportedException');
        $this->expectExceptionMessage('UNWATCH is not supported by the current command factory.');

        $commands = $this->getMockBuilder('Predis\Command\FactoryInterface')->getMock();

        $commands = $this->getMockBuilder('Predis\Command\FactoryInterface')->getMock();
        $commands
            ->expects($this->exactly(2))
            ->method('supports')
            ->withConsecutive(
                ['MULTI', 'EXEC', 'DISCARD'],
                ['UNWATCH']
            )
            ->willReturnOnConsecutiveCalls(
                true,
                false
            );

        $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
        $client = new Client($connection, ['commands' => $commands]);

        $tx = new MultiExec($client, ['options' => 'cas']);

        $tx->unwatch('foo');
    }

    /**
     * @group disconnected
     */
    public function testExecutionWithFluentInterface(): void
    {
        $commands = [];
        $expected = ['one', 'two', 'three'];

        $callback = $this->getExecuteCallback($expected, $commands);
        $tx = $this->getMockedTransaction($callback);

        $this->assertSame($expected, $tx->echo('one')->echo('two')->echo('three')->execute());
        $this->assertSame(['MULTI', 'ECHO', 'ECHO', 'ECHO', 'EXEC'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testExecutionWithCallable(): void
    {
        $commands = [];
        $expected = ['one', 'two', 'three'];

        $callback = $this->getExecuteCallback($expected, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = $tx->execute(static function ($tx) {
            $tx->echo('one');
            $tx->echo('two');
            $tx->echo('three');
        });

        $this->assertSame($expected, $responses);
        $this->assertSame(['MULTI', 'ECHO', 'ECHO', 'ECHO', 'EXEC'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testCannotMixExecutionWithFluentInterfaceAndCallable(): void
    {
        $commands = [];

        $callback = $this->getExecuteCallback(null, $commands);
        $tx = $this->getMockedTransaction($callback);

        $exception = null;

        try {
            $tx->echo('foo')->execute(static function ($tx) {
                $tx->echo('bar');
            });
        } catch (Exception $ex) {
            $exception = $ex;
        }

        $this->assertInstanceOf('Predis\ClientException', $exception);
        $this->assertSame(['MULTI', 'ECHO', 'DISCARD'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testEmptyTransactionDoesNotSendMultiExecCommands(): void
    {
        $commands = [];

        $callback = $this->getExecuteCallback(null, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = $tx->execute(static function ($tx) {
            // NOOP
        });

        $this->assertNull($responses);
        $this->assertSame([], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testThrowsExceptionOnExecInsideTransactionBlock(): void
    {
        $this->expectException('Predis\ClientException');
        $this->expectExceptionMessage('Cannot invoke "execute" or "exec" inside an active transaction context');

        $commands = [];

        $callback = $this->getExecuteCallback(null, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = $tx->execute(static function ($tx) {
            $tx->exec();
        });

        $this->assertNull($responses);
        $this->assertSame([], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testEmptyTransactionIgnoresDiscard(): void
    {
        $commands = [];

        $callback = $this->getExecuteCallback(null, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = $tx->execute(static function ($tx) {
            $tx->discard();
        });

        $this->assertNull($responses);
        $this->assertSame([], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testTransactionWithCommandsSendsDiscard(): void
    {
        $commands = [];

        $callback = $this->getExecuteCallback(null, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = $tx->execute(static function ($tx) {
            $tx->set('foo', 'bar');
            $tx->get('foo');
            $tx->discard();
        });

        $this->assertNull($responses);
        $this->assertSame(['MULTI', 'SET', 'GET', 'DISCARD'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testSendMultiOnCommandsFollowingDiscard(): void
    {
        $commands = [];
        $expected = ['after DISCARD'];

        $callback = $this->getExecuteCallback($expected, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = $tx->execute(static function ($tx) {
            $tx->echo('before DISCARD');
            $tx->discard();
            $tx->echo('after DISCARD');
        });

        $this->assertSame($responses, $expected);
        $this->assertSame(['MULTI', 'ECHO', 'DISCARD', 'MULTI', 'ECHO', 'EXEC'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testThrowsExceptionOnWatchInsideMulti(): void
    {
        $this->expectException('Predis\ClientException');

        $callback = $this->getExecuteCallback();
        $tx = $this->getMockedTransaction($callback);

        $tx->echo('foobar')->watch('foo')->execute();
    }

    /**
     * @group disconnected
     */
    public function testUnwatchInsideMulti(): void
    {
        $commands = [];
        $expected = ['foobar', true];

        $callback = $this->getExecuteCallback($expected, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = $tx->echo('foobar')->unwatch('foo')->execute();

        $this->assertSame($responses, $expected);
        $this->assertSame(['MULTI', 'ECHO', 'UNWATCH', 'EXEC'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testAutomaticWatchInOptions(): void
    {
        $txCommands = $casCommands = [];
        $expected = ['bar', 'piyo'];
        $options = ['watch' => ['foo', 'hoge']];

        $callback = $this->getExecuteCallback($expected, $txCommands, $casCommands);
        $tx = $this->getMockedTransaction($callback, $options);

        $responses = $tx->execute(static function ($tx) {
            $tx->get('foo');
            $tx->get('hoge');
        });

        $this->assertSame($responses, $expected);
        $this->assertSame(['WATCH'], self::commandsToIDs($casCommands));
        $this->assertSame(['foo', 'hoge'], $casCommands[0]->getArguments());
        $this->assertSame(['MULTI', 'GET', 'GET', 'EXEC'], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     */
    public function testCheckAndSetWithFluentInterface(): void
    {
        $txCommands = $casCommands = [];
        $expected = ['bar', 'piyo'];
        $options = ['cas' => true, 'watch' => ['foo', 'hoge']];

        $callback = $this->getExecuteCallback($expected, $txCommands, $casCommands);
        $tx = $this->getMockedTransaction($callback, $options);

        $tx->watch('foobar');
        $this->assertSame('DUMMY_RESPONSE', $tx->get('foo'));
        $this->assertSame('DUMMY_RESPONSE', $tx->get('hoge'));

        $responses = $tx
            ->multi()
            ->get('foo')
            ->get('hoge')
            ->execute();

        $this->assertSame($responses, $expected);
        $this->assertSame(['WATCH', 'WATCH', 'GET', 'GET'], self::commandsToIDs($casCommands));
        $this->assertSame(['MULTI', 'GET', 'GET', 'EXEC'], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     */
    public function testCheckAndSetWithBlock(): void
    {
        $txCommands = $casCommands = [];
        $expected = ['bar', 'piyo'];
        $options = ['cas' => true, 'watch' => ['foo', 'hoge']];

        $callback = $this->getExecuteCallback($expected, $txCommands, $casCommands);
        $tx = $this->getMockedTransaction($callback, $options);

        $test = $this;
        $responses = $tx->execute(static function ($tx) use ($test) {
            $tx->watch('foobar');

            $response1 = $tx->get('foo');
            $response2 = $tx->get('hoge');

            $test->assertSame('DUMMY_RESPONSE', $response1);
            $test->assertSame('DUMMY_RESPONSE', $response2);

            $tx->multi();

            $tx->get('foo');
            $tx->get('hoge');
        });

        $this->assertSame($responses, $expected);
        $this->assertSame(['WATCH', 'WATCH', 'GET', 'GET'], self::commandsToIDs($casCommands));
        $this->assertSame(['MULTI', 'GET', 'GET', 'EXEC'], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     */
    public function testEmptyExplicitTransactionIsDiscarded(): void
    {
        $commands = [];
        $tx = $this->getMockedTransaction($this->getExecuteCallback([], $commands));

        $tx->execute(static function ($tx) {
            $tx->multi();
        });

        $this->assertSame(['MULTI', 'DISCARD'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testCallbackPhpErrorDiscardsTransaction(): void
    {
        $commands = [];
        $tx = $this->getMockedTransaction($this->getExecuteCallback([], $commands));

        try {
            $tx->execute(static function ($tx) {
                $tx->set('foo', 'bar');

                throw new TypeError('Invalid callback argument');
            });
            $this->fail('Expected the callback error to be rethrown.');
        } catch (TypeError $exception) {
            $this->assertSame('Invalid callback argument', $exception->getMessage());
        }

        $this->assertSame(['MULTI', 'SET', 'DISCARD'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testCheckAndSetWithEmptyBlock(): void
    {
        $txCommands = $casCommands = [];
        $options = ['cas' => true];

        $callback = $this->getExecuteCallback([], $txCommands, $casCommands);
        $tx = $this->getMockedTransaction($callback, $options);

        $tx->execute(static function ($tx) {
            $tx->multi();
        });

        $this->assertSame([], self::commandsToIDs($casCommands));
        $this->assertSame(['MULTI', 'DISCARD'], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     */
    public function testCheckAndSetWithoutExec(): void
    {
        $txCommands = $casCommands = [];
        $options = ['cas' => true];

        $callback = $this->getExecuteCallback([], $txCommands, $casCommands);
        $tx = $this->getMockedTransaction($callback, $options);

        $tx->execute(static function ($tx) {
            $tx->get('foo');
            $tx->set('hoge', 'piyo');
        });

        $this->assertSame(['GET', 'SET'], self::commandsToIDs($casCommands));
        $this->assertSame([], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     */
    public function testThrowsExceptionOnAutomaticRetriesWithFluentInterface(): void
    {
        $this->expectException('Predis\ClientException');
        $this->expectExceptionMessage('Automatic retries are supported only when a callable block is provided');

        $options = ['retry' => 1];

        $callback = $this->getExecuteCallback();
        $tx = $this->getMockedTransaction($callback, $options);

        $tx->echo('message')->execute();
    }

    /**
     * @group disconnected
     */
    public function testAutomaticRetryOnServerSideTransactionAbort(): void
    {
        $casCommands = $txCommands = [];
        $expected = ['bar'];
        $options = ['watch' => ['foo', 'bar'], 'retry' => ($attempts = 2) + 1];

        $signal = $this->getMockBuilder('stdClass')
            ->addMethods(['__invoke'])
            ->getMock();
        $signal
            ->expects($this->exactly($attempts))
            ->method('__invoke');

        $callback = $this->getExecuteCallback($expected, $txCommands, $casCommands);
        $tx = $this->getMockedTransaction($callback, $options);

        $responses = $tx->execute(static function (MultiExec $tx) use ($signal, &$attempts) {
            $tx->get('foo');

            if ($attempts > 0) {
                $attempts--;
                $signal();

                $tx->echo('!!ABORT!!');
            }
        });

        $this->assertSame($responses, $expected);
        $this->assertSame(['WATCH'], self::commandsToIDs($casCommands));
        $this->assertSame(['foo', 'bar'], $casCommands[0]->getArguments());
        $this->assertSame(['MULTI', 'GET', 'EXEC'], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     */
    public function testThrowsExceptionOnServerSideTransactionAbort(): void
    {
        $this->expectException('Predis\Transaction\AbortedMultiExecException');

        $callback = $this->getExecuteCallback();
        $tx = $this->getMockedTransaction($callback);

        $tx->execute(static function ($tx) {
            $tx->echo('!!ABORT!!');
        });
    }

    /**
     * @group disconnected
     */
    public function testHandlesStandardExceptionsInBlock(): void
    {
        $commands = [];
        $expected = ['foobar', true];

        $callback = $this->getExecuteCallback($expected, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = null;

        try {
            $responses = $tx->execute(static function (MultiExec $tx) {
                $tx->set('foo', 'bar');
                $tx->get('foo');

                throw new RuntimeException('TEST');
            });
        } catch (Exception $ex) {
            // NOOP
        }

        $this->assertNull($responses);
        $this->assertIsArray($expected);
        $this->assertSame(['MULTI', 'SET', 'GET', 'DISCARD'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testHandlesServerExceptionsInBlock(): void
    {
        $commands = [];
        $expected = ['foobar', true];

        $callback = $this->getExecuteCallback($expected, $commands);
        $tx = $this->getMockedTransaction($callback);

        $responses = null;

        try {
            $responses = $tx->execute(static function (MultiExec $tx) {
                $tx->set('foo', 'bar');
                $tx->echo('ERR Invalid operation');
                $tx->get('foo');
            });
        } catch (Response\ServerException $ex) {
            $tx->discard();
        }

        $this->assertNull($responses);
        $this->assertSame(['MULTI', 'SET', 'ECHO', 'DISCARD'], self::commandsToIDs($commands));
    }

    /**
     * @group disconnected
     */
    public function testProperlyDiscardsTransactionAfterServerExceptionInBlock(): void
    {
        $connection = $this->getMockedConnection(static function (CommandInterface $command) {
            switch ($command->getId()) {
                case 'MULTI':
                    return true;

                case 'ECHO':
                    return new Response\Error('ERR simulated failure on ECHO');

                case 'EXEC':
                    return new Response\Error('EXECABORT Transaction discarded because of previous errors.');

                default:
                    return new Response\Status('QUEUED');
            }
        });

        $client = new Client($connection);

        // First attempt
        $tx = new MultiExec($client);

        try {
            $tx->multi()->set('foo', 'bar')->echo('simulated failure')->exec();
        } catch (Exception $exception) {
            $this->assertInstanceOf('Predis\Transaction\AbortedMultiExecException', $exception);
            $this->assertSame('ERR simulated failure on ECHO', $exception->getMessage());
        }

        // Second attempt
        $tx = new MultiExec($client);

        try {
            $tx->multi()->set('foo', 'bar')->echo('simulated failure')->exec();
        } catch (Exception $exception) {
            $this->assertInstanceOf('Predis\Transaction\AbortedMultiExecException', $exception);
            $this->assertSame('ERR simulated failure on ECHO', $exception->getMessage());
        }
    }

    /**
     * @group disconnected
     */
    public function testExceptionsOptionTakesPrecedenceOverClientOptionsWhenFalse(): void
    {
        $expected = ['before', new Response\Error('ERR simulated error'), 'after'];

        $connection = $this->getMockedConnection(static function (CommandInterface $command) use ($expected) {
            switch ($command->getId()) {
                case 'MULTI':
                    return true;

                case 'EXEC':
                    return $expected;

                default:
                    return new Response\Status('QUEUED');
            }
        });

        $client = new Client($connection, ['exceptions' => true]);
        $tx = new MultiExec($client, ['exceptions' => false]);

        $result = $tx
            ->multi()
            ->echo('before')
            ->echo('ERROR PLEASE!')
            ->echo('after')
            ->exec();

        $this->assertSame($expected, $result);
    }

    /**
     * @group disconnected
     */
    public function testExceptionsOptionTakesPrecedenceOverClientOptionsWhenTrue(): void
    {
        $this->expectException('Predis\Response\ServerException');
        $this->expectExceptionMessage('ERR simulated error');

        $expected = ['before', new Response\Error('ERR simulated error'), 'after'];

        $connection = $this->getMockedConnection(static function (CommandInterface $command) use ($expected) {
            switch ($command->getId()) {
                case 'MULTI':
                    return true;

                case 'EXEC':
                    return $expected;

                default:
                    return new Response\Status('QUEUED');
            }
        });

        $client = new Client($connection, ['exceptions' => false]);
        $tx = new MultiExec($client, ['exceptions' => true]);

        $tx->multi()->echo('before')->echo('ERROR PLEASE!')->echo('after')->exec();
    }

    /**
     * @group disconnected
     */
    public function testExceptionsOptionDoesNotAffectTransactionControlCommands(): void
    {
        $this->expectException('Predis\Response\ServerException');
        $this->expectExceptionMessage('ERR simulated failure on EXEC');

        $connection = $this->getMockedConnection(static function (CommandInterface $command) {
            switch ($command->getId()) {
                case 'MULTI':
                    return true;

                case 'EXEC':
                    return new Response\Error('ERR simulated failure on EXEC');

                default:
                    return new Response\Status('QUEUED');
            }
        });

        $client = new Client($connection, ['exceptions' => false]);
        $tx = new MultiExec($client);

        $tx->multi()->echo('test')->exec();
    }

    /**
     * @group disconnected
     * @throws Exception
     */
    public function testRetryReplicationPipelineOnRetryableErrors(): void
    {
        $parameters = new Parameters([
            'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
            'role' => 'master',
        ]);

        $mockConnection = $this->getMockConnection();

        $mockConnection
            ->expects($this->any())
            ->method('getParameters')
            ->willReturn($parameters);

        $mockConnection
            ->expects($this->atLeast(3))
            ->method('disconnect')
            ->withAnyParameters();

        $mockConnection
            ->expects($this->exactly(8))
            ->method('executeCommand')
            ->withAnyParameters()
            ->willReturnOnConsecutiveCalls(
                $this->throwException(new TimeoutException($mockConnection)),
                $this->throwException(new TimeoutException($mockConnection)),
                $this->throwException(new TimeoutException($mockConnection)),
                new Response\Status('OK'),
                new Response\Status('QUEUED'),
                new Response\Status('QUEUED'),
                new Response\Status('QUEUED'),
                ['OK', 'OK', 'OK']
            );

        $tx = new MultiExec(new Client($mockConnection));

        $responses = $tx->execute(static function (MultiExec $tx) {
            $tx->set('key', 'value');
            $tx->set('key', 'value');
            $tx->set('key', 'value');
        });

        $this->assertEquals(['OK', 'OK', 'OK'], $responses);
    }

    /**
     * @group disconnected
     */
    public function testClusterCheckAndSetSendsMultiOnceAfterReads(): void
    {
        $txCommands = $casCommands = [];
        $expected = ['bar', 'piyo'];
        $options = ['cas' => true, 'watch' => ['{cas}foo', '{cas}hoge']];

        $callback = $this->getExecuteCallback($expected, $txCommands, $casCommands);
        $tx = $this->getMockedClusterTransaction($callback, $options);

        $test = $this;
        $responses = $tx->execute(static function ($tx) use ($test) {
            $test->assertSame('DUMMY_RESPONSE', $tx->get('{cas}foo'));

            $tx->multi();

            $tx->get('{cas}foo');
            $tx->get('{cas}hoge');
        });

        $this->assertSame($responses, $expected);
        $this->assertSame(['WATCH', 'GET'], self::commandsToIDs($casCommands));
        $this->assertSame(['MULTI', 'GET', 'GET', 'EXEC'], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     * @dataProvider blockExceptions
     */
    public function testClusterCheckAndSetNeverLeavesMultiOpenWhenBlockThrows(Throwable $exception): void
    {
        $txCommands = $casCommands = [];
        $options = ['cas' => true, 'watch' => '{cas}foo'];

        $callback = $this->getExecuteCallback([], $txCommands, $casCommands);
        $tx = $this->getMockedClusterTransaction($callback, $options);
        $caught = null;

        try {
            $tx->execute(static function ($tx) use ($exception) {
                $tx->get('{cas}foo');
                $tx->multi();
                $tx->set('{cas}foo', 'bar');

                throw $exception;
            });
        } catch (Throwable $caught) {
            // NOOP
        }

        $this->assertSame($exception, $caught);
        $this->assertSame(['WATCH', 'GET'], array_slice(self::commandsToIDs($casCommands), 0, 2));
        $this->assertSame([], self::commandsToIDs($txCommands));
    }

    public function blockExceptions(): array
    {
        return [
            'PHP error' => [new TypeError('Invalid callback argument')],
            'server error' => [new Response\ServerException('ERR simulated failure')],
            'connection error' => [new ConnectionException(new StreamConnection(new Parameters()), 'Connection lost')],
        ];
    }

    /**
     * @group disconnected
     */
    public function testClusterAutomaticRetryOnServerSideTransactionAbort(): void
    {
        $casCommands = $txCommands = [];
        $expected = ['bar'];
        $options = ['watch' => ['{cas}foo', '{cas}bar'], 'retry' => ($attempts = 2) + 1];

        $callback = $this->getExecuteCallback($expected, $txCommands, $casCommands);
        $tx = $this->getMockedClusterTransaction($callback, $options);

        $responses = $tx->execute(static function (MultiExec $tx) use (&$attempts) {
            $tx->get('{cas}foo');

            if ($attempts > 0) {
                $attempts--;

                $tx->echo('!!ABORT!!');
            }
        });

        $this->assertSame($responses, $expected);
        $this->assertSame(0, $attempts);
        $this->assertSame(['WATCH'], self::commandsToIDs($casCommands));
        $this->assertSame(['MULTI', 'GET', 'EXEC'], self::commandsToIDs($txCommands));
    }

    /**
     * @group disconnected
     */
    public function testClusterRetryReachesTheNodeTheSlotWasMovedTo(): void
    {
        $seedCommands = $targetCommands = [];
        $slot = (new RedisStrategy())->getSlotByKey('foo');

        $seed = $this->getMockConnection('tcp://127.0.0.1:6379');
        $seed
            ->method('executeCommand')
            ->willReturnCallback(static function (CommandInterface $command) use (&$seedCommands, $slot) {
                $seedCommands[] = $id = $command->getId();

                return $id === 'GET'
                    ? new Response\Error("MOVED $slot 127.0.0.1:6380")
                    : new Response\Status('OK');
            });

        $target = $this->getMockConnection('tcp://127.0.0.1:6380');
        $target
            ->method('executeCommand')
            ->willReturnCallback(static function (CommandInterface $command) use (&$targetCommands) {
                $targetCommands[] = $id = $command->getId();

                switch ($id) {
                    case 'CLUSTER':
                        return [[0, 16383, ['127.0.0.1', 6380]]];

                    case 'MULTI':
                        return new Response\Status('OK');

                    case 'EXEC':
                        return ['bar'];

                    default:
                        return new Response\Status('QUEUED');
                }
            });

        $factory = $this->getMockBuilder(FactoryInterface::class)->getMock();
        $factory
            ->expects($this->once())
            ->method('create')
            ->with(['host' => '127.0.0.1', 'port' => '6380', 'database' => null])
            ->willReturn($target);

        $cluster = new RedisCluster($factory, new Parameters(['protocol' => 2]));
        $cluster->add($seed);

        $tx = new MultiExec(new Client($cluster), ['retry' => 1]);

        $responses = $tx->execute(static function (MultiExec $tx) {
            $tx->get('foo');
        });

        $this->assertSame(['bar'], $responses);
        $this->assertSame(['MULTI', 'GET', 'DISCARD'], $seedCommands);
        $this->assertSame(['CLUSTER', 'MULTI', 'GET', 'EXEC'], $targetCommands);
    }

    /**
     * @group disconnected
     */
    public function testClusterFollowsRedirectionOfTheFirstCommand(): void
    {
        $seedCommands = $targetCommands = [];
        $slot = (new RedisStrategy())->getSlotByKey('foo');

        $seed = $this->getMockConnection('tcp://127.0.0.1:6379');
        $seed
            ->method('executeCommand')
            ->willReturnCallback(static function (CommandInterface $command) use (&$seedCommands, $slot) {
                $seedCommands[] = $command->getId();

                return new Response\Error("MOVED $slot 127.0.0.1:6380");
            });

        $target = $this->getMockConnection('tcp://127.0.0.1:6380');
        $target
            ->method('executeCommand')
            ->willReturnCallback($this->getClusterNodeCallback($targetCommands));

        $factory = $this->getMockBuilder(FactoryInterface::class)->getMock();
        $factory
            ->expects($this->once())
            ->method('create')
            ->with(['host' => '127.0.0.1', 'port' => '6380', 'database' => null])
            ->willReturn($target);

        $cluster = new RedisCluster($factory, new Parameters(['protocol' => 2]));
        $cluster->add($seed);

        $tx = new MultiExec(new Client($cluster), ['watch' => 'foo']);

        $responses = $tx->execute(static function (MultiExec $tx) {
            $tx->get('foo');
        });

        $this->assertSame(['bar'], $responses);
        $this->assertSame(['WATCH'], $seedCommands);
        $this->assertSame(['CLUSTER', 'WATCH', 'MULTI', 'GET', 'EXEC'], $targetCommands);
    }

    /**
     * @group disconnected
     */
    public function testClusterRetriesTheFirstCommandOnConnectionFailure(): void
    {
        $deadCommands = $aliveCommands = [];

        $dead = $this->getMockConnection('tcp://127.0.0.1:6379?slots=0-16383');
        $dead
            ->method('executeCommand')
            ->willReturnCallback(static function (CommandInterface $command) use (&$deadCommands, &$dead) {
                $deadCommands[] = $command->getId();

                throw new ConnectionException($dead, 'Connection lost');
            });

        $alive = $this->getMockConnection('tcp://127.0.0.1:6380');
        $alive
            ->method('executeCommand')
            ->willReturnCallback($this->getClusterNodeCallback($aliveCommands));

        $cluster = new RedisCluster(
            $this->getMockBuilder(FactoryInterface::class)->getMock(),
            new Parameters(['protocol' => 2])
        );
        $cluster->add($dead);
        $cluster->add($alive);

        $tx = new MultiExec(new Client($cluster));

        $responses = $tx->execute(static function (MultiExec $tx) {
            $tx->get('foo');
        });

        $this->assertSame(['bar'], $responses);
        $this->assertSame(['MULTI'], $deadCommands);
        $this->assertSame(['CLUSTER', 'MULTI', 'GET', 'EXEC'], $aliveCommands);
    }

    /**
     * @group disconnected
     */
    public function testClusterTransactionReachesAnotherNodeAfterTheFirstOneDied(): void
    {
        $deadCommands = $aliveCommands = [];

        $dead = $this->getMockConnection('tcp://127.0.0.1:6379?slots=0-16383');
        $dead
            ->method('executeCommand')
            ->willReturnCallback(static function (CommandInterface $command) use (&$deadCommands, &$dead) {
                $deadCommands[] = $id = $command->getId();

                if ($id === 'GET') {
                    throw new ConnectionException($dead, 'Connection lost');
                }

                return new Response\Status('OK');
            });

        $alive = $this->getMockConnection('tcp://127.0.0.1:6380');
        $alive
            ->method('executeCommand')
            ->willReturnCallback($this->getClusterNodeCallback($aliveCommands));

        $cluster = new RedisCluster(
            $this->getMockBuilder(FactoryInterface::class)->getMock(),
            new Parameters(['protocol' => 2])
        );
        $cluster->add($dead);
        $cluster->add($alive);

        $client = new Client($cluster);
        $block = static function (MultiExec $tx) {
            $tx->get('foo');
        };

        try {
            (new MultiExec($client))->execute($block);
            $this->fail('Expected the connection failure to abort the transaction.');
        } catch (ConnectionException $exception) {
            $this->assertSame('Connection lost', $exception->getMessage());
        }

        // The commands of the aborted transaction are not sent anywhere else.
        $this->assertSame(['MULTI', 'GET'], $deadCommands);
        $this->assertSame(['CLUSTER'], $aliveCommands);

        $this->assertSame(['bar'], (new MultiExec($client))->execute($block));
        $this->assertSame(['MULTI', 'GET'], $deadCommands);
        $this->assertSame(['CLUSTER', 'MULTI', 'GET', 'EXEC'], $aliveCommands);
    }

    /**
     * Returns a callback emulating the only node of a cluster that is able to
     * complete a transaction, keeping track of the commands it receives.
     *
     * @param array $commands Reference to an array storing the IDs of the commands
     *
     * @return callable
     */
    protected function getClusterNodeCallback(array &$commands): callable
    {
        return static function (CommandInterface $command) use (&$commands) {
            $commands[] = $id = $command->getId();

            switch ($id) {
                case 'CLUSTER':
                    return [[0, 16383, ['127.0.0.1', 6380]]];

                case 'WATCH':
                case 'MULTI':
                    return new Response\Status('OK');

                case 'EXEC':
                    return ['bar'];

                default:
                    return new Response\Status('QUEUED');
            }
        };
    }

    /**
     * @group connected
     * @group cluster
     * @dataProvider clusterTransactionPersistence
     * @requiresRedisVersion >= 3.0.0
     */
    public function testClusterQueueErrorLeavesConnectionUsable(bool $persistent): void
    {
        $options = ['parameters' => ['persistent' => $persistent]];
        $redis = $this->createClient(null, $options);
        $key = '{abort}value';
        $missing = '{abort}missing';
        $redis->set($key, 'original');
        $redis->del($missing);

        try {
            $redis->transaction(static function (MultiExec $tx) use ($key) {
                $tx->set($key, 'queued');
                $tx->executeCommand(new RawCommand('SET', [$key]));
            });
            $this->fail('Expected the transaction to abort.');
        } catch (AbortedMultiExecException $exception) {
            $this->assertSame('original', $redis->get($key));
        }

        $this->assertNull($redis->get($missing));
        $this->assertEquals('OK', $redis->set($key, 'after-abort'));

        $nextClient = $this->createClient(null, $options, false);
        $this->assertSame('after-abort', $nextClient->get($key));
        $this->assertNull($nextClient->get($missing));
        $redis->disconnect();
    }

    public function clusterTransactionPersistence(): array
    {
        return [[false], [true]];
    }

    /**
     * @group connected
     * @group cluster
     * @dataProvider clusterTransactionPersistence
     * @requiresRedisVersion >= 3.0.0
     */
    public function testClusterFailedDiscardClosesThePhysicalSocket(bool $persistent): void
    {
        $connection = new class(new Parameters()) extends StreamConnection {
            public function executeCommand(CommandInterface $command)
            {
                if ($command->getId() === 'DISCARD') {
                    return new Response\Error('NOPERM simulated DISCARD failure');
                }

                return parent::executeCommand($command);
            }
        };
        $redis = $this->createClient(null, [
            'connections' => ['tcp' => get_class($connection)],
            'parameters' => ['persistent' => $persistent],
        ]);
        $key = '{failed-discard}value';
        $redis->set($key, 'original');
        $command = $redis->createCommand('GET', [$key]);
        $node = $redis->getConnection()->getConnectionByCommand($command);
        $originalClientId = $node->executeCommand(new RawCommand('CLIENT', ['ID']));

        try {
            $redis->transaction(static function (MultiExec $tx) use ($key) {
                $tx->set($key, 'queued');
                $tx->executeCommand(new RawCommand('SET', [$key]));
            });
            $this->fail('Expected the transaction to abort.');
        } catch (AbortedMultiExecException $exception) {
            $this->assertFalse($node->isConnected());
        }

        $this->assertSame('original', $redis->get($key));
        $this->assertNotSame($originalClientId, $node->executeCommand(new RawCommand('CLIENT', ['ID'])));
        $this->assertNull($redis->get('{failed-discard}missing'));
        $this->assertEquals('OK', $redis->set($key, 'after-abort'));
        $this->assertSame('after-abort', $redis->get($key));
        $redis->disconnect();
    }

    /**
     * @group connected
     * @group cluster
     * @requiresRedisVersion >= 3.0.0
     */
    public function testClusterCasReadsValueBeforeMulti(): void
    {
        $redis = $this->getClient();
        $key = '{cas}counter';
        $redis->set($key, '7');

        $response = $redis->transaction(['cas' => true, 'watch' => $key], function (MultiExec $tx) use ($key) {
            $current = $tx->get($key);
            $this->assertSame('7', $current);
            $tx->multi();
            $tx->set($key, (string) ((int) $current + 1));
        });

        $this->assertEquals(['OK'], $response);
        $this->assertSame('8', $redis->get($key));
        $this->assertNull($redis->get('{cas}missing'));
    }

    /**
     * @group connected
     * @group cluster
     * @dataProvider blockExceptions
     * @requiresRedisVersion >= 3.0.0
     */
    public function testClusterCasBlockExceptionLeavesConnectionUsable(Throwable $exception): void
    {
        $redis = $this->getClient();
        $key = '{cas}counter';
        $redis->set($key, '7');
        $caught = null;

        try {
            $redis->transaction(['cas' => true, 'watch' => $key], static function (MultiExec $tx) use ($key, $exception) {
                $tx->multi();
                $tx->set($key, 'queued');

                throw $exception;
            });
        } catch (Throwable $caught) {
            // NOOP
        }

        $this->assertSame($exception, $caught);
        $this->assertSame('7', $redis->get($key));
        $this->assertNull($redis->get('{cas}missing'));
        $this->assertEquals('OK', $redis->set($key, 'after-error'));
        $this->assertSame('after-error', $redis->get($key));
    }

    // ******************************************************************** //
    // ---- INTEGRATION TESTS --------------------------------------------- //
    // ******************************************************************** //

    /**
     * @group connected
     */
    public function testIntegrationHandlesStandardExceptionsInBlock(): void
    {
        $client = $this->getClient();
        $exception = null;

        try {
            $client->transaction(static function (MultiExec $tx) {
                $tx->set('foo', 'bar');
                throw new RuntimeException('TEST');
            });
        } catch (Exception $ex) {
            $exception = $ex;
        }

        $this->assertInstanceOf('RuntimeException', $exception);
        $this->assertSame(0, $client->exists('foo'));
    }

    /**
     * @group connected
     */
    public function testIntegrationThrowsExceptionOnRedisErrorInBlock(): void
    {
        $client = $this->getClient();
        $exception = null;
        $value = (string) rand();

        try {
            $client->transaction(static function (MultiExec $tx) use ($value) {
                $tx->set('foo', 'bar');
                $tx->lpush('foo', 'bar');
                $tx->set('foo', $value);
            });
        } catch (Response\ServerException $ex) {
            $exception = $ex;
        }

        $this->assertInstanceOf('Predis\Response\ErrorInterface', $exception);
        $this->assertSame($value, $client->get('foo'));
    }

    /**
     * @group connected
     * @group relay-incompatible
     */
    public function testIntegrationReturnsErrorObjectOnRedisErrorInBlock(): void
    {
        $client = $this->getClient([], ['exceptions' => false]);

        $responses = $client->transaction(static function (MultiExec $tx) {
            $tx->set('foo', 'bar');
            $tx->lpush('foo', 'bar');
            $tx->echo('foobar');
        });

        $this->assertInstanceOf('Predis\Response\Status', $responses[0]);
        $this->assertInstanceOf('Predis\Response\Error', $responses[1]);
        $this->assertSame('foobar', $responses[2]);
    }

    /**
     * @group connected
     * @group ext-relay
     */
    public function testIntegrationReturnsErrorObjectOnRedisErrorInBlockWhenUsingRelay(): void
    {
        $client = $this->getClient([], ['exceptions' => false]);

        $responses = $client->transaction(static function (MultiExec $tx) {
            $tx->set('foo', 'bar');
            $tx->lpush('foo', 'bar');
            $tx->echo('foobar');
        });

        $this->assertSame('OK', $responses[0]);
        $this->assertInstanceOf('Predis\Response\Error', $responses[1]);
        $this->assertSame('foobar', $responses[2]);
    }

    /**
     * @group connected
     * @requiresRedisVersion >= 2.0.0
     */
    public function testIntegrationSendMultiOnCommandsAfterDiscard(): void
    {
        $client = $this->getClient();

        $responses = $client->transaction(static function (MultiExec $tx) {
            $tx->set('foo', 'bar');
            $tx->discard();
            $tx->set('hoge', 'piyo');
        });

        $this->assertCount(1, $responses);
        $this->assertSame(0, $client->exists('foo'));
        $this->assertSame(1, $client->exists('hoge'));
    }

    /**
     * @group connected
     * @group relay-incompatible
     * @requiresRedisVersion >= 2.2.0
     */
    public function testIntegrationWritesOnWatchedKeysAbortTransaction(): void
    {
        $exception = null;
        $client1 = $this->getClient();
        $client2 = $this->getClient();

        try {
            $client1->transaction(['watch' => 'sentinel'], static function ($tx) use ($client2) {
                $tx->set('sentinel', 'client1');
                $tx->get('sentinel');
                $client2->set('sentinel', 'client2');
            });
        } catch (AbortedMultiExecException $ex) {
            $exception = $ex;
        }

        $this->assertInstanceOf('Predis\Transaction\AbortedMultiExecException', $exception);
        $this->assertSame('client2', $client1->get('sentinel'));
    }

    /**
     * @group connected
     * @group ext-relay
     * @requiresRedisVersion >= 2.2.0
     */
    public function testRelayIntegrationWritesOnWatchedKeysAbortTransaction(): void
    {
        $exception = null;
        $client1 = $this->getClient();
        $client2 = $this->getClient();

        try {
            $client1->transaction(['watch' => 'sentinel'], static function ($tx) use ($client2) {
                $tx->set('sentinel', 'client1');
                $tx->get('sentinel');
                $client2->set('sentinel', 'client2');
            });
        } catch (Response\ServerException $ex) {
            $exception = $ex;
        }

        $this->assertInstanceOf(Response\ServerException::class, $exception);
        $this->assertSame('client2', $client1->get('sentinel'));
    }

    /**
     * @group connected
     * @requiresRedisVersion >= 2.2.0
     */
    public function testIntegrationCheckAndSetWithDiscardAndRetry(): void
    {
        $client = $this->getClient();

        $client->set('foo', 'bar');
        $options = ['watch' => 'foo', 'cas' => true];

        $responses = $client->transaction($options, static function ($tx) {
            $tx->watch('foobar');
            $foo = $tx->get('foo');

            $tx->multi();
            $tx->set('foobar', $foo);
            $tx->discard();
            $tx->mget('foo', 'foobar');
        });

        $this->assertIsArray($responses);
        $this->assertSame([['bar', null]], $responses);

        $hijack = true;
        $client2 = $this->getClient();
        $client->set('foo', 'bar');

        $options = ['watch' => 'foo', 'cas' => true, 'retry' => 1];
        $responses = $client->transaction($options, static function ($tx) use ($client2, &$hijack) {
            $foo = $tx->get('foo');
            $tx->multi();

            $tx->set('foobar', $foo);
            $tx->discard();

            if ($hijack) {
                $hijack = false;
                $client2->set('foo', 'hijacked!');
            }

            $tx->mget('foo', 'foobar');
        });

        $this->assertIsArray($responses);
        $this->assertSame([['hijacked!', null]], $responses);
    }

    /**
     * @group connected
     * @group cluster
     * @return void
     * @requiresRedisVersion >= 3.0.0
     */
    public function testExecutesTransactionAgainstCluster(): void
    {
        $redis = $this->getClient();

        $response = $redis->transaction(static function (MultiExec $tx) {
            $tx->set('{foo}foo', 'value');
            $tx->set('{foo}bar', 'value');
            $tx->set('{foo}baz', 'value');
        });

        $this->assertEquals(['OK', 'OK', 'OK'], $response);
    }

    /**
     * @group connected
     * @group cluster
     * @return void
     * @requiresRedisVersion >= 3.0.0
     */
    public function testThrowsExceptionOnDifferentHashSlots(): void
    {
        $redis = $this->getClient();

        $this->expectException(AbortedMultiExecException::class);
        $this->expectExceptionMessage(
            'To be able to execute a transaction against cluster, all commands should operate on the same hash slot'
        );

        $redis->transaction(static function (MultiExec $tx) {
            $tx->set('foo_bar_baz', 'value');
            $tx->set('{foo}bar', 'value');
            $tx->set('{foo}baz', 'value');
        });
    }

    /**
     * @group connected
     * @group cluster
     * @return void
     * @requiresRedisVersion >= 3.0.0
     */
    public function testExecutesCASTransactionAgainstCluster(): void
    {
        $redis = $this->getClient();
        $options = ['cas' => true, 'watch' => ['{foo}foo', '{foo}bar', '{foo}baz']];

        $response = $redis->transaction($options, static function (MultiExec $tx) {
            $tx->multi();
            $tx->set('{foo}foo', 'value');
            $tx->set('{foo}bar', 'value');
            $tx->set('{foo}baz', 'value');
        });

        $this->assertEquals(['OK', 'OK', 'OK'], $response);
    }

    /**
     * @group connected
     * @group cluster
     * @return void
     * @requiresRedisVersion >= 3.0.0
     */
    public function testUNWATCHCurrentlyWATCHedKeys(): void
    {
        $redis = $this->getClient();
        $options = ['cas' => true, 'watch' => ['{foo}foo', '{foo}bar', '{foo}baz']];

        $response = $redis->transaction($options, static function (MultiExec $tx) {
            $tx->multi();
            $tx->set('{foo}foo', 'value');
            $tx->set('{foo}bar', 'value');
            $tx->set('{foo}baz', 'value');
            $tx->unwatch();
        });

        $this->assertEquals(['OK', 'OK', 'OK', 'OK'], $response);
    }

    /**
     * @group connected
     * @group cluster
     * @return void
     * @requiresRedisVersion >= 3.0.0
     */
    public function testThrowsExceptionOnWATCHedKeysPointsToDifferentSlots(): void
    {
        $redis = $this->getClient();
        $options = ['cas' => true, 'watch' => ['foo_bar_baz', '{foo}bar', '{foo}baz']];

        $this->expectException(TransactionException::class);
        $this->expectExceptionMessage('WATCHed keys should point to the same hash slot');

        $redis->transaction($options, static function (MultiExec $tx) {
            $tx->multi();
            $tx->set('{foo}foo', 'value');
            $tx->set('{foo}bar', 'value');
            $tx->set('{foo}baz', 'value');
        });
    }

    /**
     * @group connected
     * @group cluster
     * @return void
     * @requiresRedisVersion >= 3.0.0
     */
    public function testThrowsExceptionOnTransactionContextPointsToDifferentSlots(): void
    {
        $redis = $this->getClient();
        $options = ['cas' => true, 'watch' => ['{foo}foo', '{foo}bar', '{foo}baz']];

        $this->expectException(AbortedMultiExecException::class);
        $this->expectExceptionMessage(
            'To be able to execute a transaction against cluster, all commands should operate on the same hash slot'
        );

        $redis->transaction($options, static function (MultiExec $tx) {
            $tx->multi();
            $tx->set('{foo}foo', 'value');
            $tx->set('{foo}bar', 'value');
            $tx->set('foo_bar_baz', 'value');
        });
    }

    // ******************************************************************** //
    // ---- HELPER METHODS ------------------------------------------------ //
    // ******************************************************************** //

    /**
     * Returns a mocked instance of Predis\Connection\NodeConnectionInterface
     * using the specified callback to return values from executeCommand().
     *
     * @param callable $executeCallback
     *
     * @return NodeConnectionInterface|MockObject
     */
    protected function getMockedConnection(callable $executeCallback)
    {
        $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
        $connection
            ->expects($this->any())
            ->method('executeCommand')
            ->willReturnCallback($executeCallback);

        $connection
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

        return $connection;
    }

    /**
     * Returns a mocked instance of Predis\Transaction\MultiExec using
     * the specified callback to return values from the executeCommand method
     * of the underlying connection.
     *
     * @param callable $executeCallback
     * @param array    $txOpts
     * @param array    $clientOpts
     *
     * @return MultiExec
     */
    protected function getMockedTransaction($executeCallback, $txOpts = null, $clientOpts = null): MultiExec
    {
        $connection = $this->getMockedConnection($executeCallback);

        $connection
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));

        $client = new Client($connection, $clientOpts ?: []);
        $transaction = new MultiExec($client, $txOpts ?: []);

        return $transaction;
    }

    /**
     * Returns an instance of Predis\Transaction\MultiExec running against a
     * mocked cluster, using the specified callback to return values from the
     * executeCommand method of the node connection holding the transaction.
     *
     * @param callable $executeCallback
     * @param array    $txOpts
     *
     * @return MultiExec
     */
    protected function getMockedClusterTransaction($executeCallback, $txOpts = null): MultiExec
    {
        $cluster = $this->getMockBuilder(ClusterInterface::class)->getMock();
        $cluster
            ->method('getClusterStrategy')
            ->willReturn(new RedisStrategy());
        $connection = $this->getMockedConnection($executeCallback);

        $cluster
            ->method('getConnectionByCommand')
            ->willReturn($connection);
        $cluster
            ->method('getParameters')
            ->willReturn(new Parameters(['protocol' => 2]));
        $cluster
            ->method('executeCommand')
            ->willReturnCallback(static function (CommandInterface $command) use ($connection) {
                return $connection->executeCommand($command);
            });

        return new MultiExec(new Client($cluster), $txOpts ?: []);
    }

    /**
     * Returns a callback emulating a server-side MULTI/EXEC context.
     *
     * @param ?array $expected List of expected responses
     * @param ?array $commands Reference to an array storing the whole flow of commands
     * @param ?array $cas      Reference to an array storing CAS operations performed by the transaction
     *
     * @return callable
     */
    protected function getExecuteCallback(
        ?array $expected = [],
        ?array &$commands = [],
        ?array &$cas = []
    ): callable {
        $multi = $watch = $abort = false;

        return static function (CommandInterface $command) use (&$expected, &$commands, &$cas, &$multi, &$watch, &$abort) {
            $cmd = $command->getId();

            if ($multi || $cmd === 'MULTI') {
                $commands[] = $command;
            } else {
                $cas[] = $command;
            }

            switch ($cmd) {
                case 'WATCH':
                    if ($multi) {
                        return new Response\Error("ERR $cmd inside MULTI is not allowed");
                    }

                    return $watch = true;

                case 'MULTI':
                    if ($multi) {
                        return new Response\Error('ERR MULTI calls can not be nested');
                    }

                    return $multi = true;

                case 'EXEC':
                    if (!$multi) {
                        return new Response\Error("ERR $cmd without MULTI");
                    }

                    $watch = $multi = false;

                    if ($abort) {
                        $commands = $cas = [];
                        $abort = false;

                        return;
                    }

                    return $expected;

                case 'DISCARD':
                    if (!$multi) {
                        return new Response\Error("ERR $cmd without MULTI");
                    }

                    $watch = $multi = false;

                    return true;

                case 'ECHO':
                    @[$trigger] = $command->getArguments();
                    if (strpos($trigger, 'ERR ') === 0) {
                        throw new Response\ServerException($trigger);
                    }

                    if ($trigger === '!!ABORT!!' && $multi) {
                        $abort = true;
                    }

                    return new Response\Status('QUEUED');

                case 'UNWATCH':
                    $watch = false;
                    // no break

                default:
                    return $multi ? new Response\Status('QUEUED') : 'DUMMY_RESPONSE';
            }
        };
    }

    /**
     * Converts an array of command instances to an array of command IDs.
     *
     * @param CommandInterface[] $commands List of commands instances
     *
     * @return array
     */
    protected static function commandsToIDs(array $commands): array
    {
        return array_map(static function ($cmd) { return $cmd->getId(); }, $commands);
    }

    /**
     * Returns a client instance connected to the specified Redis
     * server instance to perform integration tests.
     *
     * @param array $parameters Additional connection parameters
     * @param array $options    Additional client options
     *
     * @return ClientInterface
     */
    protected function getClient(array $parameters = [], array $options = []): ClientInterface
    {
        return $this->createClient($parameters, $options);
    }
}