beetlex-io/mqtt

high performance .NET library for MQTT based communication support v3.x and v5.0 protocols

MQTT

介绍

基于BeetleX组件扩展的高性能MQTT通讯网关，可以轻松应对数十万的消费订阅转发.组件实现了3.x和5.0版本的协议。

运行服务

    class Program
    {
        private static MQTTServer mServer;
        static void Main(string[] args)
        {

            mServer = new MQTTServer(ProtocolType.V3);
            mServer.RegisterComponent<BeetleX.MQTT.Server.Controller>();
            mServer.MQTTListen(o =>
            {
                o.DefaultListen.Port = 8089;//mqtt服务端口
            })
            .Setting(o =>
            {
                o.LogToConsole = true;
                o.Port = 80;//web管理端口
                o.LogLevel = EventArgs.LogType.Info;
            })
            .UseJWT()
            .UseEFCore<Storages.MQTTDB>()
            .UseElement(PageStyle.ElementDashboard)
            .Initialize((http, vue, resoure) =>
            {
                resoure.AddAssemblies(typeof(BeetleX.MQTT.Server.MQTTUser).Assembly);
                resoure.AddCss("website.css");
                resoure.AddScript("echarts.js");
                vue.Debug();
            })
           .Run();

        }
    }

运行成功后可以通过浏览器访问 http://localhost 进入管理界面（如果电脑上已经有服务占用了80端口，则根据自己需要调整）。

单独使用协议包

BeetleX.MQTT.Protocols是MQTT的协议分析组件，它是基于Stream来处理MQTT协议，如果不想使用BeetleX配套网服务功能可以单独引用BeetleX.MQTT.Protocols来进行MQTT的协议分析.组件是支持V3.X和V5.0

//v5
var mqttparse = new BeetleX.MQTT.Protocols.V5.MQTTParseV5();
mqttparse.Read(stream, null);
mqttparse.Write(msg, stream, null);
//v3.x
var mqttparse = new BeetleX.MQTT.MQTTParseV3();
mqttparse.Read(stream, null);
mqttparse.Write(msg, stream, null);

主界面

帐号界面

设备查看界面

mqtt's Issues

MQTTClientPacket导致空异常

//MQTTParse中 Write函对ISession参数做了一些处理,但是MQTTClientPacke中调用直接传空,导致发送时异常,请问MQTTClientPacke中如何才能拿到ISession解决此报错呢
public void Write(MQTTMessage msg, Stream stream, ISession session)
{
IServer server = session.Server;
if (server.EnableLog(LogType.Debug))
{
server.Log(LogType.Debug, session, $"write mqtt message {msg}");
}

        MemoryStream protocolStream = GetProtocolStream();
        int num = 0;
        num |= (int)msg.Type << 4;
        if (msg.DUP)
        {
            num |= 8;
        }

        num |= (int)msg.QoS << 1;
        if (msg.Retain)
        {
            num |= 1;
        }

        stream.WriteByte((byte)num);
        msg.Write(protocolStream, session);
        if (server.EnableLog(LogType.Debug))
        {
            server.Log(LogType.Debug, session, $"write mqtt message body size {protocolStream.Length}");
        }

        mInt7BitHandler.Write(stream, (int)protocolStream.Length);
        protocolStream.Position = 0L;
        protocolStream.CopyTo(stream);
        if (server.EnableLog(LogType.Debug))
        {
            server.Log(LogType.Debug, session, "write mqtt message success");
        }
    }
}

MQTTClientPacket中加密部分中：
public void Encode(object data, IClient client, Stream stream)
{
MQTTMessage mQTTMessage = data as MQTTMessage;
if (mQTTMessage != null)
{
mMQTTParse.Write(mQTTMessage, stream.ToPipeStream(), null);
return;
}

        throw new BXException("Invalid message type!");
    }