Skip to content

Commit c041ddd

Browse files
bubba-h57bubba-h57
committed
Resolved
2 parents 9a75ae9 + c49283b commit c041ddd

File tree

4 files changed

+336
-3
lines changed

4 files changed

+336
-3
lines changed

‎config/tunneler.php

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,14 @@
77
'bash_path' => env('TUNNELER_BASH_PATH', 'bash'),
88
'ssh_path' => env('TUNNELER_SSH_PATH', 'ssh'),
99
'nohup_path' => env('TUNNELER_NOHUP_PATH', 'nohup'),
10-
10+
1111
'local_address' => env('TUNNELER_LOCAL_ADDRESS', '127.0.0.1'),
1212
'local_port' => env('TUNNELER_LOCAL_PORT'),
1313
'identity_file' => env('TUNNELER_IDENTITY_FILE'),
14-
14+
1515
'bind_address' => env('TUNNELER_BIND_ADDRESS', '127.0.0.1'),
1616
'bind_port' => env('TUNNELER_BIND_PORT'),
17-
17+
1818
'user' => env('TUNNELER_USER'),
1919
'hostname' => env('TUNNELER_HOSTNAME'),
2020
'port' => env('TUNNELER_PORT'),

‎src/Console/DaemonCommand.php

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
<?php namespace STS\Supervisor\Console;
2+
3+
use Illuminate\Console\Command;
4+
use STS\Supervisor\Daemon\Listener;
5+
use Symfony\Component\Console\Input\InputArgument;
6+
use Symfony\Component\Console\Input\InputOption;
7+
8+
class DaemonCommand extends Command
9+
{
10+
/**
11+
* The console command name.
12+
*
13+
* @var string
14+
*/
15+
protected $name = 'queue:daemon';
16+
17+
/**
18+
* The console command description.
19+
*
20+
* @var string
21+
*/
22+
protected $description = 'Listens to a given queue and gracefully handles PCNTL Signals';
23+
24+
/**
25+
* The queue listener instance.
26+
*
27+
* @var \Illuminate\Queue\Listener
28+
*/
29+
protected $listener;
30+
31+
/**
32+
* DaemonCommand constructor.
33+
*
34+
* @param Listener $listener
35+
*/
36+
public function __construct(Listener $listener)
37+
{
38+
parent::__construct();
39+
40+
$this->listener = $listener;
41+
}
42+
43+
/**
44+
* Execute the console command.
45+
*
46+
* @return void
47+
*/
48+
public function fire()
49+
{
50+
$this->setListenerOptions();
51+
52+
$delay = $this->input->getOption('delay');
53+
54+
// The memory limit is the amount of memory we will allow the script to occupy
55+
// before killing it and letting a process manager restart it for us, which
56+
// is to protect us against any memory leaks that will be in the scripts.
57+
$memory = $this->input->getOption('memory');
58+
59+
$connection = $this->input->getArgument('connection');
60+
61+
$timeout = $this->input->getOption('timeout');
62+
63+
// We need to get the right queue for the connection which is set in the queue
64+
// configuration file for the application. We will pull it based on the set
65+
// connection being run for the queue operation currently being executed.
66+
$queue = $this->getQueue($connection);
67+
68+
$this->listener->listen(
69+
$connection, $queue, $delay, $memory, $timeout
70+
);
71+
}
72+
73+
/**
74+
* Get the name of the queue connection to listen on.
75+
*
76+
* @param string $connection
77+
* @return string
78+
*/
79+
protected function getQueue($connection)
80+
{
81+
if (is_null($connection)) {
82+
$connection = $this->laravel['config']['queue.default'];
83+
}
84+
85+
$queue = $this->laravel['config']->get("queue.connections.{$connection}.queue", 'default');
86+
87+
return $this->input->getOption('queue') ?: $queue;
88+
}
89+
90+
/**
91+
* Set the options on the queue listener.
92+
*
93+
* @return void
94+
*/
95+
protected function setListenerOptions()
96+
{
97+
$this->listener->setEnvironment($this->laravel->environment());
98+
99+
$this->listener->setSleep($this->option('sleep'));
100+
101+
$this->listener->setMaxTries($this->option('tries'));
102+
103+
$this->listener->setOutputHandler(function ($type, $line) {
104+
switch ($type){
105+
case 'warn':
106+
$this->warn($line);
107+
break;
108+
case 'info':
109+
$this->info($line);
110+
break;
111+
case 'out':
112+
default:
113+
$this->output->write($line);
114+
}
115+
116+
});
117+
}
118+
119+
/**
120+
* Get the console command arguments.
121+
*
122+
* @return array
123+
*/
124+
protected function getArguments()
125+
{
126+
return [
127+
['connection', InputArgument::OPTIONAL, 'The name of connection'],
128+
];
129+
}
130+
131+
/**
132+
* Get the console command options.
133+
*
134+
* @return array
135+
*/
136+
protected function getOptions()
137+
{
138+
return [
139+
['queue', null, InputOption::VALUE_OPTIONAL, 'The queue to listen on', null],
140+
141+
['delay', null, InputOption::VALUE_OPTIONAL, 'Amount of time to delay failed jobs', 0],
142+
143+
['memory', null, InputOption::VALUE_OPTIONAL, 'The memory limit in megabytes', 128],
144+
145+
['timeout', null, InputOption::VALUE_OPTIONAL, 'Seconds a job may run before timing out', 60],
146+
147+
['sleep', null, InputOption::VALUE_OPTIONAL, 'Seconds to wait before checking queue for jobs', 3],
148+
149+
['tries', null, InputOption::VALUE_OPTIONAL, 'Number of times to attempt a job before logging it failed', 0],
150+
];
151+
}
152+
}

‎src/Daemon/Listener.php

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
<?php namespace STS\Supervisor\Daemon;
2+
3+
use Illuminate\Queue\Listener as QueueListener;
4+
use Symfony\Component\Process\Process;
5+
6+
pcntl_async_signals(true);
7+
8+
class Listener extends QueueListener{
9+
10+
/**
11+
* Set to true if we catch a signal and need to
12+
* stop after we finish what we are doing.
13+
*
14+
* @var bool
15+
*/
16+
protected $stopGracefully = false;
17+
18+
19+
/**
20+
* Create a new queue listener.
21+
*
22+
* @param string $commandPath
23+
*/
24+
public function __construct($commandPath)
25+
{
26+
if (!function_exists('pcntl_signal')){
27+
throw new \RuntimeException('You must have PCNTL functions to use this Listener');
28+
}
29+
parent::__construct($commandPath);
30+
31+
// Install the signal handler
32+
$this->setSignalHandler(SIGHUP); /* Hangup (POSIX). */
33+
$this->setSignalHandler(SIGINT); /* Interrupt (ANSI). */
34+
$this->setSignalHandler(SIGQUIT); /* Quit (POSIX). */
35+
$this->setSignalHandler(SIGABRT); /* Abort (ANSI). */
36+
$this->setSignalHandler(SIGTERM); /* Termination (ANSI). */
37+
$this->setSignalHandler(SIGTSTP); /* Keyboard stop (POSIX). */
38+
}
39+
40+
/**
41+
* Sets the signal to be handled by either the closure or the built in
42+
* signal handler.
43+
*
44+
* @param int $signal
45+
* @param callable|null $closure
46+
*
47+
* @return bool
48+
*/
49+
public function setSignalHandler(int $signal, callable $closure = null){
50+
if (empty($closure)){
51+
return pcntl_signal($signal, array($this, 'sigHandler'));
52+
}
53+
return pcntl_signal($signal, $closure);
54+
}
55+
56+
/**
57+
* Built in Signal Handler.
58+
* @param int $signo
59+
*/
60+
public function sigHandler( int $signo ){
61+
$this->handleWorkerOutput('warn', sprintf("Signal %d Caught, asking the daemon to stop gracefully.", $signo));
62+
$this->stopGracefully = true;
63+
}
64+
65+
/**
66+
* Log that we are done and exiting.
67+
*/
68+
public function gracefulStop(){
69+
$this->handleWorkerOutput('warn', "Work done, exiting now.");
70+
$this->stop();
71+
}
72+
/**
73+
* Run the given process.
74+
*
75+
* @param Process $process
76+
* @param int $memory
77+
* @return void
78+
*/
79+
public function runProcess(Process $process, $memory)
80+
{
81+
try {
82+
$process->run(function ($type, $line) {
83+
$this->handleWorkerOutput($type, $line);
84+
});
85+
}catch (\Exception $e){
86+
dd($e);
87+
}
88+
89+
// If we caught a signal and need to stop gracefully, this is the place to
90+
// do it.
91+
pcntl_signal_dispatch();
92+
if ($this->stopGracefully){
93+
$this->gracefulStop();
94+
}
95+
// Once we have run the job we'll go check if the memory limit has been
96+
// exceeded for the script. If it has, we will kill this script so a
97+
// process manager will restart this with a clean slate of memory.
98+
if ($this->memoryExceeded($memory)) {
99+
$this->stop();
100+
}
101+
}
102+
103+
/**
104+
* Handle output from the worker process.
105+
*
106+
* @param string $type
107+
* @param string $line
108+
* @return void
109+
*/
110+
protected function handleWorkerOutput($type, $line)
111+
{
112+
if (isset($this->outputHandler)) {
113+
call_user_func($this->outputHandler, $type, $line);
114+
}
115+
}
116+
}

‎src/DaemonServiceProvider.php

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
<?php namespace STS\Supervisor;
2+
3+
use Illuminate\Support\ServiceProvider;
4+
use STS\Supervisor\Console\DaemonCommand;
5+
use STS\Supervisor\Daemon\Listener;
6+
7+
class DaemonServiceProvider extends ServiceProvider {
8+
/**
9+
* Indicates if loading of the provider is deferred.
10+
*
11+
* @var bool
12+
*/
13+
protected $defer = true;
14+
15+
/**
16+
* Register the service provider.
17+
*
18+
* @return void
19+
*/
20+
public function register()
21+
{
22+
23+
$this->registerListener();
24+
}
25+
26+
/**
27+
* Register the queue listener.
28+
*
29+
* @return void
30+
*/
31+
protected function registerListener()
32+
{
33+
$this->registerListenCommand();
34+
35+
$this->app->singleton('queue.daemon.listener', function ($app) {
36+
return new Listener($app->basePath());
37+
});
38+
}
39+
40+
/**
41+
* Register the queue listener console command.
42+
*
43+
* @return void
44+
*/
45+
protected function registerListenCommand()
46+
{
47+
$this->app->singleton('command.queue.daemon', function ($app) {
48+
return new DaemonCommand($app['queue.daemon.listener']);
49+
});
50+
51+
$this->commands('command.queue.daemon');
52+
}
53+
54+
/**
55+
* Get the services provided by the provider.
56+
*
57+
* @return array
58+
*/
59+
public function provides()
60+
{
61+
return [
62+
'queue.supervisor.listener', 'command.queue.daemon'
63+
];
64+
}
65+
}

0 commit comments

Comments
 (0)