QueueForm.php 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117
  1. <?php
  2. namespace backend\forms;
  3. use components\Curl;
  4. use components\Exception;
  5. use Yii;
  6. /**
  7. * Class QueueForm
  8. * @package backend\forms
  9. */
  10. class QueueForm extends BaseForm
  11. {
  12. /**
  13. * @var string 名称
  14. */
  15. public $name;
  16. public function rules()
  17. {
  18. return [
  19. //create_queue delete_queue
  20. [['name'], 'required', 'message' => 1100, 'on' => ['create_queue', 'delete_queue']],
  21. ['name', 'string', 'message' => 1101, 'on' => ['create_queue', 'delete_queue']],
  22. ['name', 'filter', 'filter' => 'trim', 'on' => ['create_queue', 'delete_queue']],
  23. ];
  24. }
  25. /**
  26. * [创建队列]
  27. * @author: libingke
  28. * @return array
  29. * @throws Exception
  30. */
  31. public function createQueue()
  32. {
  33. try {
  34. $connect = $this->getConnect();
  35. $channel = $connect->channel();
  36. list($queue,,) = $channel->queue_declare($this->name,
  37. false, true, false, false, false);
  38. } catch (\Exception $e) {
  39. throw new Exception(1001, $e->getMessage());
  40. }
  41. return $data = ['name' => $queue];
  42. }
  43. /**
  44. * [删除队列]
  45. * @author: libingke
  46. * @return array
  47. * @throws Exception
  48. */
  49. public function deleteQueue()
  50. {
  51. try {
  52. $connect = $this->getConnect();
  53. $channel = $connect->channel();
  54. $channel->queue_delete($this->name,
  55. false, false, false, false
  56. );
  57. return [
  58. 'name' => $this->name,
  59. 'result' => '删除成功'
  60. ];
  61. } catch (\Exception $e) {
  62. throw new Exception(1001, $e->getMessage());
  63. }
  64. }
  65. /**
  66. * [获取消息列表]
  67. * @author: libingke
  68. */
  69. public function getQueueList()
  70. {
  71. $authStr = Yii::$app->Amqp->user . ':' . Yii::$app->Amqp->pass;
  72. $url = Yii::$app->Amqp->host . ':' . Yii::$app->Amqp->api_port . "/api/queues";
  73. $curl = new Curl();
  74. $curl->setOption(CURLOPT_USERPWD, $authStr);
  75. $result = json_decode($curl->get($url), true);
  76. if ($curl->responseCode != 200)
  77. throw new Exception(1002);
  78. if ($curl->errorText)
  79. throw new Exception(1002, $curl->errorText);
  80. if (isset($result['error']) && is_string($result['error']))
  81. throw new Exception(1002, $result['error']);
  82. $rows = [];
  83. foreach ($result as $k => $v) {
  84. //$name = $v['name'];
  85. $name = $v['name'];
  86. $rows[$name]['name'] = $name;
  87. $rows[$name]['messages_count'] = $v['messages'];
  88. $rows[$name]['message_bytes'] = $v['message_bytes'];
  89. $rows[$name]['messages_ready'] = $v['messages_ready'];
  90. //$rows[$name]['message_stats'] = $v['message_stats'];
  91. $rows[$name]['consumers_count'] = $v['consumers'];
  92. $rows[$name]['auto_delete'] = $v['auto_delete'];
  93. $rows[$name]['durable'] = $v['durable'];
  94. //$rows[$name]['arguments'] = $v['arguments'];
  95. $rows[$name]['state'] = $v['state'];
  96. $rows[$name]['idle_since'] = $v['idle_since'];
  97. }
  98. unset($result);
  99. return ['count' => count($rows), 'rows' => $rows];
  100. }
  101. }