开发者社区> y2hr4cjzssnlm> 正文

我的mqtt协议和emqttd开源项目个人理解(23) - websocket客户端连接过程分析(Wireshark抓包+源码分析)

简介: 我的mqtt协议和emqttd开源项目个人理解(23) - websocket客户端连接过程分析(Wireshark抓包+源码分析)
+关注继续查看

我们可以使用emq自带的Dashboard插件,进行websocket调试,打开谷歌浏览器输入网址,其中192.168.83.128是emq所在的IP地址:


http://192.168.83.128:18083/#/websocket


用户名:admin,密码:public


WebSocket URI:ws(s)://192.168.83.128:8083/mqtt


TCP URI:tcp://192.168.83.128:1883



一、测试实践


测试环境:使用MQTT 3.1.1版本协议,Wireshark 2.4.1版本,使用wireshark抓包分析(如果是虚拟机,要抓VMnet8虚拟网卡)。


clientid:861694030142473

username:libaineu2004

password:12345678


cleanSession:true


keepalive:60s


1、测试方案1,使用普通的tcp连接,TCP URI:tcp://192.168.83.128:1883,wireshark过滤条件tcp.port == 1883。

(1)、MQTT Connect Command

鼠标点击wireshark的Frame页面,右键菜单,复制,选中树的所有可见项目

MQ Telemetry Transport Protocol, Connect Command

   Header Flags: 0x10 (Connect Command)

   Msg Len: 51

   Protocol Name Length: 4

   Protocol Name: MQTT

   Version: 4

   Connect Flags: 0xc2

   Keep Alive: 60

   Client ID Length: 15

   Client ID: 861694030142473

   User Name Length: 12

   User Name: libaineu2004

   Password Length: 8

   Password: 12345678


右键菜单,复制,字节为HEX + ASCII转储

0000   10 33 00 04 4d 51 54 54 04 c2 00 3c 00 0f 38 36  .3..MQTT...<..86

0010   31 36 39 34 30 33 30 31 34 32 34 37 33 00 0c 6c  1694030142473..l

0020   69 62 61 69 6e 65 75 32 30 30 34 00 08 31 32 33  ibaineu2004..123

0030   34 35 36 37 38                                   45678


(2)、MQTT Connect Ack

MQ Telemetry Transport Protocol, Connect Ack

   Header Flags: 0x20 (Connect Ack)

   Msg Len: 2

   Acknowledge Flags: 0x00

   Return Code: Connection Accepted (0)


0000   20 02 00 00                                       ...



2、测试方案2,使用websocket连接,WebSocket URI:ws(s)://192.168.83.128:8083/mqtt,wireshark过滤条件tcp.port == 8083。我们可以观察到mqtt连接的过程,websocket协议,完整的数据被拆分了好几帧传输。每一帧报文,都有WebSocket和Data字段。我们仅仅需要关注Data字段即可。我们发现,把所有Data字段组合拼接起来,就是和测试方案1完全相同的数据。MQTT Connect Command和Ack都完全相同。说明了即使使用了Websocket方式来传输,仍然遵循mqtt协议。


从本质上来讲,Websocket也是基于 TCP 协议的,同时借用了HTTP的协议来完成一部分握手。

主要解决 HTTP 协议中一个 request 对应一个 response 的尴尬。(http server 不能主动发送消息给 http client) 。通过 HTTP 完成 websocket 的握手过程,接着按照 websocket 协议进行通讯。


websocket 也有他自己的数据帧格式: http://blog.csdn.net/u010487568/article/details/20569027


(1)MQTT Connect Command


0000   10                                               .

0000   33                                               3

0000   00 04                                            ..

0000   4d 51 54 54                                      MQTT

0000   04                                               .

0000   c2                                               .

0000   00 3c                                            .<

0000   00 0f                                            ..

0000   38 36 31 36 39 34 30 33 30 31 34 32 34 37 33     861694030142473

0000   00 0c                                            ..

0000   6c 69 62 61 69 6e 65 75 32 30 30 34              libaineu2004

0000   00 08                                            ..

0000   31 32 33 34 35 36 37 38                          12345678


(2)MQTT Connect Ack


Data (4 bytes)

   Data: 20020000

   [Length: 4]

0000   20 02 00 00                                       ...



二、以下是源码分析,先来看emq v1.1.3版本的源码:


1、-module(emqttd_app).


%%--------------------------------------------------------------------

%% Start Servers

%%--------------------------------------------------------------------

start_servers(Sup) ->

   Servers = [

              {"emqttd wsclient supervisor", {supervisor, emqttd_ws_client_sup}},

%% Start http listener

start_listener({http, ListenOn, Opts}) ->

   mochiweb:start_http(http, ListenOn, Opts, {emqttd_http, handle_request, []});

打开8083端口,开启http服务器。Websocket也是基于 TCP 协议的,同时借用了HTTP的协议来完成一部分握手。



2、-module(emqttd_http).


%%--------------------------------------------------------------------

%% MQTT Over WebSocket

%%--------------------------------------------------------------------

handle_request('GET', "/mqtt", Req) ->

   lager:info("WebSocket Connection from: ~s", [Req:get(peer)]),

   Upgrade = Req:get_header_value("Upgrade"),

   Proto   = Req:get_header_value("Sec-WebSocket-Protocol"),

   case {is_websocket(Upgrade), Proto} of

       {true, "mqtt" ++ _Vsn} ->

           emqttd_ws:handle_request(Req);

       {false, _} ->

           lager:error("Not WebSocket: Upgrade = ~s", [Upgrade]),

           Req:respond({400, [], <<"Bad Request">>});

       {_, Proto} ->

           lager:error("WebSocket with error Protocol: ~s", [Proto]),

           Req:respond({400, [], <<"Bad WebSocket Protocol">>})

   end;

websocket的入口函数。重点关注emqttd_ws:handle_request(Req);



3、-module(emqttd_ws).

ws_loop(Data, State = #wsocket_state{peer = Peer, client_pid = ClientPid,
                                     parser_fun = ParserFun}, ReplyChannel) ->
    ?WSLOG(debug, Peer, "RECV ~p", [Data]),
    case catch ParserFun(iolist_to_binary(Data)) of
        {more, NewParser} ->
            State#wsocket_state{parser_fun = NewParser};
        {ok, Packet, Rest} ->
            gen_server:cast(ClientPid, {received, Packet}),
            ws_loop(Rest, reset_parser(State), ReplyChannel);
        {error, Error} ->
            ?WSLOG(error, Peer, "Frame error: ~p", [Error]),
            exit({shutdown, Error});
        {'EXIT', Reason} ->
            ?WSLOG(error, Peer, "Frame error: ~p", [Reason]),
            ?WSLOG(error, Peer, "Error data: ~p", [Data]),
            exit({shutdown, parser_error})
    end.


gen_server:cast(ClientPid, {received, Packet}), 接收来自客户端的消息。



4、-module(emqttd_ws_client).


handle_cast({received, Packet}, State = #wsclient_state{peer = Peer, proto_state = ProtoState}) ->
    case emqttd_protocol:received(Packet, ProtoState) of
        {ok, ProtoState1} ->
            noreply(State#wsclient_state{proto_state = ProtoState1});
        {error, Error} ->
            ?WSLOG(error, Peer, "Protocol error - ~p", [Error]),
            shutdown(Error, State);
        {error, Error, ProtoState1} ->
            shutdown(Error, State#wsclient_state{proto_state = ProtoState1});
        {stop, Reason, ProtoState1} ->
            stop(Reason, State#wsclient_state{proto_state = ProtoState1})
    end;

emqttd_protocol:received(Packet, ProtoState)。接收处理消息。



5、-module(emqttd_protocol).

process(Packet = ?CONNECT_PACKET(Var), State0) ->
    #mqtt_packet_connect{proto_ver  = ProtoVer,
                         proto_name = ProtoName,
                         username   = Username,
                         password   = Password,
                         clean_sess = CleanSess,
                         keep_alive = KeepAlive,
                         client_id  = ClientId} = Var,
    State1 = State0#proto_state{proto_ver  = ProtoVer,
                                proto_name = ProtoName,
                                username   = Username,
                                client_id  = ClientId,
                                clean_sess = CleanSess,
                                keepalive  = KeepAlive,
                                will_msg   = willmsg(Var),
                                connected_at = os:timestamp()},
    trace(recv, Packet, State1),
    {ReturnCode1, SessPresent, State3} =
    case validate_connect(Var, State1) of


validate_connect(Var, State1),校验clientid,username和password的有效性。


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
冬季实战营第一期:从零到一上手玩转云服务器 | 学习报告
训练营日期:2022年1月17日 - 2022年1月23日 需要先报名后学习,实际是17-21共5天5个场景的学习,链接地址:https://developer.aliyun.com/adc/series/wintercamplist1?spm=a2c6h.26268694.J_3660455150.2.503610f8RnszVx 我是5月份学习的,并没有实时参加训练营,因为觉着本次训练营课程设置很实用,所以对本次学习做个总结。
6 0
手撸一款简单高效的线程池(五)
在之前的内容中,我们给大家介绍了 C++实现线程池过程中的一些常用线优化方案,并分析了不同机制使用时的利弊。这一篇,是线程池系列的最后一章。我们会介绍一下 CGraph 中的 threadpool 如何使用,给出性能对比,并对接下来的工作做一些展望。让我们在线程池性能优化和功能提升的道路上,越走越远。
5 0
记住这两个二分模板,秒杀所有二分查找算法题!
二分查找是一种在每次比较之后将查找空间一分为二的算法。当我们要处理的问题具备单调性,或者要搜寻序列的边界时,应该考虑使用二分查找算法
9 0
Travis CI简介
什么是持续集成? Travis CI 提供的是持续集成服务(Continuous Integration,简称 CI)。它绑定 Github 上面的项目,只要有新的代码,就会自动抓取。然后,提供一个运行环境,执行测试,完成构建,还能部署到服务器。
8 0
Win系统 - 重装系统(微PE、MSDN镜像)【绝对纯净】
Win系统 - 重装系统(微PE、MSDN镜像)【绝对纯净】
10 0
682
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
OceanBase 入门到实战教程
立即下载
阿里云图数据库GDB,加速开启“图智”未来.ppt
立即下载
实时数仓Hologres技术实战一本通2.0版(下)
立即下载