iOS中MQTT和WS的简单混合使用

2024-02-13 10:29:46

历经三天的寻找和尝试，终于完成了一个任务，任务要求的服务器访问用WebSocket的URL格式的，而访问结束后要进行订阅和消息发送，这更多出现在MQTT协议的框架里，例如这个文章

https://www.jianshu.com/p/80ea4507ca74 《iOS MQTT----MQTTClient实战-看这篇的就够了》。但是这个文章的框架访问示例代码的host和端口port设置的方式如下：

-(void)loginMQTT{
    /*设置ip和端口号*/
    
    self.transport.host=_ip;
    self.transport.port=_port;
    
    /*设置MQTT账号和密码*/
    self.mqttSession.transport=self.transport;//给MQTTSession对象设置基本信息
    self.mqttSession.delegate=self;//设置代理
    [self.mqttSession setUserName:_userName];
    [self.mqttSession setPassword:_password];

    //会话链接并设置超时时间
    [self.mqttSession connectAndWaitTimeout:5];
}

该方式的host格式严格是“xxx.xxx.xx.xx”，如果格式错误，会直接访问报错，域名解析错误。报错信息如下：

2021-05-27 09:49:58.630527+0800 Intelligent_fire_protection[28024:9675835] Error Domain=kCFErrorDomainCFNetwork Code=2 "(null)" UserInfo={_kCFStreamErrorDomainKey=12, _kCFStreamErrorCodeKey=8, kCFGetAddrInfoFailureKey=8}

有时也会返回404，报错信息如下

Error Domain=SRWebSocketErrorDomain Code=2132 "received bad response code from server 404" UserInfo={NSLocalizedDescription=received bad response code from server 404, HTTPResponseStatusCode=404}

如果格式包含了path，例如“ws://xxxx/xxxx.xx.xx/mqtt”，需要将 self.transport.host=_ip;和self.transport.port=_port;删去，改成self.transport.url=“ws://xxxx/xxxx.xx.xx/mqtt”。

实现效果我没具体实践，应该不会有问题，起初我没发现这个url属性,host地址又被框架定死没办法和后台统一，访问又一直失败，于是我换了一个网上的demo，在另一个demo里发现了url属性，并完成了我的需求，后来回头查看这个框架的时候，发现也有url属性。爬坑了。

接下来我简单介绍一下我应用的这个框架：

该框架引用来自：https://blog.csdn.net/sinat_23907467/article/details/84840699，《iOS MQTT 简单使用流程》

配置MQTT。 MQTTClient 配置更多是可持续更新，可配置 SSL


pod 'MQTTClient'
pod 'MQTTClient/MinL'
pod 'MQTTClient/ManagerL'
pod 'MQTTClient/WebsocketL'

.h文件中添加文件头和MQTTsession变量


#import <MQTTClient.h>
#import <MQTTWebsocketTransport.h>
#define WEAKSELF   __typeof(&*self) __weak weakSelf = self;



@property (nonatomic, strong) MQTTSession *mySession;

.m文件中添加协议<MQTTSessionDelegate,MQTTSessionManagerDelegate>，定义相应的变量，本文后续代码都在改.m文件中添加

@property(nonatomic,strong)NSString *server_ip;
@property(nonatomic,strong)NSString *userName;
@property(nonatomic,strong)NSString *passWord;
@property(nonatomic,strong)NSArray *topics;

文章中用到了下面的代码，但我在应用时候链接服务器报错了，我便把这部分代码注释掉了，实际上这部分代码只是避免持续等待连接卡死的情况，当连接时间超过某一个时间长度就强制断开申请连接。后文中我另加了一行代码实现这个功能。

        MQTTCFSocketTransport *transport = [[MQTTCFSocketTransport alloc] init];
        transport.host = @"localhost";
        transport.port = 1883;
        MQTTSession *session = [[MQTTSession alloc] init];
        session.transport = transport;
    	session.delegate = self;
    	[session connectAndWaitTimeout:30];

WebSocket的初始化和链接代码：

引用的文章中用的是host和port，但这个不符合web的地址格式，往往web会有一个path路径，所以访问会报错，错误代码是5，MQTTSessionEventConnectionClosedByBroker。所以还需要在代码中添加一行transport.path=@"ws";或者将post和port两行注释掉，直接手动输入一个url 即 NSURL *url=[NSURL URLWithString:_server_ip]; transport.url=url; 其中_server_ip的格式为“ws://xxxx.xxxx.xx.xx/ws”

另外，上文中我注释掉的代码在此处实现了，即 [session connectAndWaitTimeout:10];

引用的文章中有个 [self reconnect];代码，实际是重新连接功能，实际使用时候文章没提供，已经没有这个函数了，所以我用的[self.mySession connect];代替。

    WEAKSELF
//    NSString *myId=[NSString stringWithFormat:@"%d",arc4random()%10000];
    NSString *clientID =[NSString stringWithFormat:@"%@|iOS|%@",[[NSBundle mainBundle] bundleIdentifier],[UIDevice currentDevice].identifierForVendor.UUIDString];;
    MQTTWebsocketTransport *transport = [[MQTTWebsocketTransport alloc] init];
//    transport.host = [NSString stringWithFormat:@"%@",server_ip];
//    transport.port =  9999;  // 端口号
    NSURL *url=[NSURL URLWithString:_server_ip];
    transport.url=url;
//    transport.path=@"ws";
    transport.tls = YES; //  根据需要配置  YES 开起 SSL 验证 此处为单向验证 双向验证 根据SDK 提供方法直接添加
    MQTTSession *session = [[MQTTSession alloc] init];
    NSString *linkUserName = _userName;
    NSString *linkPassWord = _passWord;
    [session setUserName:linkUserName];
    [session setClientId:clientID];
    [session setPassword:linkPassWord];
    [session setKeepAliveInterval:5];
    session.transport = transport;
    session.delegate = self;
    [session connectAndWaitTimeout:10];
    self.mySession = session;
    [self.mySession connect];//self reconnect
    [self.mySession addObserver:self forKeyPath:@"state" options:NSKeyValueObservingOptionNew|NSKeyValueObservingOptionOld context:nil]; //添加事件监听

WebSocket 监听，响应事件：

- (void)observeValueForKeyPath:(NSString *)keyPath ofObject:(id)object change:(NSDictionary *)change context:(void *)context {
    switch (self.mySession.status) {
        case MQTTSessionManagerStateClosed:
        NSLog(@"连接已经关闭");
        break;
        case MQTTSessionManagerStateClosing:
        NSLog(@"连接正在关闭");
        break;
        case MQTTSessionManagerStateConnected:
        NSLog(@"已经连接");
        break;
        case MQTTSessionManagerStateConnecting:
        NSLog(@"正在连接中");
        
        break;
        case MQTTSessionManagerStateError: {
            //            NSString *errorCode = self.mySession.lastErrorCode.localizedDescription;
            NSString *errorCode = self.mySession.description;
            NSLog(@"连接异常 ----- %@",errorCode);
        }
        break;
        case MQTTSessionManagerStateStarting:
        NSLog(@"开始连接");
        break;
        default:
        break;
    }
}

`session Delegate 协议`

连接返回状态

根据我的项目需求，我直接在订阅成功后直接对topic进行了订阅，即代码中for循环，读者可以自行更改订阅逻辑。

-(void)handleEvent:(MQTTSession *)session event:(MQTTSessionEvent)eventCode error:(NSError *)error{
    if (eventCode == MQTTSessionEventConnected) {
        NSLog(@"链接MQTT 成功");
        
        // 方法 封装 可外部调用
        for (NSString *topic in _topics) {
            [session subscribeToTopic:topic atLevel:2 subscribeHandler:^(NSError *error, NSArray<NSNumber *> *gQoss){
                if (error) {
                    NSLog(@"Subscription failed %@", error.localizedDescription);
                } else {
                    NSLog(@"Subscription sucessfull! Granted Qos: %@", gQoss);

    //                [self send];
                }
             }];
        }
 // this is part of the block API

    }else if (eventCode == MQTTSessionEventConnectionRefused) {
            NSLog(@"MQTT拒绝链接");
   }else if (eventCode == MQTTSessionEventConnectionClosed){
            NSLog(@"MQTT链接关闭");
  }else if (eventCode == MQTTSessionEventConnectionError){
            NSLog(@"MQTT 链接错误");
  }else if (eventCode == MQTTSessionEventProtocolError){
            NSLog(@"MQTT 不可接受的协议");
  }else{//MQTTSessionEventConnectionClosedByBroker
            NSLog(@"MQTT链接 其他错误");
  }
   if (error) {
        NSLog(@"链接报错  -- %@",error);
   }
}

收到来自服务器的信息，并对改信息进行相应处理

-(void)newMessage:(MQTTSession *)session data:(NSData *)data onTopic:(NSString *)topic qos:(MQTTQosLevel)qos retained:(BOOL)retained mid:(unsigned int)mid
{
    NSDictionary *dic = [NSJSONSerialization JSONObjectWithData:data options:NSJSONReadingMutableContainers error:nil];
//    NSLog(@"EasyMqttService mqtt connect success  %@",dic);
    // 做相对应的操作
}

取消订阅，断开与服务器的链接：

-(void)closeMQTTClient{
    [self.mySession disconnect];
    [self.mySession unsubscribeTopics:@[@"已经订阅的主题"] unsubscribeHandler:^(NSError *error) {
        if (error) {
            NSLog(@"取消订阅失败");
        }else{
            NSLog(@"取消订阅成功");
        }
    }];
}

以下内容是我封装了该框架，并留了三个方法用于链接，发送数据和断开连接。读者自行在.h文件中添加对应的方法：

-(void)connectServer:(NSString *)urlString userName:(NSString *)userName passWord:(NSString *)passWord topic:(NSArray *)topics{
    _server_ip=urlString;
    _userName=userName;
    _passWord=passWord;
    _topics=topics;
    [self websocket];
   
}
-(void)disConnectServer{
    [_mySession closeAndWait:1];
    self.mySession.delegate=nil;//代理
    _mySession=nil;
//    _transport=nil;//连接服务器属性
    _server_ip=nil;//服务器ip地址
//    _port=0;//服务器ip地址
    _userName=nil;//用户名
    _passWord=nil;//密码
//    _topic=nil;//单个主题订阅
    _topics=nil;//多个主题订阅
}
-(void)sendMassage:(NSData *)msg topic:(NSString *)topic{

    [self.mySession publishData:msg onTopic:topic retain:NO qos:MQTTQosLevelExactlyOnce publishHandler:^(NSError *error) {
        if (error) {
            NSLog(@"发送失败 - %@",error);

            
        }else{
            NSLog(@"发送成功");
           
            
        }
    }];
}

码农公寓

session Delegate 协议

相关文章

`session Delegate 协议`