lanzu_api_2021-12-26/app/TaskWorker/MQTTClientTask.php

<?php
declare(strict_types=1);

namespace App\TaskWorker;

use App\Libs\MQTTClient;
use Hyperf\Task\Annotation\Task;

class MQTTClientTask
{
    protected $mqttClient = null;

    /**
     * @Task
     * @param string|number $message 消息内容
     * @param string $topic 发布消息到主题，主题名
     * @param string $type 消息类型，cash或tts
     * @param string $payId 支付方式，如“支付宝”、“微信”等
     * @param string $toClientId 终端id，如IMEI码
     * @param string $curClientId 当前客户端id
     */
    public function publish(
        $message,
        $topic,
        $toClientId = '',
        $type = '',
        $payId = '',
        $curClientId = ''
    ) {
        $this->getClient();

        $msgArr = [];
        if ( (empty($type)&&is_numeric($message)) || 'cash' === $type ) {
            $msgArr['cash'] = $message;
            $payId AND $msgArr['payid'] = $payId;
        } else {
            $msgArr['message'] = $message;
        }

        if (!empty($toClientId)) {
            $topic .= '/'.$toClientId;
        }

        $curClientId OR $curClientId = (string)rand(1,999999999);
        $success = $this->mqttClient->sendConnect($curClientId);

        if ($success) {
            $this->mqttClient->sendPublish($topic, json_encode($msgArr), MQTTClient::MQTT_QOS2);
            $this->mqttClient->sendDisconnect();
        }
        $this->mqttClient->close();
    }

    protected function getClient()
    {
        $this->mqttClient = new MQTTClient(env('MQTT_HOST'), env('MQTT_PORT'));
        $this->mqttClient->setAuthentication(env('MQTT_NAME'), env('MQTT_PASS'));

        if (env('MQTT_CERT')) {
            $this->mqttClient->setEncryption(env('MQTT_CERT'));
        }

        return $this->mqttClient;
    }
}