Android使用的MQTT客户端

Android使用的MQTT客户端，支持订阅、发送消息；
支持创建连接到本地保存；
支持话题消息筛选；
使用视频：https://dwz.cn/undJFEnq
小米应用商店也有【蘑菇IoT】~
核心代码贴一下，做个记录
import android.app.Service;import android.content.Context;import android.content.Intent;import android.os.IBinder;import androidx.annotation.Nullable;import com.annimon.stream.Collectors;import com.annimon.stream.Stream;import com.freddon.android.snackkit.extension.regex.RegexHelper;import com.freddon.android.snackkit.extension.tools.NetSuit;import com.freddon.android.snackkit.log.Loger;import com.qiniu.util.StringUtils;import com.sagocloud.ntworker.agent.App;import com.sagocloud.ntworker.agent.RxEventBus;import com.sagocloud.ntworker.mqtt.ActionEventType;import com.sagocloud.ntworker.mqtt.EventType;import com.sagocloud.ntworker.mqtt.bean.MQTTConnectUserEntity;import com.sagocloud.ntworker.mqtt.bean.MQTTMessage;import com.sagocloud.ntworker.mqtt.bean.MqttConnectPoint;import com.sagocloud.ntworker.mqtt.event.MQTTClientActionEvent;import com.sagocloud.ntworker.mqtt.event.MQTTTransferEvent;import com.sagocloud.ntworker.mqtt.event.MQTTMessageEvent;import com.sagocloud.ntworker.mqtt.event.MQTTStateEvent;import com.sagocloud.ntworker.mqtt.event.MQTTTraceEvent;import org.eclipse.paho.android.service.MqttAndroidClient;import org.eclipse.paho.android.service.MqttTraceHandler;import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;import org.eclipse.paho.client.mqttv3.IMqttActionListener;import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;import org.eclipse.paho.client.mqttv3.IMqttToken;import org.eclipse.paho.client.mqttv3.MqttCallback;import org.eclipse.paho.client.mqttv3.MqttConnectOptions;import org.eclipse.paho.client.mqttv3.MqttException;import org.eclipse.paho.client.mqttv3.MqttMessage;import java.util.ArrayList;import java.util.Arrays;import java.util.List;import java.util.Locale;import java.util.concurrent.TimeUnit;import io.reactivex.Observable;import io.reactivex.disposables.CompositeDisposable;import io.reactivex.disposables.Disposable;public class MQTTService extends Service {    public final static String CONN = "CON_MQTT_CF";    private MqttAndroidClient mqttAndroidClient;    private MQTTMessageEvent mQTTConnectEvent;    private MQTTConnectUserEntity connectPoint;    private MqttConnectOptions mMqttConnectOptions;    private MqttCallback mqttCallback = new MqttCallback() {        @Override        public void connectionLost(Throwable cause) {            mQTTConnectEvent = new MQTTMessageEvent();            mQTTConnectEvent.setType(EventType.connectionLost);            Loger.e("?connectionLost:", cause.getMessage());            RxEventBus.post(mQTTConnectEvent);            RxEventBus.post(new MQTTStateEvent(App.mqttIsConnected = false));        }        @Override        public void messageArrived(String topic, MqttMessage message) throws Exception {            mQTTConnectEvent = new MQTTMessageEvent();            mQTTConnectEvent.setType(EventType.messageArrived);            mQTTConnectEvent.setTopic(topic);            mQTTConnectEvent.setMessage(message);            Loger.e("✉️messageArrived:", topic);            RxEventBus.post(mQTTConnectEvent);        }        @Override        public void deliveryComplete(IMqttDeliveryToken token) {            mQTTConnectEvent = new MQTTMessageEvent();            mQTTConnectEvent.setType(EventType.deliveryComplete);            try {                mQTTConnectEvent.setTopic(StringUtils.join(token.getTopics(), ","));                mQTTConnectEvent.setMessage(token.getMessage());            } catch (MqttException e) {                e.printStackTrace();            }            Loger.e("?deliveryComplete:", token.toString());            RxEventBus.post(mQTTConnectEvent);        }    };    private IMqttActionListener iMqttActionListener = new IMqttActionListener() {        @Override        public void onSuccess(IMqttToken asyncActionToken) {            RxEventBus.post(new MQTTTransferEvent(asyncActionToken, null));            Loger.e("?onSuccess:", "" + Arrays.toString(asyncActionToken.getTopics()));            App.mqttIsConnected = true;            RxEventBus.post(new MQTTStateEvent(true));            DisconnectedBufferOptions disconnectedBufferOptions = new DisconnectedBufferOptions();            disconnectedBufferOptions.setBufferEnabled(true);            disconnectedBufferOptions.setBufferSize(100);            disconnectedBufferOptions.setPersistBuffer(false);            disconnectedBufferOptions.setDeleteOldestMessages(false);            mqttAndroidClient.setBufferOpts(disconnectedBufferOptions);        }        @Override        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {            RxEventBus.post(new MQTTTransferEvent(asyncActionToken, exception.getMessage()));            Loger.e("?onFailure:", "" + exception.getMessage());            App.mqttIsConnected = false;            RxEventBus.post(new MQTTStateEvent(false));        }    };    private CompositeDisposable subscription;    private MqttTraceHandler traceCallback = new MqttTraceHandler() {        @Override        public void traceDebug(String tag, String message) {            Loger.e("?traceDebug:" + tag, "" + message);//            LiveDataBus.post(MQTTTraceEvent.class,new MQTTTraceEvent(MQTTTraceEvent.Type.DEBUG, tag, message));            RxEventBus.post(new MQTTTraceEvent(MQTTTraceEvent.Type.DEBUG, tag, message));        }        @Override        public void traceError(String tag, String message) {            Loger.e("?traceError:" + tag, "" + message);//            LiveDataBus.post(MQTTTraceEvent.class,new MQTTTraceEvent(MQTTTraceEvent.Type.ERROR, tag, message));            RxEventBus.post(new MQTTTraceEvent(MQTTTraceEvent.Type.ERROR, tag, message));        }        @Override        public void traceException(String tag, String message, Exception e) {            Loger.e("?traceException:" + tag, "" + message + e.getMessage());            RxEventBus.post(new MQTTTraceEvent(MQTTTraceEvent.Type.EXCEPTION, tag, message));//            LiveDataBus.post(MQTTTraceEvent.class,new MQTTTraceEvent(MQTTTraceEvent.Type.EXCEPTION, tag, message));        }    };    private Disposable actSubscription;    private Disposable actTimerSubscription;    public static void startService(Context context, MQTTConnectUserEntity point) {        Intent service = new Intent();        service.setClass(context, MQTTService.class);        service.putExtra(CONN, point);        context.startService(service);    }    private void $prepareActionHandler() {        if (subscription == null) {            subscription = new CompositeDisposable();        }        subscription.clear();        if (actSubscription != null && actSubscription.isDisposed()) {            actSubscription.dispose();        }        if (actTimerSubscription != null && actTimerSubscription.isDisposed()) {            actTimerSubscription.dispose();        }        actSubscription = RxEventBus.subscribeIOEvent(                MQTTClientActionEvent.class,                event -> {                    ActionEventType type = event.getEventType();                    Object payload = event.getPayload();                    switch (type) {                        case connect:                            connect();                            break;                        case publish:                            if (payload instanceof MQTTMessage) {                                publish((MQTTMessage) payload);                            }                            break;                        case subscribe:                            if (payload instanceof MQTTMessage) {                                subscribe((MQTTMessage) payload);                            }                            break;                        case unsubscribe:                            if (payload instanceof MQTTMessage) {                                unsubscribe((MQTTMessage) payload);                            }                            break;                        case unsubscribe_all:                            if (payload instanceof String[]) {                                unsubscribeAll((String[]) payload);                            }                            break;                        case close:                            disconnect();                            mqttAndroidClient = null;                            App.mqttIsConnected = false;                            RxEventBus.post(new MQTTStateEvent(false));                            stopSelf();                            break;                    }                },                error -> {                    Loger.d("error", error.getMessage());                }        );        actTimerSubscription = Observable.interval(2000, TimeUnit.MILLISECONDS)                .subscribe((i) -> {                    App.mqttIsConnected = mqttAndroidClient != null && mqttAndroidClient.isConnected();                    RxEventBus.post(new MQTTStateEvent(App.mqttIsConnected));                });        subscription.add(actSubscription);        subscription.add(actTimerSubscription);    }    @Override    public void onDestroy() {        disconnect();        RxEventBus.unsubscribeEvent(subscription);        super.onDestroy();    }    @Nullable    @Override    public IBinder onBind(Intent intent) {        return null;    }    @Override    public int onStartCommand(Intent intent, int flags, int startId) {        if (intent != null) {            connectPoint = intent.getParcelableExtra(CONN);            if (connectPoint != null) {                $prepareActionHandler();                connect();            } else {                if (mqttAndroidClient != null && mqttAndroidClient.isConnected()) {                    connect();                }            }        }        return super.onStartCommand(intent, flags, startId);    }    private void $prepared(MqttConnectPoint connectPoint) {        String serverURI = String.format(Locale.ENGLISH, "%s://%s:%s", connectPoint.isUseSSL() ? "ssl" : "tcp", connectPoint.getHost(), connectPoint.getPort());        if (mqttAndroidClient != null) {            disconnect();        }        mqttAndroidClient = new MqttAndroidClient(this, serverURI, connectPoint.getClientId());        mqttAndroidClient.setCallback(mqttCallback); //设置监听订阅消息的回调        mqttAndroidClient.setTraceEnabled(true);        mqttAndroidClient.setTraceCallback(traceCallback);        mMqttConnectOptions = new MqttConnectOptions();        mMqttConnectOptions.setMqttVersion(connectPoint.getVersion());        mMqttConnectOptions.setMaxInflight(connectPoint.getMaxInflight());        mMqttConnectOptions.setAutomaticReconnect(connectPoint.isAutoReconnect());        mMqttConnectOptions.setCleanSession(connectPoint.isClearSession()); //设置是否清除缓存        mMqttConnectOptions.setConnectionTimeout(connectPoint.getConnectTimeout()); //设置超时时间，单位：秒        mMqttConnectOptions.setKeepAliveInterval(connectPoint.getTickTime()); //设置心跳包发送间隔，单位：秒        if (RegexHelper.isAllNotEmpty(connectPoint.getUserName(), connectPoint.getUserPasswort())) {            mMqttConnectOptions.setUserName(connectPoint.getUserName()); //设置用户名            mMqttConnectOptions.setPassword(connectPoint.getUserPasswort().toCharArray()); //设置密码        }        if (connectPoint.isUseSSL() && connectPoint.getSslProperties() != null) {            mMqttConnectOptions.setSSLProperties(connectPoint.getSslProperties());        }        if (RegexHelper.isNotEmpty(connectPoint.getLwt())) {            mMqttConnectOptions.setWill(connectPoint.getLwt().getTopic(), connectPoint.getLwt().getMessage().getBytes(), connectPoint.getLwt().getQos(), connectPoint.getLwt().isRetained());        }    }    private void connect() {        if (mqttAndroidClient == null) {            $prepared(connectPoint);        }        if (!mqttAndroidClient.isConnected() && NetSuit.checkEnable(this)) {            try {                mqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);            } catch (MqttException e) {                e.printStackTrace();            }        }    }    private void disconnect() {        try {            if (mqttAndroidClient == null) return;            mqttAndroidClient.unregisterResources();            mqttAndroidClient.disconnect();            mqttAndroidClient.close();        } catch (MqttException e) {            e.printStackTrace();        } finally {            mqttAndroidClient = null;        }    }    private void subscribe(MQTTMessage subscribe) {        try {            if (mqttAndroidClient == null || subscribe == null || subscribe.getTopic() == null)                return;            MqttMessage.validateQos(subscribe.getQos());            List sub = connectPoint.getSubTopics();            if (sub == null) {                sub = new ArrayList<>();            }            Long count = Stream.of(sub)                    .filter(item -> subscribe.getTopic().equalsIgnoreCase(item.getTopic()))                    .collect(Collectors.counting());            if (count > 0) {                return;            }            sub.add(subscribe);            connectPoint.setSubTopics(sub);            mqttAndroidClient.subscribe(subscribe.getTopic(), subscribe.getQos());        } catch (IllegalArgumentException e) {            e.printStackTrace();        } catch (MqttException e) {            e.printStackTrace();        }    }    private void subscribeAll(String[] topics, int[] qos) {        if (RegexHelper.isAnyEmpty(topics, qos)) return;        if (mqttAndroidClient == null) return;        if (topics.length != qos.length) return;        try {            mqttAndroidClient.subscribe(topics, qos);        } catch (MqttException e) {            e.printStackTrace();        }    }    private void unsubscribe(MQTTMessage subscribe) {        try {            if (mqttAndroidClient == null || connectPoint == null) return;            List sub = connectPoint.getSubTopics();            if (sub != null) {                List filtered = Stream.of(sub)                        .filter(item -> !subscribe.getTopic().equalsIgnoreCase(item.getTopic()))                        .collect(Collectors.toList());                connectPoint.setSubTopics(filtered);            }            mqttAndroidClient.unsubscribe(subscribe.getTopic());        } catch (MqttException e) {            e.printStackTrace();        }    }    private void unsubscribeAll(String[] topics) {        try {            if (mqttAndroidClient == null) return;            if (topics == null) mqttAndroidClient.unsubscribe("#");            else {                mqttAndroidClient.unsubscribe(topics);            }            if (connectPoint != null) {                connectPoint.setSubTopics(null);            }        } catch (MqttException e) {            e.printStackTrace();        }    }    private void publish(MQTTMessage subscribe) {        try {            MqttMessage.validateQos(subscribe.getQos());            mqttAndroidClient.publish(subscribe.getTopic(), subscribe.getMessage().getBytes(), subscribe.getQos(), subscribe.isRetained());        } catch (IllegalArgumentException e) {            e.printStackTrace();        } catch (MqttException e) {            e.printStackTrace();        }    }}
更多相关文章

随机推荐
