Skip to content

Commit b429025

Browse files
transistivetransistive
authored
Merge pull request #77 from neo4j-php/network-optimisation
Network optimisation
2 parents 1ebc585 + e3a27e9 commit b429025

27 files changed

+410
-216
lines changed

‎src/Authentication/BasicAuth.php‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,4 +46,9 @@ public function authenticateBolt(Bolt $bolt, UriInterface $uri, string $userAgen
4646
{
4747
$bolt->init($userAgent, $this->username, $this->password);
4848
}
49+
50+
public function extractFromUri(UriInterface $uri): AuthenticateInterface
51+
{
52+
return $this;
53+
}
4954
}

‎src/Authentication/KerberosAuth.php‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,4 +38,9 @@ public function authenticateBolt(Bolt $bolt, UriInterface $uri, string $userAgen
3838
$bolt->setScheme('kerberos');
3939
$bolt->init($userAgent, $this->token, $this->token);
4040
}
41+
42+
public function extractFromUri(UriInterface $uri): AuthenticateInterface
43+
{
44+
return $this;
45+
}
4146
}

‎src/Authentication/NoAuth.php‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,4 +30,9 @@ public function authenticateBolt(Bolt $bolt, UriInterface $uri, string $userAgen
3030
$bolt->setScheme('none');
3131
$bolt->init($userAgent, '', '');
3232
}
33+
34+
public function extractFromUri(UriInterface $uri): AuthenticateInterface
35+
{
36+
return $this;
37+
}
3338
}

‎src/Authentication/UrlAuth.php‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -35,12 +35,18 @@ public function authenticateHttp(RequestInterface $request, UriInterface $uri, s
3535
}
3636

3737
public function authenticateBolt(Bolt $bolt, UriInterface $uri, string $userAgent): void
38+
{
39+
$this->extractFromUri($uri)->authenticateBolt($bolt, $uri, $userAgent);
40+
}
41+
42+
public function extractFromUri(UriInterface $uri): AuthenticateInterface
3843
{
3944
if (substr_count($uri->getUserInfo(), ':') === 1) {
4045
[$user, $pass] = explode(':', $uri->getUserInfo());
41-
Authenticate::basic($user, $pass)->authenticateBolt($bolt, $uri, $userAgent);
42-
} else {
43-
Authenticate::disabled()->authenticateBolt($bolt, $uri, $userAgent);
46+
47+
return Authenticate::basic($user, $pass);
4448
}
49+
50+
return Authenticate::disabled();
4551
}
4652
}

‎src/Bolt/BoltConnectionPool.php‎

Lines changed: 85 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,23 +13,31 @@
1313

1414
namespace Laudis\Neo4j\Bolt;
1515

16+
use function array_flip;
1617
use Bolt\Bolt;
1718
use Bolt\connection\StreamSocket;
1819
use Exception;
1920
use function explode;
20-
use Laudis\Neo4j\Common\TransactionHelper;
21+
use const FILTER_VALIDATE_IP;
22+
use function filter_var;
23+
use Laudis\Neo4j\Common\BoltConnection;
2124
use Laudis\Neo4j\Contracts\AuthenticateInterface;
2225
use Laudis\Neo4j\Contracts\ConnectionInterface;
2326
use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
27+
use Laudis\Neo4j\Databags\DatabaseInfo;
2428
use Laudis\Neo4j\Databags\SessionConfiguration;
29+
use Laudis\Neo4j\Enum\ConnectionProtocol;
30+
use Laudis\Neo4j\Neo4j\RoutingTable;
2531
use Psr\Http\Message\UriInterface;
26-
use function str_starts_with;
2732

2833
/**
2934
* @implements ConnectionPoolInterface<Bolt>
3035
*/
3136
final class BoltConnectionPool implements ConnectionPoolInterface
3237
{
38+
/** @var array<string, list<ConnectionInterface<Bolt>>> */
39+
private static array $connectionCache = [];
40+
3341
/**
3442
* @throws Exception
3543
*/
@@ -38,24 +46,92 @@ public function acquire(
3846
AuthenticateInterface $authenticate,
3947
float $socketTimeout,
4048
string $userAgent,
41-
SessionConfiguration $config
49+
SessionConfiguration $config,
50+
?RoutingTable $table = null,
51+
?UriInterface $server = null
4252
): ConnectionInterface {
43-
$host = $uri->getHost();
44-
$socket = new StreamSocket($host, $uri->getPort() ?? 7687, $socketTimeout);
53+
$connectingTo = $server ?? $uri;
54+
$key = $connectingTo->getHost().':'.($connectingTo->getPort() ?? '7687');
55+
if (!isset($this->connectionCache[$key])) {
56+
self::$connectionCache[$key] = [];
57+
}
58+
59+
foreach (self::$connectionCache[$key] as $connection) {
60+
if ($connection->isOpen()) {
61+
return $connection;
62+
}
63+
}
64+
65+
$socket = new StreamSocket($connectingTo->getHost(), $connectingTo->getPort() ?? 7687, $socketTimeout);
66+
67+
$this->configureSsl($uri, $connectingTo, $socket, $table);
68+
69+
$bolt = new Bolt($socket);
70+
$authenticate->authenticateBolt($bolt, $connectingTo, $userAgent);
71+
72+
/**
73+
* @var array{'name': 0, 'version': 1, 'edition': 2}
74+
* @psalm-suppress all
75+
*/
76+
$fields = array_flip($bolt->run(<<<'CYPHER'
77+
CALL dbms.components()
78+
YIELD name, versions, edition
79+
UNWIND versions AS version
80+
RETURN name, version, edition
81+
CYPHER)['fields']);
4582

46-
$this->configureSsl($uri, $host, $socket);
83+
/** @var array{0: array{0: string, 1: string, 2: string}} $results */
84+
$results = $bolt->pullAll();
4785

48-
return TransactionHelper::connectionFromSocket($socket, $uri, $userAgent, $authenticate, $config);
86+
$connection = new BoltConnection(
87+
$bolt,
88+
$socket,
89+
$results[0][$fields['name']].'-'.$results[0][$fields['edition']].'/'.$results[0][$fields['version']],
90+
$connectingTo,
91+
$results[0][$fields['version']],
92+
ConnectionProtocol::determineBoltVersion($bolt),
93+
$config->getAccessMode(),
94+
new DatabaseInfo($config->getDatabase())
95+
);
96+
97+
self::$connectionCache[$key][] = $connection;
98+
99+
return $connection;
49100
}
50101

51-
private function configureSsl(UriInterface $uri, string $host, StreamSocket $socket): void
102+
private function configureSsl(UriInterface $uri, UriInterface $server, StreamSocket $socket, ?RoutingTable $table): void
52103
{
53104
$scheme = $uri->getScheme();
54105
$explosion = explode('+', $scheme, 2);
55106
$sslConfig = $explosion[1] ?? '';
56107

57108
if (str_starts_with('s', $sslConfig)) {
58-
TransactionHelper::enableSsl($host, $sslConfig, $socket);
109+
// We have to pass a different host when working with ssl on aura.
110+
// There is a strange behaviour where if we pass the uri host on a single
111+
// instance aura deployment, we need to pass the original uri for the
112+
// ssl configuration to be valid.
113+
if ($table && $table->getWithRole()->count() > 1) {
114+
$this->enableSsl($server->getHost(), $sslConfig, $socket);
115+
} else {
116+
$this->enableSsl($uri->getHost(), $sslConfig, $socket);
117+
}
118+
}
119+
}
120+
121+
private function enableSsl(string $host, string $sslConfig, StreamSocket $sock): void
122+
{
123+
$options = [
124+
'verify_peer' => true,
125+
'peer_name' => $host,
126+
];
127+
if (!filter_var($host, FILTER_VALIDATE_IP)) {
128+
$options['SNI_enabled'] = true;
129+
}
130+
if ($sslConfig === 's') {
131+
$sock->setSslContextOptions($options);
132+
} elseif ($sslConfig === 'ssc') {
133+
$options['allow_self_signed'] = true;
134+
$sock->setSslContextOptions($options);
59135
}
60136
}
61137
}

‎src/Bolt/BoltUnmanagedTransaction.php‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@
1515

1616
use Bolt\Bolt;
1717
use Bolt\error\MessageException;
18-
use Ds\Vector;
1918
use Exception;
2019
use Laudis\Neo4j\Common\TransactionHelper;
2120
use Laudis\Neo4j\Contracts\ConnectionInterface;
@@ -38,6 +37,7 @@
3837
*/
3938
final class BoltUnmanagedTransaction implements UnmanagedTransactionInterface
4039
{
40+
/** @var FormatterInterface<T> */
4141
private FormatterInterface $formatter;
4242
/** @var ConnectionInterface<Bolt> */
4343
private ConnectionInterface $connection;
@@ -60,15 +60,15 @@ public function commit(iterable $statements = []): CypherList
6060
$tbr = $this->runStatements($statements);
6161

6262
if ($this->finished) {
63-
throw new Neo4jException(new Vector([new Neo4jError('0', 'Transaction already finished')]));
63+
throw new Neo4jException([new Neo4jError('0', 'Transaction already finished')]);
6464
}
6565

6666
try {
6767
$this->getBolt()->commit();
6868
$this->finished = true;
6969
} catch (Exception $e) {
7070
$code = TransactionHelper::extractCode($e);
71-
throw new Neo4jException(new Vector([new Neo4jError($code ?? '', $e->getMessage())]), $e);
71+
throw new Neo4jException([new Neo4jError($code ?? '', $e->getMessage())], $e);
7272
}
7373

7474
return $tbr;
@@ -77,15 +77,15 @@ public function commit(iterable $statements = []): CypherList
7777
public function rollback(): void
7878
{
7979
if ($this->finished) {
80-
throw new Neo4jException(new Vector([new Neo4jError('0', 'Transaction already finished')]));
80+
throw new Neo4jException([new Neo4jError('0', 'Transaction already finished')]);
8181
}
8282

8383
try {
8484
$this->connection->getImplementation()->rollback();
8585
$this->finished = true;
8686
} catch (Exception $e) {
8787
$code = TransactionHelper::extractCode($e) ?? '';
88-
throw new Neo4jException(new Vector([new Neo4jError($code, $e->getMessage())]), $e);
88+
throw new Neo4jException([new Neo4jError($code, $e->getMessage())], $e);
8989
}
9090
}
9191

@@ -110,8 +110,8 @@ public function runStatement(Statement $statement)
110110
*/
111111
public function runStatements(iterable $statements): CypherList
112112
{
113-
/** @var Vector<T> $tbr */
114-
$tbr = new Vector();
113+
/** @var list<T> $tbr */
114+
$tbr = [];
115115
foreach ($statements as $statement) {
116116
$extra = ['db' => $this->database];
117117
$parameters = ParameterHelper::formatParameters($statement->getParameters());
@@ -126,18 +126,18 @@ public function runStatements(iterable $statements): CypherList
126126
} catch (Throwable $e) {
127127
if ($e instanceof MessageException) {
128128
$code = TransactionHelper::extractCode($e) ?? '';
129-
throw new Neo4jException(new Vector([new Neo4jError($code, $e->getMessage())]), $e);
129+
throw new Neo4jException([new Neo4jError($code, $e->getMessage())], $e);
130130
}
131131
throw $e;
132132
}
133-
$tbr->push($this->formatter->formatBoltResult(
133+
$tbr[] = $this->formatter->formatBoltResult(
134134
$meta,
135135
$results,
136136
$this->connection,
137137
$run - $start,
138138
$end - $start,
139139
$statement
140-
));
140+
);
141141
}
142142

143143
return new CypherList($tbr);
@@ -147,4 +147,9 @@ private function getBolt(): Bolt
147147
{
148148
return $this->connection->getImplementation();
149149
}
150+
151+
public function __destruct()
152+
{
153+
$this->connection->close();
154+
}
150155
}

‎src/Bolt/Session.php‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414
namespace Laudis\Neo4j\Bolt;
1515

1616
use Bolt\Bolt;
17-
use Ds\Vector;
1817
use Exception;
1918
use Laudis\Neo4j\Common\TransactionHelper;
2019
use Laudis\Neo4j\Contracts\AuthenticateInterface;
@@ -134,11 +133,10 @@ public function beginTransaction(?iterable $statements = null, ?TransactionConfi
134133
*/
135134
private function beginInstantTransaction(SessionConfiguration $config): TransactionInterface
136135
{
137-
return new BoltUnmanagedTransaction(
138-
$this->config->getDatabase(),
139-
$this->formatter,
140-
$this->acquireConnection(TransactionConfiguration::default(), $config)
141-
);
136+
$connection = $this->acquireConnection(TransactionConfiguration::default(), $config);
137+
$connection->open();
138+
139+
return new BoltUnmanagedTransaction($this->config->getDatabase(), $this->formatter, $connection);
142140
}
143141

144142
/**
@@ -161,14 +159,14 @@ private function startTransaction(TransactionConfiguration $config, SessionConfi
161159
$begin = $bolt->getImplementation()->begin(['db' => $this->config->getDatabase()]);
162160

163161
if (!$begin) {
164-
throw new Neo4jException(new Vector([new Neo4jError('', 'Cannot open new transaction')]));
162+
throw new Neo4jException([new Neo4jError('', 'Cannot open new transaction')]);
165163
}
166164
} catch (Exception $e) {
167165
if ($e instanceof Neo4jException) {
168166
throw $e;
169167
}
170168
$code = TransactionHelper::extractCode($e) ?? '';
171-
throw new Neo4jException(new Vector([new Neo4jError($code, $e->getMessage())]), $e);
169+
throw new Neo4jException([new Neo4jError($code, $e->getMessage())], $e);
172170
}
173171

174172
return new BoltUnmanagedTransaction($this->config->getDatabase(), $this->formatter, $bolt);

0 commit comments

Comments
 (0)