PoolingShardManager.php 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. <?php
  2. namespace Doctrine\DBAL\Sharding;
  3. use Doctrine\DBAL\Sharding\ShardChoser\ShardChoser;
  4. use RuntimeException;
  5. /**
  6. * Shard Manager for the Connection Pooling Shard Strategy
  7. */
  8. class PoolingShardManager implements ShardManager
  9. {
  10. /** @var PoolingShardConnection */
  11. private $conn;
  12. /** @var ShardChoser */
  13. private $choser;
  14. /** @var string|null */
  15. private $currentDistributionValue;
  16. public function __construct(PoolingShardConnection $conn)
  17. {
  18. $params = $conn->getParams();
  19. $this->conn = $conn;
  20. $this->choser = $params['shardChoser'];
  21. }
  22. /**
  23. * {@inheritDoc}
  24. */
  25. public function selectGlobal()
  26. {
  27. $this->conn->connect(0);
  28. $this->currentDistributionValue = null;
  29. }
  30. /**
  31. * {@inheritDoc}
  32. */
  33. public function selectShard($distributionValue)
  34. {
  35. $shardId = $this->choser->pickShard($distributionValue, $this->conn);
  36. $this->conn->connect($shardId);
  37. $this->currentDistributionValue = $distributionValue;
  38. }
  39. /**
  40. * {@inheritDoc}
  41. */
  42. public function getCurrentDistributionValue()
  43. {
  44. return $this->currentDistributionValue;
  45. }
  46. /**
  47. * {@inheritDoc}
  48. */
  49. public function getShards()
  50. {
  51. $params = $this->conn->getParams();
  52. $shards = [];
  53. foreach ($params['shards'] as $shard) {
  54. $shards[] = ['id' => $shard['id']];
  55. }
  56. return $shards;
  57. }
  58. /**
  59. * {@inheritDoc}
  60. *
  61. * @throws RuntimeException
  62. */
  63. public function queryAll($sql, array $params, array $types)
  64. {
  65. $shards = $this->getShards();
  66. if (! $shards) {
  67. throw new RuntimeException('No shards found.');
  68. }
  69. $result = [];
  70. $oldDistribution = $this->getCurrentDistributionValue();
  71. foreach ($shards as $shard) {
  72. $this->conn->connect($shard['id']);
  73. foreach ($this->conn->fetchAll($sql, $params, $types) as $row) {
  74. $result[] = $row;
  75. }
  76. }
  77. if ($oldDistribution === null) {
  78. $this->selectGlobal();
  79. } else {
  80. $this->selectShard($oldDistribution);
  81. }
  82. return $result;
  83. }
  84. }