The question is very related to Consume only N messages from RabbitMQ with react\stomp, ack them separately and then exit, but a bit more general.
For example, I have a slow I/O operation:
$port = 4000;
$loop = React\EventLoop\Factory::create();
$socket = new React\Socket\Server($loop);
$socket->on('connection', function ($conn) use ($loop){
$conn->on('data', function ($data) use ($conn, $loop) {
if ($data == 42) {
// this instantly stop the loop
$loop->stop();
}
$process = new React\ChildProcess\Process('sleep 5; echo ' . $data);
$loop->addTimer(0.001, function($timer) use ($process, $conn) {
$process->start($timer->getLoop());
$process->stdout->on('data', function($output) use ($conn) {
if ($output) {
$conn->write("> $output");
}
});
});
});
});
echo "Socket server listening on port $port.\n";
echo "You can connect to it by running: telnet localhost $port\n";
$socket->listen($port);
$loop->run();
echo "exited";
when I run $loop->run();
I want to stop it at some point, e.g. by timer, after accepting N requests, or any other event, like pcntl_signal
, or data assertion, as in the example.
The challenge is to finish all started jobs before exit, and I cannot figure out how to achieve it.
In the server console I have:
Socket server listening on port 4000.
You can connect to it by running: telnet localhost 4000
exited
In the client console I have:
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
1
2
3
4
5
> 1
> 2
42
Connection closed by foreign host.
where 1,2,3,4,5 were entered with 1 sec interval
Instead I would like to see something like this:
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
1
2
3
4
5
> 1
> 2
42
> 3
6
> 4
> 5
Connection closed by foreign host.
As was mentioned in the comments, to exit gracefully you need to track running processes and stop the loop only when all processes are finished:
$socket->on('connection', function ($conn) use ($loop) {
$processes = new SplObjectStorage();
$stop = false;
$conn->on('data', function ($data) use ($conn, $loop, $processes, &$stop) {
if ('42' === trim($data)) {
$stop = true;
if (!$processes->count()) {
$loop->stop();
}
}
if ($stop) {
return;
}
$process = new React\ChildProcess\Process('sleep 5; echo ' . $data);
$processes->attach($process);
$process->on('exit', function () use ($process, $processes, &$stop, $loop) {
$processes->detach($process);
if ($stop && !$processes->count()) {
$loop->stop();
}
});
$loop->addTimer(0.001, function($timer) use ($process, $conn) {
$process->start($timer->getLoop());
$process->stdout->on('data', function($output) use ($conn) {
if ($output && '42' !== trim($output)) {
$conn->write("> $output");
}
});
});
});
});
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With