swooleMqtt.php

<?php

namespace App\Console\Commands;

use App\Http\Controllers\Auth\SwooleCommandController;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Log;

class swooleMqtt extends Command
{
    public $mqtt;
    protected $signature = 'swooleMqtt {action?}';
    protected $description = 'swooleMqtt';
    public function __construct()
    {
        parent::__construct();
    }
    public function handle()
    {
        $action = $this->argument('action');
        switch ($action) {
            case 'close':
                break;
            default:
                $this->start();//开启mqtt服务
                break;
        }

    }
   public function decodeValue($data)
    {
        return 256 * ord($data[0]) + ord($data[1]);
    }

    public function decodeString($data)
    {
        $length = $this->decodeValue($data);
        return substr($data, 2, $length);
    }

    public function mqttGetHeader($data)
    {
        $byte = ord($data[0]);

        $header['type'] = ($byte & 0xF0) >> 4;
        $header['dup'] = ($byte & 0x08) >> 3;
        $header['qos'] = ($byte & 0x06) >> 1;
        $header['retain'] = $byte & 0x01;

        return $header;
    }

    public function eventConnect($header, $data)
    {
        $connect_info['protocol_name'] = $this->decodeString($data);
        $offset = strlen($connect_info['protocol_name']) + 2;

        $connect_info['version'] = ord(substr($data, $offset, 1));
        $offset += 1;

        $byte = ord($data[$offset]);
        $connect_info['willRetain'] = ($byte & 0x20 == 0x20);
        $connect_info['willQos'] = ($byte & 0x18 >> 3);
        $connect_info['willFlag'] = ($byte & 0x04 == 0x04);
        $connect_info['cleanStart'] = ($byte & 0x02 == 0x02);
        $offset += 1;

        $connect_info['keepalive'] = $this->decodeValue(substr($data, $offset, 2));
        $offset += 2;
        $connect_info['clientId'] = $this->decodeString(substr($data, $offset));
        return $connect_info;
    }
    public function start()
    {
        $this->mqtt = new \swoole_server("0.0.0.0", 9506, SWOOLE_BASE,SWOOLE_SOCK_TCP | SWOOLE_SSL);
        $this->mqtt->set([
            'open_mqtt_protocol' => true, // 启用 mqtt 协议
            'worker_num' => 1,
        ]);
        $this->mqtt->on('connect', function ($server, $fd) {
            echo "Client:Connect.\n";
        });
        $this->mqtt->on('receive', function ($server, $fd, $from_id, $data) {
            $header = $this->mqttGetHeader($data);
            print_r($data);
            if ($header['type'] == 1) {
                $resp = chr(32) . chr(2) . chr(0) . chr(0);
                $this->eventConnect($header, substr($data, 2));
//                print_r($resp);
                $this->mqtt->send($fd, $resp);
            } elseif ($header['type'] == 3) {
                $resp = chr(32) . chr(2) . chr(0) . chr(0);
                $this->eventConnect($header, substr($data, 2));
                $this->mqtt->send($fd, $resp);
                print_r($data);
                //file_put_contents(__DIR__.'/data.log', $data);
            }
            echo "received length=" . strlen($data) . "\n";
        });

        $this->mqtt->on('close', function ($server, $fd) {
            echo "Client: Close.\n";
        });
        $this->mqtt->start();
    }
}