skolodyazhnyy /
butter-amqplib
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like Channel often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes. You can also have a look at the cohesion graph to spot any un-connected, or weakly-connected components.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
While breaking up the class, it is a good idea to analyze how other classes use Channel, and based on these observations, apply Extract Interface, too.
| 1 | <?php |
||
| 49 | class Channel implements ChannelInterface, WireSubscriberInterface |
||
| 50 | { |
||
| 51 | const STATUS_CLOSED = 0; |
||
| 52 | const STATUS_READY = 1; |
||
| 53 | const STATUS_INACTIVE = 2; |
||
| 54 | |||
| 55 | const MODE_NORMAL = 0; |
||
| 56 | const MODE_CONFIRM = 1; |
||
| 57 | const MODE_TX = 2; |
||
| 58 | |||
| 59 | /** |
||
| 60 | * @var int |
||
| 61 | */ |
||
| 62 | private $id; |
||
| 63 | |||
| 64 | /** |
||
| 65 | * @var WireInterface |
||
| 66 | */ |
||
| 67 | private $wire; |
||
| 68 | |||
| 69 | /** |
||
| 70 | * @var int |
||
| 71 | */ |
||
| 72 | private $status = self::STATUS_CLOSED; |
||
| 73 | |||
| 74 | /** |
||
| 75 | * @var int |
||
| 76 | */ |
||
| 77 | private $mode = self::MODE_NORMAL; |
||
| 78 | |||
| 79 | /** |
||
| 80 | * @var callable[] |
||
| 81 | */ |
||
| 82 | private $consumers = []; |
||
| 83 | |||
| 84 | /** |
||
| 85 | * @var callable |
||
| 86 | */ |
||
| 87 | private $returnCallable; |
||
| 88 | |||
| 89 | /** |
||
| 90 | * @var callable |
||
| 91 | */ |
||
| 92 | private $confirmCallable; |
||
| 93 | |||
| 94 | /** |
||
| 95 | * @param WireInterface $wire |
||
| 96 | * @param int $id |
||
| 97 | */ |
||
| 98 | 56 | public function __construct(WireInterface $wire, $id) |
|
| 99 | { |
||
| 100 | 56 | $this->id = $id; |
|
| 101 | 56 | $this->wire = $wire; |
|
| 102 | 56 | } |
|
| 103 | |||
| 104 | /** |
||
| 105 | * {@inheritdoc} |
||
| 106 | */ |
||
| 107 | 19 | public function open() |
|
| 108 | { |
||
| 109 | 19 | if ($this->status != self::STATUS_CLOSED) { |
|
| 110 | 1 | return $this; |
|
| 111 | } |
||
| 112 | |||
| 113 | 19 | $this->wire->subscribe($this->id, $this); |
|
| 114 | |||
| 115 | 19 | $this->send(new ChannelOpen($this->id, '')) |
|
| 116 | 19 | ->wait(ChannelOpenOk::class); |
|
| 117 | |||
| 118 | 19 | $this->status = self::STATUS_READY; |
|
| 119 | 19 | $this->mode = self::MODE_NORMAL; |
|
| 120 | |||
| 121 | 19 | return $this; |
|
| 122 | } |
||
| 123 | |||
| 124 | /** |
||
| 125 | * {@inheritdoc} |
||
| 126 | */ |
||
| 127 | 1 | public function flow($active) |
|
| 128 | { |
||
| 129 | /** @var ChannelFlowOk $frame */ |
||
| 130 | 1 | $frame = $this->send(new ChannelFlow($this->id, $active)) |
|
| 131 | 1 | ->wait(ChannelFlowOk::class); |
|
| 132 | |||
| 133 | 1 | $this->status = $frame->isActive() ? self::STATUS_READY : |
|
| 134 | 1 | self::STATUS_INACTIVE; |
|
| 135 | |||
| 136 | 1 | return $this; |
|
| 137 | } |
||
| 138 | |||
| 139 | /** |
||
| 140 | * {@inheritdoc} |
||
| 141 | */ |
||
| 142 | 1 | public function serve($blocking = true) |
|
| 143 | { |
||
| 144 | 1 | $this->wire->next($blocking); |
|
| 145 | |||
| 146 | 1 | return $this; |
|
| 147 | } |
||
| 148 | |||
| 149 | /** |
||
| 150 | * {@inheritdoc} |
||
| 151 | */ |
||
| 152 | 1 | public function close() |
|
| 153 | { |
||
| 154 | 1 | $this->send(new ChannelClose($this->id, 0, '', 0, 0)) |
|
| 155 | 1 | ->wait(ChannelCloseOk::class); |
|
| 156 | |||
| 157 | 1 | $this->status = self::STATUS_CLOSED; |
|
| 158 | |||
| 159 | 1 | return $this; |
|
| 160 | } |
||
| 161 | |||
| 162 | /** |
||
| 163 | * {@inheritdoc} |
||
| 164 | */ |
||
| 165 | 1 | public function qos($prefetchSize, $prefetchCount, $globally = false) |
|
| 166 | { |
||
| 167 | 1 | $this->send(new BasicQos($this->id, $prefetchSize, $prefetchCount, $globally)) |
|
| 168 | 1 | ->wait(BasicQosOk::class); |
|
| 169 | |||
| 170 | 1 | return $this; |
|
| 171 | } |
||
| 172 | |||
| 173 | /** |
||
| 174 | * {@inheritdoc} |
||
| 175 | */ |
||
| 176 | 4 | public function exchange($name) |
|
| 177 | { |
||
| 178 | 4 | return new Exchange($this->wire, $this->id, $name); |
|
| 179 | } |
||
| 180 | |||
| 181 | /** |
||
| 182 | * {@inheritdoc} |
||
| 183 | */ |
||
| 184 | 14 | public function queue($name = '') |
|
| 185 | { |
||
| 186 | 14 | return new Queue($this->wire, $this->id, $name); |
|
| 187 | } |
||
| 188 | |||
| 189 | /** |
||
| 190 | * {@inheritdoc} |
||
| 191 | */ |
||
| 192 | 11 | public function consume($queue, callable $callback, $flags = 0, $tag = '', array $arguments = []) |
|
| 193 | { |
||
| 194 | 11 | if (empty($tag) && $flags & Consumer::FLAG_NO_WAIT) { |
|
| 195 | 1 | $tag = uniqid('php-consumer-'); |
|
| 196 | 1 | } |
|
| 197 | |||
| 198 | 11 | $this->send(new BasicConsume( |
|
| 199 | 11 | $this->id, |
|
| 200 | 11 | 0, |
|
| 201 | 11 | $queue, |
|
| 202 | 11 | $tag, |
|
| 203 | 11 | $flags & Consumer::FLAG_NO_LOCAL, |
|
| 204 | 11 | $flags & Consumer::FLAG_NO_ACK, |
|
| 205 | 11 | $flags & Consumer::FLAG_EXCLUSIVE, |
|
| 206 | 11 | $flags & Consumer::FLAG_NO_WAIT, |
|
| 207 | $arguments |
||
| 208 | 11 | )); |
|
| 209 | |||
| 210 | 11 | if (!($flags & Consumer::FLAG_NO_WAIT)) { |
|
| 211 | 7 | $tag = $this->wait(BasicConsumeOk::class) |
|
| 212 | 7 | ->getConsumerTag(); |
|
| 213 | 7 | } |
|
| 214 | |||
| 215 | 11 | $this->consumers[$tag] = $callback; |
|
| 216 | |||
| 217 | 11 | return new Consumer($this, $tag); |
|
| 218 | } |
||
| 219 | |||
| 220 | /** |
||
| 221 | * {@inheritdoc} |
||
| 222 | */ |
||
| 223 | 7 | public function get($queue, $withAck = true) |
|
| 224 | { |
||
| 225 | /** @var BasicGetOk|BasicGetEmpty $frame */ |
||
| 226 | 7 | $frame = $this->send(new BasicGet($this->id, 0, $queue, !$withAck)) |
|
| 227 | 7 | ->wait([BasicGetOk::class, BasicGetEmpty::class]); |
|
| 228 | |||
| 229 | 7 | if ($frame instanceof BasicGetEmpty) { |
|
|
1 ignored issue
–
show
|
|||
| 230 | 2 | return null; |
|
| 231 | } |
||
| 232 | |||
| 233 | /** @var Header $header */ |
||
| 234 | 5 | $header = $this->wait(Header::class); |
|
| 235 | 5 | $content = ''; |
|
| 236 | |||
| 237 | 5 | while ($header->getSize() > strlen($content)) { |
|
| 238 | 3 | $content .= $this->wait(Content::class)->getData(); |
|
| 239 | 3 | } |
|
| 240 | |||
| 241 | 5 | return new Delivery( |
|
| 242 | 5 | $this, |
|
| 243 | 5 | '', |
|
| 244 | 5 | $frame->getDeliveryTag(), |
|
| 245 | 5 | $frame->isRedelivered(), |
|
| 246 | 5 | $frame->getExchange(), |
|
| 247 | 5 | $frame->getRoutingKey(), |
|
| 248 | 5 | $content, |
|
| 249 | 5 | $header->getProperties() |
|
| 250 | 5 | ); |
|
| 251 | } |
||
| 252 | |||
| 253 | /** |
||
| 254 | * {@inheritdoc} |
||
| 255 | */ |
||
| 256 | 2 | public function recover($requeue = true) |
|
| 257 | { |
||
| 258 | 2 | $this->send(new BasicRecover($this->id, $requeue)) |
|
| 259 | 2 | ->wait(BasicRecoverOk::class); |
|
| 260 | |||
| 261 | 2 | return $this; |
|
| 262 | } |
||
| 263 | |||
| 264 | /** |
||
| 265 | * {@inheritdoc} |
||
| 266 | */ |
||
| 267 | 6 | public function cancel($tag, $flags = 0) |
|
| 268 | { |
||
| 269 | 6 | $this->send(new BasicCancel($this->id, $tag, $flags & Consumer::FLAG_NO_WAIT)); |
|
| 270 | |||
| 271 | 6 | unset($this->consumers[$tag]); |
|
| 272 | |||
| 273 | 6 | if ($flags & Consumer::FLAG_NO_WAIT) { |
|
| 274 | 1 | return $this; |
|
| 275 | } |
||
| 276 | |||
| 277 | 5 | $this->wait(BasicCancelOk::class); |
|
| 278 | |||
| 279 | 5 | return $this; |
|
| 280 | } |
||
| 281 | |||
| 282 | /** |
||
| 283 | * {@inheritdoc} |
||
| 284 | */ |
||
| 285 | 15 | public function publish(Message $message, $exchange = '', $routingKey = '', $flags = 0) |
|
| 286 | { |
||
| 287 | 15 | $this->send(new BasicPublish( |
|
| 288 | 15 | $this->id, |
|
| 289 | 15 | 0, |
|
| 290 | 15 | $exchange, |
|
| 291 | 15 | $routingKey, |
|
| 292 | 15 | (bool) ($flags & Message::FLAG_MANDATORY), |
|
| 293 | 15 | (bool) ($flags & Message::FLAG_IMMEDIATE) |
|
| 294 | 15 | )); |
|
| 295 | |||
| 296 | 15 | $body = $message->getBody(); |
|
| 297 | |||
| 298 | 15 | $this->send(new Header($this->id, 60, 0, strlen($body), $message->getProperties())); |
|
| 299 | 15 | $this->send(new Content($this->id, $body)); |
|
| 300 | |||
| 301 | 15 | return $this; |
|
| 302 | } |
||
| 303 | |||
| 304 | /** |
||
| 305 | * {@inheritdoc} |
||
| 306 | */ |
||
| 307 | 6 | public function ack($deliveryTag, $multiple = false) |
|
| 308 | { |
||
| 309 | 6 | $this->send(new BasicAck($this->id, $deliveryTag, $multiple)); |
|
| 310 | |||
| 311 | 6 | return $this; |
|
| 312 | } |
||
| 313 | |||
| 314 | /** |
||
| 315 | * {@inheritdoc} |
||
| 316 | */ |
||
| 317 | 4 | public function reject($deliveryTag, $requeue = true, $multiple = false) |
|
| 318 | { |
||
| 319 | 4 | $multiple ? $this->send(new BasicNack($this->id, $deliveryTag, $multiple, $requeue)) : |
|
| 320 | 3 | $this->send(new BasicReject($this->id, $deliveryTag, $requeue)); |
|
| 321 | |||
| 322 | 4 | return $this; |
|
| 323 | } |
||
| 324 | |||
| 325 | /** |
||
| 326 | * {@inheritdoc} |
||
| 327 | */ |
||
| 328 | 2 | public function onReturn(callable $callable) |
|
| 329 | { |
||
| 330 | 2 | $this->returnCallable = $callable; |
|
| 331 | |||
| 332 | 2 | return $this; |
|
| 333 | } |
||
| 334 | |||
| 335 | /** |
||
| 336 | * {@inheritdoc} |
||
| 337 | */ |
||
| 338 | 5 | public function selectConfirm(callable $callable, $noWait = false) |
|
| 339 | { |
||
| 340 | 5 | $this->confirmCallable = $callable; |
|
| 341 | |||
| 342 | 5 | $this->send(new ConfirmSelect($this->id, $noWait)); |
|
| 343 | |||
| 344 | 5 | if (!$noWait) { |
|
| 345 | 4 | $this->wait(ConfirmSelectOk::class); |
|
| 346 | 4 | } |
|
| 347 | |||
| 348 | 5 | $this->mode = self::MODE_CONFIRM; |
|
| 349 | |||
| 350 | 5 | return $this; |
|
| 351 | } |
||
| 352 | |||
| 353 | /** |
||
| 354 | * {@inheritdoc} |
||
| 355 | */ |
||
| 356 | 7 | public function selectTx() |
|
| 357 | { |
||
| 358 | 7 | $this->send(new TxSelect($this->id)) |
|
| 359 | 7 | ->wait(TxSelectOk::class); |
|
| 360 | |||
| 361 | 7 | $this->mode = self::MODE_TX; |
|
| 362 | |||
| 363 | 7 | return $this; |
|
| 364 | } |
||
| 365 | |||
| 366 | /** |
||
| 367 | * {@inheritdoc} |
||
| 368 | */ |
||
| 369 | 6 | public function txCommit() |
|
| 370 | { |
||
| 371 | 6 | if ($this->mode != self::MODE_TX) { |
|
| 372 | 1 | throw new TransactionNotSelectedException('Channel is not in transaction mode. Use Channel::selectTx() to select transaction mode on this channel.'); |
|
| 373 | } |
||
| 374 | |||
| 375 | 5 | $this->send(new TxCommit($this->id)) |
|
| 376 | 5 | ->wait(TxCommitOk::class); |
|
| 377 | |||
| 378 | 5 | return; |
|
| 379 | } |
||
| 380 | |||
| 381 | /** |
||
| 382 | * {@inheritdoc} |
||
| 383 | */ |
||
| 384 | 4 | public function txRollback() |
|
| 385 | { |
||
| 386 | 4 | if ($this->mode != self::MODE_TX) { |
|
| 387 | 1 | throw new TransactionNotSelectedException('Channel is not in transaction mode. Use Channel::selectTx() to select transaction mode on this channel.'); |
|
| 388 | } |
||
| 389 | |||
| 390 | 3 | $this->send(new TxRollback($this->id)) |
|
| 391 | 3 | ->wait(TxRollbackOk::class); |
|
| 392 | |||
| 393 | 3 | return; |
|
| 394 | } |
||
| 395 | |||
| 396 | /** |
||
| 397 | * {@inheritdoc} |
||
| 398 | */ |
||
| 399 | 10 | public function hasConsumer($tag) |
|
| 400 | { |
||
| 401 | 10 | return isset($this->consumers[(string) $tag]); |
|
| 402 | } |
||
| 403 | |||
| 404 | /** |
||
| 405 | * {@inheritdoc} |
||
| 406 | */ |
||
| 407 | 3 | public function getConsumerTags() |
|
| 408 | { |
||
| 409 | 3 | return array_keys($this->consumers); |
|
| 410 | } |
||
| 411 | |||
| 412 | /** |
||
| 413 | * @return string |
||
| 414 | */ |
||
| 415 | 3 | public function getStatus() |
|
| 416 | { |
||
| 417 | 3 | return $this->status; |
|
| 418 | } |
||
| 419 | |||
| 420 | /** |
||
| 421 | * Sends frame to the server. |
||
| 422 | * |
||
| 423 | * @param Frame $frame |
||
| 424 | * |
||
| 425 | * @return $this |
||
| 426 | */ |
||
| 427 | 46 | private function send(Frame $frame) |
|
| 433 | |||
| 434 | /** |
||
| 435 | * @param string|array $type |
||
| 436 | * |
||
| 437 | * @return Frame |
||
| 438 | */ |
||
| 439 | 37 | private function wait($type) |
|
| 443 | |||
| 444 | /** |
||
| 445 | * @param Frame $frame |
||
| 446 | */ |
||
| 447 | 29 | public function dispatch(Frame $frame) |
|
| 465 | |||
| 466 | /** |
||
| 467 | * @param BasicDeliver $frame |
||
| 468 | * |
||
| 469 | * @throws \Exception |
||
| 470 | */ |
||
| 471 | 6 | private function onBasicDeliver(BasicDeliver $frame) |
|
| 501 | |||
| 502 | /** |
||
| 503 | * @param BasicReturn $frame |
||
| 504 | * |
||
| 505 | * @throws \Exception |
||
| 506 | */ |
||
| 507 | 4 | private function onBasicReturn(BasicReturn $frame) |
|
| 536 | |||
| 537 | /** |
||
| 538 | * @param BasicAck $frame |
||
| 539 | */ |
||
| 540 | 3 | View Code Duplication | private function onBasicAck(BasicAck $frame) |
| 550 | |||
| 551 | /** |
||
| 552 | * @param BasicNack $frame |
||
| 553 | */ |
||
| 554 | 2 | View Code Duplication | private function onBasicNack(BasicNack $frame) |
| 564 | |||
| 565 | /** |
||
| 566 | * @param BasicCancel $frame |
||
| 567 | */ |
||
| 568 | 3 | private function onBasicCancel(BasicCancel $frame) |
|
| 576 | |||
| 577 | /** |
||
| 578 | * @param ChannelFlow $frame |
||
| 579 | */ |
||
| 580 | 1 | private function onChannelFlow(ChannelFlow $frame) |
|
| 586 | |||
| 587 | /** |
||
| 588 | * @param ChannelClose $frame |
||
| 589 | * |
||
| 590 | * @throws AMQPException |
||
| 591 | */ |
||
| 592 | 3 | private function onChannelClose(ChannelClose $frame) |
|
| 600 | } |
||
| 601 |
Loading history...
This error could be the result of:
1. Missing dependencies
PHP Analyzer uses your
composer.jsonfile (if available) to determine the dependencies of your project and to determine all the available classes and functions. It expects thecomposer.jsonto be in the root folder of your repository.Are you sure this class is defined by one of your dependencies, or did you maybe not list a dependency in either the
requireorrequire-devsection?2. Missing use statement
PHP does not complain about undefined classes in
ìnstanceofchecks. For example, the following PHP code will work perfectly fine:If you have not tested against this specific condition, such errors might go unnoticed.