Skip to content

Commit 820ed34

Browse files
realFlowControlrealFlowControl
authored
Merge branch 'master' into patch-1
2 parents 12d0a7e + f0ada36 commit 820ed34

File tree

7 files changed

+295
-100
lines changed

7 files changed

+295
-100
lines changed

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
.*
22
*.lock
33
vendor
4+
phpunit.xml

‎src/Connection.php‎

Lines changed: 115 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -3,29 +3,35 @@
33
namespace React\MySQL;
44

55
use React\EventLoop\LoopInterface;
6-
use React\Stream\Stream;
6+
use React\Socket\ConnectionInterface as SocketConnectionInterface;
77
use React\Socket\Connector;
88
use React\Socket\ConnectorInterface;
99
use React\MySQL\Commands\AuthenticateCommand;
1010
use React\MySQL\Commands\PingCommand;
1111
use React\MySQL\Commands\QueryCommand;
1212
use React\MySQL\Commands\QuitCommand;
1313

14-
class Connection extends EventEmitter
14+
/**
15+
* Class Connection
16+
*
17+
* @package React\MySQL
18+
*/
19+
class Connection extends EventEmitter implements ConnectionInterface
1520
{
16-
const STATE_INIT = 0;
17-
const STATE_CONNECT_FAILED = 1;
18-
const STATE_AUTHENTICATE_FAILED = 2;
19-
const STATE_CONNECTING = 3;
20-
const STATE_CONNECTED = 4;
21-
const STATE_AUTHENTICATED = 5;
22-
const STATE_CLOSEING = 6;
23-
const STATE_CLOSED = 7;
2421

22+
/**
23+
* @var LoopInterface
24+
*/
2525
private $loop;
2626

27+
/**
28+
* @var Connector
29+
*/
2730
private $connector;
2831

32+
/**
33+
* @var array
34+
*/
2935
private $options = [
3036
'host' => '127.0.0.1',
3137
'port' => 3306,
@@ -34,20 +40,38 @@ class Connection extends EventEmitter
3440
'dbname' => '',
3541
];
3642

43+
/**
44+
* @var array
45+
*/
3746
private $serverOptions;
3847

48+
/**
49+
* @var Executor
50+
*/
3951
private $executor;
4052

53+
/**
54+
* @var integer
55+
*/
4156
private $state = self::STATE_INIT;
4257

58+
/**
59+
* @var SocketConnectionInterface
60+
*/
4361
private $stream;
4462

45-
private $buffer;
4663
/**
4764
* @var Protocal\Parser
4865
*/
4966
public $parser;
5067

68+
/**
69+
* Connection constructor.
70+
*
71+
* @param LoopInterface $loop ReactPHP event loop instance.
72+
* @param array $connectOptions MySQL connection options.
73+
* @param ConnectorInterface $connector (optional) socket sonnector instance.
74+
*/
5175
public function __construct(LoopInterface $loop, array $connectOptions = array(), ConnectorInterface $connector = null)
5276
{
5377
$this->loop = $loop;
@@ -62,38 +86,26 @@ public function __construct(LoopInterface $loop, array $connectOptions = array()
6286
}
6387

6488
/**
65-
* Do a async query.
66-
*
67-
* @param string $sql
68-
* @param mixed ...
69-
* @param callable $callback
70-
* @return \React\MySQL\Command|NULL
89+
* {@inheritdoc}
7190
*/
72-
public function query()
91+
public function query($sql, $callback = null, $params = null)
7392
{
74-
$numArgs = func_num_args();
75-
76-
if ($numArgs === 0) {
77-
throw new \InvalidArgumentException('Required at least 1 argument');
78-
}
79-
80-
$args = func_get_args();
81-
$query = new Query(array_shift($args));
82-
83-
$callback = array_pop($args);
93+
$query = new Query($sql);
8494

8595
$command = new QueryCommand($this);
8696
$command->setQuery($query);
8797

98+
$args = func_get_args();
99+
array_shift($args); // Remove $sql parameter.
100+
88101
if (!is_callable($callback)) {
89-
if ($numArgs > 1) {
90-
$args[] = $callback;
91-
}
92102
$query->bindParamsFromArray($args);
93103

94104
return $this->_doCommand($command);
95105
}
96106

107+
array_shift($args); // Remove $callback
108+
97109
$query->bindParamsFromArray($args);
98110
$this->_doCommand($command);
99111

@@ -106,8 +118,13 @@ public function query()
106118
$command->on('success', function ($command) use ($callback) {
107119
$callback($command, $this);
108120
});
121+
122+
return null;
109123
}
110124

125+
/**
126+
* {@inheritdoc}
127+
*/
111128
public function ping($callback)
112129
{
113130
if (!is_callable($callback)) {
@@ -122,22 +139,34 @@ public function ping($callback)
122139
});
123140
}
124141

142+
/**
143+
* {@inheritdoc}
144+
*/
125145
public function selectDb($dbname)
126146
{
127-
return $this->query(sprinf('USE `%s`', $dbname));
147+
return $this->query(sprintf('USE `%s`', $dbname));
128148
}
129149

150+
/**
151+
* {@inheritdoc}
152+
*/
130153
public function listFields()
131154
{
132155
}
133156

157+
/**
158+
* {@inheritdoc}
159+
*/
134160
public function setOption($name, $value)
135161
{
136162
$this->options[$name] = $value;
137163

138164
return $this;
139165
}
140166

167+
/**
168+
* {@inheritdoc}
169+
*/
141170
public function getOption($name, $default = null)
142171
{
143172
if (isset($this->options[$name])) {
@@ -147,13 +176,16 @@ public function getOption($name, $default = null)
147176
return $default;
148177
}
149178

179+
/**
180+
* {@inheritdoc}
181+
*/
150182
public function getState()
151183
{
152184
return $this->state;
153185
}
154186

155187
/**
156-
* Close the connection.
188+
* {@inheritdoc}
157189
*/
158190
public function close($callback = null)
159191
{
@@ -170,60 +202,69 @@ public function close($callback = null)
170202
}
171203

172204
/**
173-
* Connnect to mysql server.
174-
*
175-
* @param callable $callback
176-
*
177-
* @throws \Exception
205+
* {@inheritdoc}
178206
*/
179-
public function connect()
207+
public function connect($callback)
180208
{
181209
$this->state = self::STATE_CONNECTING;
182210
$options = $this->options;
183211
$streamRef = $this->stream;
184-
$args = func_get_args();
185212

186-
if (count($args) > 0) {
187-
$errorHandler = function ($reason) use ($args) {
188-
$this->state = self::STATE_AUTHENTICATE_FAILED;
189-
$args[0]($reason, $this);
190-
};
191-
$connectedHandler = function ($serverOptions) use ($args) {
192-
$this->state = self::STATE_AUTHENTICATED;
193-
$this->serverOptions = $serverOptions;
194-
$args[0](null, $this);
195-
};
213+
$errorHandler = function ($reason) use ($callback) {
214+
$this->state = self::STATE_AUTHENTICATE_FAILED;
215+
$callback($reason, $this);
216+
};
217+
$connectedHandler = function ($serverOptions) use ($callback) {
218+
$this->state = self::STATE_AUTHENTICATED;
219+
$this->serverOptions = $serverOptions;
220+
$callback(null, $this);
221+
};
196222

197-
$this->connector
198-
->connect($this->options['host'] . ':' . $this->options['port'])
199-
->then(function ($stream) use (&$streamRef, $options, $errorHandler, $connectedHandler) {
200-
$streamRef = $stream;
223+
$this->connector
224+
->connect($this->options['host'] . ':' . $this->options['port'])
225+
->then(function ($stream) use (&$streamRef, $options, $errorHandler, $connectedHandler) {
226+
$streamRef = $stream;
201227

202-
$stream->on('error', [$this, 'handleConnectionError']);
203-
$stream->on('close', [$this, 'handleConnectionClosed']);
228+
$stream->on('error', [$this, 'handleConnectionError']);
229+
$stream->on('close', [$this, 'handleConnectionClosed']);
204230

205-
$parser = $this->parser = new Protocal\Parser($stream, $this->executor);
231+
$parser = $this->parser = new Protocal\Parser($stream, $this->executor);
206232

207-
$parser->setOptions($options);
233+
$parser->setOptions($options);
208234

209-
$command = $this->_doCommand(new AuthenticateCommand($this));
210-
$command->on('authenticated', $connectedHandler);
211-
$command->on('error', $errorHandler);
235+
$command = $this->_doCommand(new AuthenticateCommand($this));
236+
$command->on('authenticated', $connectedHandler);
237+
$command->on('error', $errorHandler);
212238

213-
//$parser->on('close', $closeHandler);
214-
$parser->start();
239+
//$parser->on('close', $closeHandler);
240+
$parser->start();
215241

216-
}, [$this, 'handleConnectionError']);
217-
} else {
218-
throw new \Exception('Not Implemented');
219-
}
242+
}, [$this, 'handleConnectionError']);
243+
}
244+
245+
/**
246+
* {@inheritdoc}
247+
*/
248+
public function getServerOptions()
249+
{
250+
return $this->serverOptions;
220251
}
221252

253+
/**
254+
* @param mixed $err Error from socket.
255+
*
256+
* @return void
257+
* @internal
258+
*/
222259
public function handleConnectionError($err)
223260
{
224261
$this->emit('error', [$err, $this]);
225262
}
226263

264+
/**
265+
* @return void
266+
* @internal
267+
*/
227268
public function handleConnectionClosed()
228269
{
229270
if ($this->state < self::STATE_CLOSEING) {
@@ -232,6 +273,13 @@ public function handleConnectionClosed()
232273
}
233274
}
234275

276+
/**
277+
* @param Command $command The command which should be executed.
278+
*
279+
* @return CommandInterface
280+
*
281+
* @throws Exception Cann't send command
282+
*/
235283
protected function _doCommand(Command $command)
236284
{
237285
if ($command->equals(Command::INIT_AUTHENTICATE)) {
@@ -242,9 +290,4 @@ protected function _doCommand(Command $command)
242290
throw new Exception("Cann't send command");
243291
}
244292
}
245-
246-
public function getServerOptions()
247-
{
248-
return $this->serverOptions;
249-
}
250293
}

0 commit comments

Comments
 (0)