更新

app/api/controller/IndexController.php

@@ -4,6 +4,7 @@ namespace app\api\controller;
 
 use app\BaseController;
 use support\Log;
+use think\facade\Db;
 
 class IndexController extends BaseController{
     public function index(){
@@ -16,5 +17,21 @@ class IndexController extends BaseController{
         return json(['msg'=>'ok']);
     }
 
- 
+    public function mqtt(){
+        $parmas=$this->request->post();
+        $data=[
+            'username'=>$parmas['username'],
+            'topic'=>$parmas['topic'],
+            'qos'=>$parmas['qos'],
+            'data'=>$parmas['payload'],
+            'clientid'=>$parmas['clientid'],
+            'create_time'=>date('Y-m-d H:i:s'),
+        ];
+        $res=Db::name('msg')->instal($data);
+        if($res){
+            return json(['msg'=>'ok']);
+        }else{
+            return json(['msg'=>'添加失败']);
+        }
+    }
 }

app/mqtt/Publish.php

@@ -1,22 +1,53 @@
 <?php
 
 namespace app\mqtt;
+
+use Workerman\Connection\TcpConnection;
+
+
+/**
+ * 发送
+ */
 class Publish
 {
 
-    public static function onWorkerStart()
+    public  function onConnect(TcpConnection $connection)
     {
-        $mqtt = new \Workerman\Mqtt\Client('mqtt://ceshi-mqtt.lihaink.cn/tcp', array(
+        $mqtt = new \Workerman\Mqtt\Client('mqtt://ceshi-mqtt.lihaink.cn:1883', array(
-            "username"=>"demo2",
+            "username" => "demo2",
-            "password"=>"123456",
+            "password" => "123456",
-            "client_id"=>"mqttx_4fde83eb"
+            "client_id" => "admin_123",
         ));
-        $mqtt->onConnect = function($mqtt) {
-            $mqtt->publish('demo', 'hello workerman mqtt');
-        };
-        // $mqtt->onMessage = function($topic, $content) {
-        //     echo "topic:$topic content:$content\n";
-        // };
         $mqtt->connect();
+        $connection->mqtt = $mqtt;
     }
-}
+
+    /**
+     * {"topic":"demo","content":"asdasd"}
+     */
+    public function onMessage(TcpConnection $connection, $data)
+    {
+
+        $data = json_decode($data, true);
+        if ($data == null) {
+            $connection->send("参数不能为空");
+            return false;
+        }
+        if ($data['topic'] == '') {
+            $connection->send("topic为空");
+            return false;
+        }
+        if ($data['content'] == '') {
+            $connection->send("content为空");
+            return false;
+        }
+        $topic = $data['topic'];
+        $content = $data['content'];
+        $res = $connection->mqtt->publish($topic, $content);
+        if ($res == null) {
+            $connection->send("发布成功");
+        } else {
+            $connection->send("发布失败");
+        }
+    }
+}

app/mqtt/Subscribe.php

@@ -2,25 +2,59 @@
 
 namespace app\mqtt;
 
+use Workerman\Connection\TcpConnection;
+
+/**
+ * 订阅
+ */
 class Subscribe
 {
-
+    public  function onConnect(TcpConnection $connection)
-    public static function onWorkerStart()
     {
-        $mqtt = new \Workerman\Mqtt\Client('mqtt://ceshi-mqtt.lihaink.cn/tcp', array(
+        $mqtt = new \Workerman\Mqtt\Client('mqtt://ceshi-mqtt.lihaink.cn:1883', array(
-            // 'debug' => true,
+            "username" => "demo2",
-            "username"=>"demo2",
+            "password" => "123456",
-            "password"=>"123456",
+            "client_id" => "admin_123",
-            "client_id"=>"mqttx_4fde83eb"
         ));
-        $mqtt->onConnect = function($mqtt) {
-            $mqtt->subscribe('demo');
-        };
-        $mqtt->onMessage = function($topic, $content) {
-            echo "topic:$topic content:$content\n";
-        };
         $mqtt->connect();
+        $connection->mqtt = $mqtt;
     }
 
+    /**
+     * {"topic":"demo","type":"add"}
+     */
+    public function onMessage(TcpConnection $connection, $data)
+    {
 
-}
+        $data = json_decode($data, true);
+        if ($data == null) {
+            $connection->send("参数不能为空");
+            return false;
+        }
+        if (!isset($data['topic']) || $data['topic'] == '') {
+            $connection->send("topic为空");
+            return false;
+        }
+        if (!isset($data['type']) || $data['type'] == '') {
+            $connection->send("type为空");
+            return false;
+        }
+        $topic = $data['topic'];
+        $type = $data['type'];
+        if ($type == 'add') {
+            $res = $connection->mqtt->subscribe($topic);
+            if ($res == null) {
+                $connection->send("添加成功");
+            } else {
+                $connection->send("添加失败");
+            }
+        } else {
+            $res = $connection->mqtt->unsubscribe($topic);
+            if ($res == null) {
+                $connection->send("删除成功");
+            } else {
+                $connection->send("删除失败");
+            }
+        }
+    }
+}

config/process.php

@@ -40,11 +40,18 @@ return [
             ]
         ]
     ],
-    // 'mqtt_push' => [
+
-    //     'handler' => app\mqtt\Publish::class,
+    /**
-    //     'listen'  => 'websocket://0.0.0.0:8955',
+     * 发送
-    //     'count'   => 1,
+     */
-    // ],
+    'mqtt_push' => [
+        'handler' => app\mqtt\Publish::class,
+        'listen'  => 'websocket://0.0.0.0:8955',
+        'count'   => 1,
+    ],
+    /**
+     * 订阅
+     */
     'mqtt_sub' => [
         'handler' => app\mqtt\Subscribe::class,
         'listen'  => 'websocket://0.0.0.0:8956',