| 148 |
lars |
1 |
<?php declare(strict_types=1);
|
|
|
2 |
|
|
|
3 |
/*
|
|
|
4 |
* This file is part of the Monolog package.
|
|
|
5 |
*
|
|
|
6 |
* (c) Jordi Boggiano <j.boggiano@seld.be>
|
|
|
7 |
*
|
|
|
8 |
* For the full copyright and license information, please view the LICENSE
|
|
|
9 |
* file that was distributed with this source code.
|
|
|
10 |
*/
|
|
|
11 |
|
|
|
12 |
namespace Monolog\Handler;
|
|
|
13 |
|
|
|
14 |
use Monolog\Logger;
|
|
|
15 |
use Monolog\Formatter\FormatterInterface;
|
|
|
16 |
use Monolog\Formatter\JsonFormatter;
|
|
|
17 |
use PhpAmqpLib\Message\AMQPMessage;
|
|
|
18 |
use PhpAmqpLib\Channel\AMQPChannel;
|
|
|
19 |
use AMQPExchange;
|
|
|
20 |
|
|
|
21 |
/**
|
|
|
22 |
* @phpstan-import-type Record from \Monolog\Logger
|
|
|
23 |
*/
|
|
|
24 |
class AmqpHandler extends AbstractProcessingHandler
|
|
|
25 |
{
|
|
|
26 |
/**
|
|
|
27 |
* @var AMQPExchange|AMQPChannel $exchange
|
|
|
28 |
*/
|
|
|
29 |
protected $exchange;
|
|
|
30 |
/** @var array<string, mixed> */
|
|
|
31 |
private $extraAttributes = [];
|
|
|
32 |
|
|
|
33 |
/**
|
|
|
34 |
* @return array<string, mixed>
|
|
|
35 |
*/
|
|
|
36 |
public function getExtraAttributes(): array
|
|
|
37 |
{
|
|
|
38 |
return $this->extraAttributes;
|
|
|
39 |
}
|
|
|
40 |
|
|
|
41 |
/**
|
|
|
42 |
* Configure extra attributes to pass to the AMQPExchange (if you are using the amqp extension)
|
|
|
43 |
*
|
|
|
44 |
* @param array<string, mixed> $extraAttributes One of content_type, content_encoding,
|
|
|
45 |
* message_id, user_id, app_id, delivery_mode,
|
|
|
46 |
* priority, timestamp, expiration, type
|
|
|
47 |
* or reply_to, headers.
|
|
|
48 |
* @return AmqpHandler
|
|
|
49 |
*/
|
|
|
50 |
public function setExtraAttributes(array $extraAttributes): self
|
|
|
51 |
{
|
|
|
52 |
$this->extraAttributes = $extraAttributes;
|
|
|
53 |
return $this;
|
|
|
54 |
}
|
|
|
55 |
|
|
|
56 |
/**
|
|
|
57 |
* @var string
|
|
|
58 |
*/
|
|
|
59 |
protected $exchangeName;
|
|
|
60 |
|
|
|
61 |
/**
|
|
|
62 |
* @param AMQPExchange|AMQPChannel $exchange AMQPExchange (php AMQP ext) or PHP AMQP lib channel, ready for use
|
|
|
63 |
* @param string|null $exchangeName Optional exchange name, for AMQPChannel (PhpAmqpLib) only
|
|
|
64 |
*/
|
|
|
65 |
public function __construct($exchange, ?string $exchangeName = null, $level = Logger::DEBUG, bool $bubble = true)
|
|
|
66 |
{
|
|
|
67 |
if ($exchange instanceof AMQPChannel) {
|
|
|
68 |
$this->exchangeName = (string) $exchangeName;
|
|
|
69 |
} elseif (!$exchange instanceof AMQPExchange) {
|
|
|
70 |
throw new \InvalidArgumentException('PhpAmqpLib\Channel\AMQPChannel or AMQPExchange instance required');
|
|
|
71 |
} elseif ($exchangeName) {
|
|
|
72 |
@trigger_error('The $exchangeName parameter can only be passed when using PhpAmqpLib, if using an AMQPExchange instance configure it beforehand', E_USER_DEPRECATED);
|
|
|
73 |
}
|
|
|
74 |
$this->exchange = $exchange;
|
|
|
75 |
|
|
|
76 |
parent::__construct($level, $bubble);
|
|
|
77 |
}
|
|
|
78 |
|
|
|
79 |
/**
|
|
|
80 |
* {@inheritDoc}
|
|
|
81 |
*/
|
|
|
82 |
protected function write(array $record): void
|
|
|
83 |
{
|
|
|
84 |
$data = $record["formatted"];
|
|
|
85 |
$routingKey = $this->getRoutingKey($record);
|
|
|
86 |
|
|
|
87 |
if ($this->exchange instanceof AMQPExchange) {
|
|
|
88 |
$attributes = [
|
|
|
89 |
'delivery_mode' => 2,
|
|
|
90 |
'content_type' => 'application/json',
|
|
|
91 |
];
|
|
|
92 |
if ($this->extraAttributes) {
|
|
|
93 |
$attributes = array_merge($attributes, $this->extraAttributes);
|
|
|
94 |
}
|
|
|
95 |
$this->exchange->publish(
|
|
|
96 |
$data,
|
|
|
97 |
$routingKey,
|
|
|
98 |
0,
|
|
|
99 |
$attributes
|
|
|
100 |
);
|
|
|
101 |
} else {
|
|
|
102 |
$this->exchange->basic_publish(
|
|
|
103 |
$this->createAmqpMessage($data),
|
|
|
104 |
$this->exchangeName,
|
|
|
105 |
$routingKey
|
|
|
106 |
);
|
|
|
107 |
}
|
|
|
108 |
}
|
|
|
109 |
|
|
|
110 |
/**
|
|
|
111 |
* {@inheritDoc}
|
|
|
112 |
*/
|
|
|
113 |
public function handleBatch(array $records): void
|
|
|
114 |
{
|
|
|
115 |
if ($this->exchange instanceof AMQPExchange) {
|
|
|
116 |
parent::handleBatch($records);
|
|
|
117 |
|
|
|
118 |
return;
|
|
|
119 |
}
|
|
|
120 |
|
|
|
121 |
foreach ($records as $record) {
|
|
|
122 |
if (!$this->isHandling($record)) {
|
|
|
123 |
continue;
|
|
|
124 |
}
|
|
|
125 |
|
|
|
126 |
/** @var Record $record */
|
|
|
127 |
$record = $this->processRecord($record);
|
|
|
128 |
$data = $this->getFormatter()->format($record);
|
|
|
129 |
|
|
|
130 |
$this->exchange->batch_basic_publish(
|
|
|
131 |
$this->createAmqpMessage($data),
|
|
|
132 |
$this->exchangeName,
|
|
|
133 |
$this->getRoutingKey($record)
|
|
|
134 |
);
|
|
|
135 |
}
|
|
|
136 |
|
|
|
137 |
$this->exchange->publish_batch();
|
|
|
138 |
}
|
|
|
139 |
|
|
|
140 |
/**
|
|
|
141 |
* Gets the routing key for the AMQP exchange
|
|
|
142 |
*
|
|
|
143 |
* @phpstan-param Record $record
|
|
|
144 |
*/
|
|
|
145 |
protected function getRoutingKey(array $record): string
|
|
|
146 |
{
|
|
|
147 |
$routingKey = sprintf('%s.%s', $record['level_name'], $record['channel']);
|
|
|
148 |
|
|
|
149 |
return strtolower($routingKey);
|
|
|
150 |
}
|
|
|
151 |
|
|
|
152 |
private function createAmqpMessage(string $data): AMQPMessage
|
|
|
153 |
{
|
|
|
154 |
return new AMQPMessage(
|
|
|
155 |
$data,
|
|
|
156 |
[
|
|
|
157 |
'delivery_mode' => 2,
|
|
|
158 |
'content_type' => 'application/json',
|
|
|
159 |
]
|
|
|
160 |
);
|
|
|
161 |
}
|
|
|
162 |
|
|
|
163 |
/**
|
|
|
164 |
* {@inheritDoc}
|
|
|
165 |
*/
|
|
|
166 |
protected function getDefaultFormatter(): FormatterInterface
|
|
|
167 |
{
|
|
|
168 |
return new JsonFormatter(JsonFormatter::BATCH_MODE_JSON, false);
|
|
|
169 |
}
|
|
|
170 |
}
|