|
24 | 24 | use CodeIgniter\Queue\Payloads\PayloadMetadata;
|
25 | 25 | use CodeIgniter\Queue\QueuePushResult;
|
26 | 26 | use PhpAmqpLib\Channel\AMQPChannel;
|
27 |
| -use PhpAmqpLib\Connection\AMQPStreamConnection; |
| 27 | +use PhpAmqpLib\Connection\AbstractConnection; |
| 28 | +use PhpAmqpLib\Connection\AMQPConnectionConfig; |
| 29 | +use PhpAmqpLib\Connection\AMQPConnectionFactory; |
28 | 30 | use PhpAmqpLib\Message\AMQPMessage;
|
29 | 31 | use PhpAmqpLib\Wire\AMQPTable;
|
30 | 32 | use Throwable;
|
31 | 33 |
|
32 | 34 | class RabbitMQHandler extends BaseHandler implements QueueInterface
|
33 | 35 | {
|
34 |
| - private readonly AMQPStreamConnection $connection; |
| 36 | + private readonly AbstractConnection $connection; |
35 | 37 | private readonly AMQPChannel $channel;
|
36 | 38 | private array $declaredQueues = [];
|
37 | 39 | private array $declaredExchanges = [];
|
38 | 40 |
|
39 | 41 | public function __construct(protected QueueConfig $config)
|
40 | 42 | {
|
41 | 43 | try {
|
42 |
| - $this->connection = new AMQPStreamConnection( |
43 |
| - $config->rabbitmq['host'], |
44 |
| - $config->rabbitmq['port'], |
45 |
| - $config->rabbitmq['user'], |
46 |
| - $config->rabbitmq['password'], |
47 |
| - $config->rabbitmq['vhost'] ?? '/', |
48 |
| - $config->rabbitmq['insist'] ?? false, |
49 |
| - $config->rabbitmq['loginMethod'] ?? 'AMQPLAIN', |
50 |
| - null, |
51 |
| - $config->rabbitmq['locale'] ?? 'en_US', |
52 |
| - $config->rabbitmq['connectionTimeout'] ?? 3.0, |
53 |
| - $config->rabbitmq['readWriteTimeout'] ?? 3.0, |
54 |
| - null, |
55 |
| - $config->rabbitmq['keepalive'] ?? false, |
56 |
| - $config->rabbitmq['heartbeat'] ?? 0, |
57 |
| - ); |
| 44 | + $amqp = new AMQPConnectionConfig(); |
| 45 | + $amqp->setHost($config->rabbitmq['host']); |
| 46 | + $amqp->setPort($config->rabbitmq['port']); |
| 47 | + $amqp->setUser($config->rabbitmq['user']); |
| 48 | + $amqp->setPassword($config->rabbitmq['password']); |
| 49 | + $amqp->setVhost($config->rabbitmq['vhost'] ?? '/'); |
| 50 | + |
| 51 | + // Enable SSL/TLS |
| 52 | + if ($config->rabbitmq['ssl'] ?? ($config->rabbitmq['port'] === 5671)) { |
| 53 | + $amqp->setIsSecure(true); |
| 54 | + } |
58 | 55 |
|
59 |
| - $this->channel = $this->connection->channel(); |
| 56 | + $this->connection = AMQPConnectionFactory::create($amqp); |
| 57 | + $this->channel = $this->connection->channel(); |
60 | 58 |
|
61 | 59 | // Set QoS for consumer (prefetch limit)
|
62 |
| - $prefetch = $config->rabbitmq['prefetch'] ?? 1; |
63 |
| - $this->channel->basic_qos(0, $prefetch, false); |
| 60 | + $this->channel->basic_qos(0, 1, false); |
64 | 61 |
|
65 | 62 | // Enable publisher confirms if configured
|
66 | 63 | if ($config->rabbitmq['publisherConfirms'] ?? false) {
|
|
0 commit comments