SafeClusterExecutor.php 1.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. <?php
  2. namespace Predis\Pipeline;
  3. use Predis\Network\IConnection;
  4. class SafeClusterExecutor implements IPipelineExecutor {
  5. public function execute(IConnection $connection, &$commands) {
  6. $connectionExceptions = array();
  7. $sizeofPipe = count($commands);
  8. $values = array();
  9. foreach ($commands as $command) {
  10. $cmdConnection = $connection->getConnection($command);
  11. if (isset($connectionExceptions[spl_object_hash($cmdConnection)])) {
  12. continue;
  13. }
  14. try {
  15. $cmdConnection->writeCommand($command);
  16. }
  17. catch (\Predis\CommunicationException $exception) {
  18. $connectionExceptions[spl_object_hash($cmdConnection)] = $exception;
  19. }
  20. }
  21. for ($i = 0; $i < $sizeofPipe; $i++) {
  22. $command = $commands[$i];
  23. unset($commands[$i]);
  24. $cmdConnection = $connection->getConnection($command);
  25. $connectionObjectHash = spl_object_hash($cmdConnection);
  26. if (isset($connectionExceptions[$connectionObjectHash])) {
  27. $values[] = $connectionExceptions[$connectionObjectHash];
  28. continue;
  29. }
  30. try {
  31. $response = $cmdConnection->readResponse($command);
  32. $values[] = ($response instanceof \Iterator
  33. ? iterator_to_array($response)
  34. : $response
  35. );
  36. }
  37. catch (\Predis\ServerException $exception) {
  38. $values[] = $exception->toResponseError();
  39. }
  40. catch (\Predis\CommunicationException $exception) {
  41. $values[] = $exception;
  42. $connectionExceptions[$connectionObjectHash] = $exception;
  43. }
  44. }
  45. return $values;
  46. }
  47. }