Skip to content

Commit cda1332

Browse files
hasanparastehhasanparasteh
authored
Merge pull request #14 from skydiablo/main
react/mysql 0.7 support
2 parents c3b9323 + eca7cde commit cda1332

File tree

2 files changed

+77
-64
lines changed

2 files changed

+77
-64
lines changed

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
],
1515
"require": {
1616
"php": ">=8.0",
17-
"react/mysql": "^0.6",
17+
"react/mysql": "^0.7",
1818
"react/async": "^4 || ^3 || ^2"
1919
},
2020
"autoload": {

‎src/QueryBuilder/Core/DBFactory.php‎

Lines changed: 76 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
use React\EventLoop\LoopInterface;
66
use React\MySQL\Factory;
7+
use React\Mysql\MysqlClient;
78
use React\Promise\PromiseInterface;
89
use React\Stream\ReadableStreamInterface;
910
use Saraf\QB\QueryBuilder\Exceptions\DBFactoryException;
@@ -17,7 +18,6 @@ class DBFactory
1718
private array $logs = [];
1819
private const MAX_CONNECTION_COUNT = 1000000000;
1920

20-
protected Factory $factory;
2121
protected array $writeConnections = [];
2222
protected array $readConnections = [];
2323

@@ -26,31 +26,30 @@ class DBFactory
2626
*/
2727
public function __construct(
2828
protected ?LoopInterface $loop,
29-
protected string $host,
30-
protected string $dbName,
31-
protected string $username,
32-
protected string $password,
33-
protected int $writePort = 6446,
34-
protected int $readPort = 6447,
35-
protected int $writeInstanceCount = 2,
36-
protected int $readInstanceCount = 2,
37-
protected int $timeout = 2,
38-
protected int $idle = 2,
39-
protected string $charset = 'utf8mb4',
40-
protected bool $debugMode = false
41-
)
42-
{
43-
$this->factory = new Factory($loop);
29+
protected string $host,
30+
protected string $dbName,
31+
protected string $username,
32+
protected string $password,
33+
protected int $writePort = 6446,
34+
protected int $readPort = 6447,
35+
protected int $writeInstanceCount = 2,
36+
protected int $readInstanceCount = 2,
37+
protected int $timeout = 2,
38+
protected int $idle = 2,
39+
protected string $charset = 'utf8mb4',
40+
protected bool $debugMode = false,
41+
) {
4442
$this->createConnections();
4543
}
4644

4745
public function getTrace(): array
4846
{
49-
if ($this->debugMode === false)
47+
if ($this->debugMode === false) {
5048
return [
5149
'result' => false,
52-
'error' => 'NOT_IN_DEBUG_MODE'
50+
'error' => 'NOT_IN_DEBUG_MODE',
5351
];
52+
}
5453

5554
$jobs = [];
5655
foreach ($this->writeConnections as $i => $writeConnection) {
@@ -62,9 +61,9 @@ public function getTrace(): array
6261
}
6362

6463
return [
65-
'result' => true,
66-
'logs' => $this->logs,
67-
'workers' => $jobs
64+
'result' => true,
65+
'logs' => $this->logs,
66+
'workers' => $jobs,
6867
];
6968
}
7069

@@ -73,45 +72,45 @@ public function getTrace(): array
7372
*/
7473
protected function createConnections(): static
7574
{
76-
if (count($this->readConnections) > 0 || count($this->writeConnections) > 0)
75+
if (count($this->readConnections) > 0 || count($this->writeConnections) > 0) {
7776
throw new DBFactoryException("Connections Already Created");
77+
}
7878

7979
for ($i = 0; $i < $this->writeInstanceCount; ++$i) {
8080
$this->writeConnections[] = new DBWorker(
81-
$this->factory
82-
->createLazyConnection(
83-
sprintf(
84-
"%s:%s@%s:%s/%s?idle=%s&timeout=%s&charset=%s",
85-
$this->username,
86-
urlencode($this->password),
87-
$this->host,
88-
$this->writePort,
89-
$this->dbName,
90-
$this->idle,
91-
$this->timeout,
92-
$this->charset
93-
)
94-
)
81+
new MysqlClient(
82+
sprintf(
83+
"%s:%s@%s:%s/%s?idle=%s&timeout=%s&charset=%s",
84+
$this->username,
85+
urlencode($this->password),
86+
$this->host,
87+
$this->writePort,
88+
$this->dbName,
89+
$this->idle,
90+
$this->timeout,
91+
$this->charset,
92+
),
93+
),
9594
);
9695
}
9796

9897
for ($s = 0; $s < $this->readInstanceCount; ++$s) {
9998
$this->readConnections[] = new DBWorker(
100-
$this->factory
101-
->createLazyConnection(
102-
sprintf(
103-
"%s:%s@%s:%s/%s?idle=%s&timeout=%s",
104-
$this->username,
105-
urlencode($this->password),
106-
$this->host,
107-
$this->readPort,
108-
$this->dbName,
109-
$this->idle,
110-
$this->timeout
111-
)
112-
)
99+
new MysqlClient(
100+
sprintf(
101+
"%s:%s@%s:%s/%s?idle=%s&timeout=%s",
102+
$this->username,
103+
urlencode($this->password),
104+
$this->host,
105+
$this->readPort,
106+
$this->dbName,
107+
$this->idle,
108+
$this->timeout,
109+
),
110+
),
113111
);
114112
}
113+
115114
return $this;
116115
}
117116

@@ -120,8 +119,9 @@ protected function createConnections(): static
120119
*/
121120
public function getQueryBuilder(): QueryBuilder
122121
{
123-
if (count($this->readConnections) == 0 || count($this->writeConnections) == 0)
122+
if (count($this->readConnections) == 0 || count($this->writeConnections) == 0) {
124123
throw new DBFactoryException("Connections Not Created");
124+
}
125125

126126
return new QueryBuilder($this);
127127
}
@@ -134,29 +134,35 @@ public function query(string $query): PromiseInterface
134134
$isWrite = true;
135135
if (str_starts_with(strtolower($query), "select")
136136
|| str_starts_with(strtolower($query), "show")
137-
) $isWrite = false;
137+
) {
138+
$isWrite = false;
139+
}
138140

139141
$bestConnections = $this->getBestConnection();
140142

141143
$connection = $isWrite
142144
? $this->writeConnections[$bestConnections['write']]
143145
: $this->readConnections[$bestConnections['read']];
144146

145-
if (!($connection instanceof DBWorker))
147+
if (!($connection instanceof DBWorker)) {
146148
throw new DBFactoryException("Connections Not Instance of Worker / Restart App");
149+
}
147150

148-
if (!$this->debugMode)
151+
if (!$this->debugMode) {
149152
return $connection->query($query);
153+
}
150154

151155
$startTime = QBHelper::getCurrentMicroTime();
152-
return $connection->query($query)
156+
157+
return $connection
158+
->query($query)
153159
->then(function ($result) use ($isWrite, $startTime, $query) {
154160
$endTime = QBHelper::getCurrentMicroTime();
155161
$this->logs[] = [
156-
'query' => $query,
157-
'took' => $endTime - $startTime,
162+
'query' => $query,
163+
'took' => $endTime - $startTime,
158164
'isWrite' => $isWrite,
159-
'status' => $result['result']
165+
'status' => $result['result'],
160166
];
161167

162168
return $result;
@@ -171,16 +177,19 @@ public function streamQuery(string $query): StreamEventHandler
171177
$isWrite = true;
172178
if (str_starts_with(strtolower($query), "select")
173179
|| str_starts_with(strtolower($query), "show")
174-
) $isWrite = false;
180+
) {
181+
$isWrite = false;
182+
}
175183

176184
$bestConnections = $this->getBestConnection();
177185

178186
$connection = $isWrite
179187
? $this->writeConnections[$bestConnections['write']]
180188
: $this->readConnections[$bestConnections['read']];
181189

182-
if (!($connection instanceof DBWorker))
190+
if (!($connection instanceof DBWorker)) {
183191
throw new DBFactoryException("Connections Not Instance of Worker / Restart App");
192+
}
184193

185194
return $connection->streamQuery($query);
186195
}
@@ -193,16 +202,19 @@ public function streamQueryRaw(string $query): ReadableStreamInterface
193202
$isWrite = true;
194203
if (str_starts_with(strtolower($query), "select")
195204
|| str_starts_with(strtolower($query), "show")
196-
) $isWrite = false;
205+
) {
206+
$isWrite = false;
207+
}
197208

198209
$bestConnections = $this->getBestConnection();
199210

200211
$connection = $isWrite
201212
? $this->writeConnections[$bestConnections['write']]
202213
: $this->readConnections[$bestConnections['read']];
203214

204-
if (!($connection instanceof DBWorker))
215+
if (!($connection instanceof DBWorker)) {
205216
throw new DBFactoryException("Connections Not Instance of Worker / Restart App");
217+
}
206218

207219
return $connection->streamQueryRaw($query);
208220
}
@@ -212,8 +224,9 @@ public function streamQueryRaw(string $query): ReadableStreamInterface
212224
*/
213225
private function getBestConnection(): array
214226
{
215-
if (count($this->readConnections) == 0 || count($this->writeConnections) == 0)
227+
if (count($this->readConnections) == 0 || count($this->writeConnections) == 0) {
216228
throw new DBFactoryException("Connections Not Created");
229+
}
217230

218231
// Best Writer
219232
$minWriteJobs = self::MAX_CONNECTION_COUNT;
@@ -239,7 +252,7 @@ private function getBestConnection(): array
239252

240253
return [
241254
'write' => $minJobsWriterConnection,
242-
'read' => $minJobsReaderConnection
255+
'read' => $minJobsReaderConnection,
243256
];
244257
}
245258

0 commit comments

Comments
 (0)