1
|
|
|
<?php |
2
|
|
|
|
3
|
|
|
namespace ElasticsearchModule\Service; |
4
|
|
|
|
5
|
|
|
use ArrayObject; |
6
|
|
|
use Elasticsearch\ConnectionPool\AbstractConnectionPool; |
7
|
|
|
use Elasticsearch\Connections\ConnectionFactoryInterface; |
8
|
|
|
use Elasticsearch\Connections\ConnectionInterface; |
9
|
|
|
use Zend\ServiceManager\Exception\InvalidArgumentException; |
10
|
|
|
use Zend\ServiceManager\Exception\ServiceNotCreatedException; |
11
|
|
|
use Zend\ServiceManager\ServiceLocatorInterface; |
12
|
|
|
use Zend\Stdlib\ArrayObject as ZendArrayObject; |
13
|
|
|
|
14
|
|
|
/** |
15
|
|
|
* @author Pedro Alves <[email protected]> |
16
|
|
|
*/ |
17
|
|
|
class ConnectionPoolFactory extends AbstractFactory |
18
|
|
|
{ |
19
|
|
|
use InvalidTypeExceptionTrait; |
20
|
|
|
|
21
|
|
|
/** |
22
|
|
|
* {@inheritDoc} |
23
|
|
|
*/ |
24
|
20 |
|
protected function create(ServiceLocatorInterface $serviceLocator, $config) |
25
|
|
|
{ |
26
|
20 |
|
$className = $config['class']; |
27
|
|
|
|
28
|
20 |
|
if (!in_array(AbstractConnectionPool::class, class_parents($className))) { |
29
|
1 |
|
throw new ServiceNotCreatedException(sprintf( |
30
|
1 |
|
"'%s' does not implement %s", |
31
|
1 |
|
$className, |
32
|
|
|
AbstractConnectionPool::class |
33
|
1 |
|
)); |
34
|
|
|
} |
35
|
19 |
|
$selector = $this->getServiceOrClassObject($serviceLocator, $config['selector']); |
36
|
19 |
|
$connectionFactory = $this->getServiceOrClassObject($serviceLocator, $config['connection_factory']); |
37
|
|
|
|
38
|
19 |
|
if (!$connectionFactory instanceof ConnectionFactoryInterface) { |
39
|
1 |
|
throw $this->getException( |
40
|
1 |
|
'connection factory', |
41
|
1 |
|
ConnectionFactoryInterface::class, |
42
|
1 |
|
ServiceNotCreatedException::class, |
43
|
|
|
$connectionFactory |
44
|
1 |
|
); |
45
|
|
|
} |
46
|
18 |
|
$parameters = isset($config['parameters']) ? $config['parameters'] : []; |
47
|
18 |
|
$connections = $this->getConnectionsFromConfiguration($config, $connectionFactory); |
48
|
11 |
|
return new $className( |
49
|
11 |
|
$connections, |
50
|
11 |
|
$selector, |
51
|
11 |
|
$connectionFactory, |
52
|
|
|
$parameters |
53
|
11 |
|
); |
54
|
|
|
} |
55
|
|
|
|
56
|
|
|
/** |
57
|
|
|
* {@inheritDoc} |
58
|
|
|
*/ |
59
|
20 |
|
public function getServiceType() |
60
|
|
|
{ |
61
|
20 |
|
return 'connection_pool'; |
62
|
|
|
} |
63
|
|
|
|
64
|
|
|
/** |
65
|
|
|
* @param array|ArrayObject|ZendArrayObject $config |
66
|
|
|
* @param ConnectionFactoryInterface $connectionFactory |
67
|
|
|
* @return ConnectionInterface[] |
68
|
|
|
* @throws ServiceNotCreatedException |
69
|
|
|
*/ |
70
|
18 |
|
private function getConnectionsFromConfiguration($config, ConnectionFactoryInterface $connectionFactory) |
71
|
|
|
{ |
72
|
18 |
|
$connections = []; |
73
|
|
|
|
74
|
18 |
|
foreach ($config['hosts'] as $host) { |
75
|
8 |
|
if (!is_string($host) && !is_array($host)) { |
76
|
6 |
|
throw new ServiceNotCreatedException("Could not parse host: " . print_r($host, true)); |
77
|
|
|
} |
78
|
|
|
|
79
|
2 |
|
if (is_string($host)) { |
80
|
1 |
|
$host = $this->extractURIParts($host); |
81
|
1 |
|
} |
82
|
2 |
|
$host = $this->normalizeExtendedHost($host); |
83
|
1 |
|
$connections[] = $connectionFactory->create($host); |
84
|
11 |
|
} |
85
|
11 |
|
return $connections; |
86
|
|
|
} |
87
|
|
|
|
88
|
|
|
/** |
89
|
|
|
* @param $host |
90
|
|
|
* @return array |
91
|
|
|
*/ |
92
|
2 |
|
private function normalizeExtendedHost($host) { |
93
|
2 |
|
if (isset($host['host']) === false) { |
94
|
1 |
|
throw new ServiceNotCreatedException( |
95
|
1 |
|
"Required 'host' was not defined in extended format: " . print_r($host, true) |
96
|
1 |
|
); |
97
|
|
|
} |
98
|
1 |
|
return array_merge(['scheme' => 'http', 'port' => 9200], $host); |
99
|
|
|
} |
100
|
|
|
|
101
|
|
|
/** |
102
|
|
|
* @param string $host |
103
|
|
|
* |
104
|
|
|
* @throws InvalidArgumentException |
105
|
|
|
* @return array |
106
|
|
|
*/ |
107
|
1 |
|
private function extractURIParts($host) |
108
|
|
|
{ |
109
|
1 |
|
$host = filter_var($host, FILTER_VALIDATE_URL, FILTER_FLAG_SCHEME_REQUIRED) ? $host : 'http://' . $host; |
110
|
1 |
|
$parts = parse_url($host); |
111
|
|
|
|
112
|
1 |
|
if ($parts === false) { |
113
|
|
|
throw new InvalidArgumentException("Could not parse URI"); |
114
|
|
|
} |
115
|
1 |
|
return array_merge(['port' => 9200], $parts); |
116
|
|
|
} |
117
|
|
|
|
118
|
|
|
} |
119
|
|
|
|