You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

119 lines
4.7 KiB

3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
  1. <?php
  2. namespace App\Console\Commands;
  3. use Anik\Amqp\Exchanges;
  4. use GuzzleHttp\Psr7\Header;
  5. use Illuminate\Support\Arr;
  6. use PhpAmqpLib\Wire\AMQPTable;
  7. use Illuminate\Console\Command;
  8. use Anik\Amqp\ProducibleMessage;
  9. use Anik\Laravel\Amqp\Facades\Amqp;
  10. use PhpAmqpLib\Message\AMQPMessage;
  11. use Anik\Amqp\Exchanges\{Fanout, Direct, Topic, Headers, Exchange};
  12. class Publisher extends Command
  13. {
  14. protected $signature = 'publish
  15. {exchange}
  16. {--declare=1}
  17. {--type=direct}
  18. {--bind=key}
  19. {--durable=1}
  20. {--delivery=1}
  21. {--mandatory=1}
  22. {--count=1}
  23. {--usleep=1}
  24. ';
  25. public function handle()
  26. {
  27. $time = -microtime(true);
  28. echo PHP_EOL;
  29. for ($index = 1; $index <= (int) $this->option('count'); $index++) {
  30. $chance = rand(1, 100);
  31. $payload = $chance <= 20 ? "finish #$index" : ($chance <= 40 ? "requeue #$index" : "Message #$index");
  32. $message = [
  33. 'content_type' => 'text/plain',
  34. 'delivery_mode' => $this->option('delivery') === "1" ? AMQPMessage::DELIVERY_MODE_NON_PERSISTENT : AMQPMessage::DELIVERY_MODE_PERSISTENT,
  35. 'content_encoding' => 'utf-8',
  36. ];
  37. $publish = [
  38. // When a published message cannot be routed to any queue,
  39. // and the publisher set the mandatory message property to
  40. // true, the message will be returned to it. The publisher
  41. // must have a returned message handler set up in order to
  42. // handle the return (e.g. by logging an error or retrying with a different exchange).
  43. 'mandatory' => $this->option('mandatory') === "1",
  44. // This flag is deprecated as of RabbitMQ 2.9 and will raise
  45. // an exception and close the channel if used.
  46. 'immediate' => false,
  47. 'ticket' => null,
  48. 'batch_count' => 500,
  49. ];
  50. $exchange = [
  51. 'name' => $this->argument('exchange'),
  52. 'declare' => $this->option('declare') === "1", // should declare
  53. 'passive' => false,
  54. // By using durability (Durable, Transient) properties, we can
  55. // make the message to survive even after server restarts. If
  56. // we select Durable, then the message will survive even
  57. // after server restart. In case, if we select Tansient,
  58. // then message will not service after server restart.
  59. 'durable' => $this->option('durable') === "1",
  60. // By using auto delete property, we can set whether an exchange
  61. // can delete if we unbind assigned queue.
  62. 'auto_delete' => false,
  63. // If we set this property yes, then the exchange may not be
  64. // used directly by publishers, but only when bound to other
  65. // exchanges.
  66. 'internal' => false,
  67. 'no_wait' => false,
  68. 'arguments' => [],
  69. 'ticket' => null,
  70. ];
  71. $message = new ProducibleMessage(
  72. message: $payload,
  73. properties: $publish + $message,
  74. );
  75. $other_options = [];
  76. Amqp::publish(
  77. messages: $message,
  78. routingKey: $this->option('bind'),
  79. // In rabbitmq, direct exchange will deliver a messages to the
  80. // queues based on the message routing key
  81. exchange: $this->getExchangeType($this->option('type'), $exchange),
  82. // In rabbitmq, fanout exchange will route messages to all of
  83. // the queues that are bound to it.
  84. // exchange: Fanout::make($exchange),
  85. // https://www.tutlane.com/tutorial/rabbitmq/rabbitmq-exchanges
  86. // exchange: Topic::make($exchange),
  87. // exchange: Headers::make($exchange),
  88. options: $other_options,
  89. );
  90. usleep((int) $this->option('usleep') * 1000);
  91. }
  92. echo sprintf('%f', $time += microtime(true));
  93. echo PHP_EOL;
  94. }
  95. public function getExchangeType(string $type, array $exchange): Exchange
  96. {
  97. return match ($type) {
  98. 'direct' => Direct::make($exchange),
  99. 'fanout' => Fanout::make($exchange),
  100. 'topic' => Topic::make($exchange),
  101. 'headers' => Headers::make($exchange),
  102. };
  103. }
  104. }