11<?php
22
3+ declare (strict_types=1 );
4+
35namespace iamfarhad \LaravelRabbitMQ \Console ;
46
57use Symfony \Component \Console \Attribute \AsCommand ;
68use Illuminate \Queue \Console \WorkCommand ;
79use Illuminate \Support \Str ;
10+ use RuntimeException ;
811
912#[AsCommand(name: 'rabbitmq:consume ' )]
1013final class ConsumeCommand extends WorkCommand
1114{
15+ /**
16+ * The console command signature.
17+ *
18+ * @var string
19+ */
1220 protected $ signature'rabbitmq:consume
1321 {connection? : The name of the queue connection to work}
1422 {--name=default : The name of the consumer}
1523 {--queue= : The names of the queues to work}
1624 {--once : Only process the next job on the queue}
1725 {--stop-when-empty : Stop when the queue is empty}
18- {--delay=0 : The number of seconds to delay failed jobs (Deprecated) }
26+ {--delay=0 : The number of seconds to delay failed jobs}
1927 {--backoff=0 : The number of seconds to wait before retrying a job that encountered an uncaught exception}
2028 {--max-jobs=0 : The number of jobs to process before stopping}
2129 {--max-time=0 : The maximum number of seconds the worker should run}
@@ -25,69 +33,143 @@ final class ConsumeCommand extends WorkCommand
2533 {--timeout=60 : The number of seconds a child process can run}
2634 {--tries=1 : Number of times to attempt a job before logging it failed}
2735 {--rest=0 : Number of seconds to rest between jobs}
28- {--max-priority=}
36+ {--max-priority=null : Maximum priority level to consume }
2937 {--consumer-tag}
3038 {--prefetch-size=0}
3139 {--prefetch-count=1000}
3240 {--num-processes=2 : Number of processes to run in parallel}
3341 '
3442
35- protected $ description'Consume messages ' ;
36-
37- public function handle (): int |null
43+ /**
44+ * The console command description.
45+ *
46+ * @var string
47+ */
48+ protected $ description'Consume messages from RabbitMQ queue ' ;
49+
50+ /**
51+ * Execute the console command.
52+ */
53+ public function handle (): int
3854 {
39- $ numProcesses$ this option ('num-processes ' );
55+ $ numProcessesint ) $ this option ('num-processes ' );
56+
57+ if ($ numProcesses1 ) {
58+ $ this error ('Number of processes must be at least 1 ' );
59+ return 1 ;
60+ }
61+
62+ // Skip forking if only one process is needed
63+ if ($ numProcesses1 ) {
64+ return $ this consume ();
65+ }
66+
67+ // Check if pcntl extension is available
68+ if (!extension_loaded ('pcntl ' )) {
69+ $ this error ('The pcntl extension is required for parallel processing ' );
70+ return 1 ;
71+ }
72+
73+ $ childPids
4074
4175 for ($ i0 ; $ i$ numProcesses$ i
4276 $ pidpcntl_fork ();
4377
4478 if ($ pid1 ) {
45- // Error handling
46- echo "Could not fork process \n" ;
47- exit (1 );
79+ $ this error ("Failed to fork process $ i );
80+ continue ;
4881 }
4982
5083 if ($ pid0 ) {
51- // This is the child process
52- $ this consume ();
53- exit (0 );
84+ // Child process
85+ exit ($ this consume ());
86+ } else {
87+ // Parent process
88+ $ childPids$ pid
89+ $ this info ("Started worker process $ pid );
5490 }
5591 }
5692
93+ // Set up signal handling for graceful termination
94+ if (function_exists ('pcntl_signal ' )) {
95+ pcntl_signal (SIGTERM , function () use (&$ childPids
96+ foreach ($ childPidsas $ pid
97+ if (function_exists ('posix_kill ' )) {
98+ posix_kill ($ pidSIGTERM );
99+ }
100+ }
101+ });
102+ }
103+
57104 // Wait for all child processes to finish
58- // while (pcntl_waitpid(0, $status) !== -1) {
59- // Handle exit status if needed
60- // }
105+ foreach ($ childPidsas $ pid
106+ pcntl_waitpid ($ pid$ status
107+
108+ if (pcntl_wifexited ($ status
109+ $ exitCodepcntl_wexitstatus ($ status
110+ if ($ exitCode0 ) {
111+ $ this warn ("Process $ pid exited with code $ exitCode );
112+ }
113+ } else {
114+ $ this warn ("Process $ pid terminated abnormally " );
115+ }
116+ }
61117
62118 return 0 ;
63119 }
64120
65- private function consume (): void
121+ /**
122+ * Configure and run the consumer.
123+ */
124+ private function consume (): int
66125 {
67- $ consumer$ this worker ;
126+ try {
127+ $ consumer$ this worker ;
128+
129+ if (!$ consumer
130+ throw new RuntimeException ('Worker instance not initialized ' );
131+ }
68132
69- $ consumersetContainer ($ this laravel );
70- $ consumersetName ($ this option ('name ' ));
71- $ consumersetConsumerTag ($ this consumerTag ());
72- $ consumersetMaxPriority ((int ) $ this option ('max-priority ' ));
73- $ consumersetPrefetchSize ((int ) $ this option ('prefetch-size ' ));
74- $ consumersetPrefetchCount ((int ) $ this option ('prefetch-count ' ));
133+ $ consumersetContainer ($ this laravel );
134+ $ consumersetName ($ this option ('name ' ));
135+ $ consumersetConsumerTag ($ this generateConsumerTag ());
75136
76- parent ::handle ();
137+ // Initialize prefetch size and count first
138+ $ consumersetPrefetchSize ((int ) $ this option ('prefetch-size ' ));
139+ $ consumersetPrefetchCount ((int ) $ this option ('prefetch-count ' ));
140+
141+ // Only set max priority if it's provided and not null
142+ $ maxPriority$ this option ('max-priority ' );
143+ if ($ maxPrioritynull && $ maxPriority'' ) {
144+ $ consumersetMaxPriority ((int ) $ maxPriority
145+ }
146+
147+ return parent ::handle () ?? 0 ;
148+ } catch (\Throwable $ e
149+ $ this error ($ egetMessage ());
150+ return 1 ;
151+ }
77152 }
78153
79- private function consumerTag (): string
154+ /**
155+ * Generate a unique consumer tag.
156+ */
157+ private function generateConsumerTag (): string
80158 {
81159 if ($ consumerTag$ this option ('consumer-tag ' )) {
82160 return $ consumerTag
83161 }
84162
163+ $ appNameconfig ('app.name ' , 'laravel ' );
164+ $ consumerName$ this option ('name ' );
165+ $ uniqueIdmd5 (serialize ($ this options ()) . Str::random (16 ) . getmypid ());
166+
85167 $ consumerTagimplode (
86168 '_ ' ,
87169 [
88- Str::slug (config ( ' app.name ' , ' laravel ' ) ),
89- Str::slug ($ this -> option ( ' name ' ) ),
90- md5 ( serialize ( $ this -> options ()) . Str:: random ( 16 ) . getmypid ()) ,
170+ Str::slug ($ appName
171+ Str::slug ($ consumerName
172+ $ uniqueId
91173 ]
92174 );
93175
0 commit comments