logSystem-advance/backend/forms/TopicForm.php

<?php

namespace backend\forms;

use components\Curl;
use components\Exception;
use Yii;

/**
 * Class QueueForm
 * @package backend\forms
 */
class TopicForm extends BaseForm
{
	/**
	 * @var string 名称
	 */
	public $name;

	public $name1;
	
	public $name2;

	public function rules()
	{
		return [
			//create_queue  delete_queue
			[['name'], 'required', 'message' => 1100, 'on' => ['create_topic', 'delete_topic']],
			[['name1','name2'], 'required', 'message' => 1100, 'on' => ['update_topic']],
			['name', 'string', 'message' => 1101, 'on' => ['create_topic', 'delete_topic']],
			['name', 'filter', 'filter' => 'trim', 'on' => ['create_topic', 'delete_topic']],
		];
	}

	/**
	 * [创建队列]
	 * @author: libingke
	 * @return array
	 * @throws Exception
	 */
	public function createTopic()
	{
		try {
			$connect = $this->getConnect();
			$channel = $connect->channel();
			list($topic,,) = $channel->exchange_declare($this->name, 'topic', false, true, false);
        } catch (\Exception $e) {
			throw new Exception(1001, $e->getMessage());
		}
		return $data = ['TopicName' => empty($topic)?$this->name:$topic];
	}

	/**
	 * [删除队列]
	 * @author: libingke
	 * @return array
	 * @throws Exception
	 */
	public function deleteTopic()
	{
		try {
			$connect = $this->getConnect();
			$channel = $connect->channel();
			$channel->exchange_delete($this->name, false, false );
			return [
				'name' => $this->name,
				'result' => '删除成功'
			];

		} catch (\Exception $e) {
			throw new Exception(1001, $e->getMessage());
		}
	}



    /**
     * [获取消息列表]
     * @author: libingke
     */
    public function getTopicList()
    {
        $authStr = Yii::$app->Amqp->user . ':' . Yii::$app->Amqp->pass;
        $url = Yii::$app->Amqp->host . ':' . Yii::$app->Amqp->api_port . "/api/exchanges";

        $curl = new Curl();
        $curl->setOption(CURLOPT_USERPWD, $authStr);
        $result = json_decode($curl->get($url), true);


        if ($curl->responseCode != 200)
            throw new Exception(1002);

        if ($curl->errorText)
            throw new Exception(1002, $curl->errorText);

        if (isset($result['error']) && is_string($result['error']))
            throw new Exception(1002, $result['error']);

        $rows = [];
        foreach ($result as $k => $v) {
            if($v['type'] = 'topic'){
                $name = $v['name'];
                $rows[$name]['name'] 			= $name;
                $rows[$name]['auto_delete']		= $v['auto_delete'];
                $rows[$name]['durable']			= $v['durable'];
                $rows[$name]['arguments']		= $v['arguments'];
                $rows[$name]['type']			= $v['type'];
                $rows[$name]['vhost']		= $v['vhost'];
            }
        }
        unset($result);

        return ['count' => count($rows), 'rows' => $rows];
    }

    /**
     * [修改队列]
     * @author: hanguangxu
     * @return array
     * @throws Exception
     */
    public function updateTopic()
    {
        try {
            $connect = $this->getConnect();
            $channel = $connect->channel();
            $channel->exchange_delete($this->name1, false, false );
            list($topic,,) = $channel->exchange_declare($this->name2, 'topic', false, true, false);
        } catch (\Exception $e) {
            throw new Exception(1001, $e->getMessage());
        }
        return $data = ['TopicName' => empty($topic)?$this->name2:$topic];
    }

}