mirror of
https://github.com/predis/predis.git
synced 2026-11-08 19:06:36 +00:00
Till Krüss
1766 lines
56 KiB
PHP
1766 lines
56 KiB
PHP
<?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);
|
|
}
|
|
}
|