在本文中,将使用MQTT通信协议的方式实现APP远程控制(外网)智能插座。
免费的MQTT服务器
首先推荐两个免费的MQTT服务器:
1)https://www.cloudmqtt.com
2)https://yunba.io/
云巴推送功能很强大,支持离线消息和群组推送等功能。可惜的是云巴推送对MQTT协议进行了进一步的包装,目前我还没有办法将它应用到ESP8266的远程通信中。
基本原理
智能插座连接到家庭路由器中以便连接上外网的MQTT服务器。Android 手机通过WIFI或4G流量连接到MQTT服务器。APP和ESP8266采用发布和订阅的机制进行通信。APP向ESP8266发送控制命令。ESP8266接收到控制命令后,执行相应的操作并返回结果。
通信帧格式
通信帧格式与上一小节一致。
Topic
//ESP8266订阅该Topic,APP向该Topic发送消息
const String TOPIC_REQUEST = "/WIFISocket/Request";
//APP订阅该Topic,ESP8266向该Topic发送消息
const String TOPIC_RESPONSE = "/WIFISocket/Response";
实现过程
源代码目前已开源:https://git.coding.net/zimengyu1992/WIFISocketProject.git
设备端
1)采用了Nick O’Leary提供的PubSubClient库,用于MQTT通信。
PubSubClient.h - A simple client for MQTT.
Nick O'Leary
http://knolleary.net
核心源码如下:
#ifndef __WIFISOCKETMQTTSERVER_H__
#define __WIFISOCKETMQTTSERVER_H__
#include <Arduino.h>
#include <WiFiUdp.h>
#include <ESP8266WiFi.h>
#include "PubSubClient.h"
class WIFISocketMqttServerClass
{
private:
WiFiClient wifiClient;
PubSubClient mqttClient;
private:
static void callback(char* topic, uint8_t* payload, unsigned int length);
public:
void begin();
void loop();
};
extern WIFISocketMqttServerClass WIFISocketMqttServer;
#endif
#include <ArduinoJson.h>
#include "WIFISocketSwitch.h"
#include "WIFISocketMqttServer.h"
WIFISocketMqttServerClass WIFISocketMqttServer;
const String TOPIC_REQUEST = "/WIFISocket/Request";
const String TOPIC_RESPONSE = "/WIFISocket/Response";
const int STATUS_OK = 200;
const int STATUS_ERR = 304;
const int MESSAGE_GETSWITCHSTATE_REQUEST = 1000;
const int MESSAGE_GETSWITCHSTATE_RESPONSE = 1001;
const int MESSAGE_SETSWITCHSTATE_REQUEST = 1002;
const int MESSAGE_SETSWITCHSTATE_RESPONSE = 1003;
void WIFISocketMqttServerClass::callback(char* topic, uint8_t* payload, unsigned int length)
{
const int PACKET_MAXSIZE = 128;
uint8_t packet[PACKET_MAXSIZE];
memset(packet, 0, PACKET_MAXSIZE*sizeof(uint8_t));
memcpy(packet, payload, length);
Serial.println("MQTT: New Message Coming");
Serial.println((const char *)topic);
Serial.println((const char *)packet);
DynamicJsonBuffer jsonBuffer(PACKET_MAXSIZE);
JsonObject &root = jsonBuffer.parseObject(packet);
if (!root.success())
{
return;
}
if (!root.containsKey("Cmd"))
{
return;
}
int cmd = (int)root["Cmd"];
if (cmd == MESSAGE_GETSWITCHSTATE_REQUEST)
{
jsonBuffer.clear();
JsonObject& Root = jsonBuffer.createObject();
Root["Cmd"] = MESSAGE_GETSWITCHSTATE_RESPONSE;
Root["Status"] = STATUS_OK;
Root["SwitchState"] = WIFISocketSwitch.getState();
memset(packet, 0, PACKET_MAXSIZE);
size_t length = Root.printTo((char *)packet, PACKET_MAXSIZE);
WIFISocketMqttServer.mqttClient.publish(TOPIC_RESPONSE.c_str(), packet, length);
}
else if (cmd == MESSAGE_SETSWITCHSTATE_REQUEST)
{
boolean Successed = false;
if (root.containsKey("SwitchState"))
{
boolean SwitchState = (boolean)root["SwitchState"];
WIFISocketSwitch.Switch(SwitchState);
Successed = true;
}
jsonBuffer.clear();
JsonObject& Root = jsonBuffer.createObject();
Root["Cmd"] = MESSAGE_SETSWITCHSTATE_RESPONSE;
Root["Status"] = Successed ? STATUS_OK : STATUS_ERR;
memset(packet, 0, PACKET_MAXSIZE);
size_t length = Root.printTo((char *)packet, PACKET_MAXSIZE);
WIFISocketMqttServer.mqttClient.publish(TOPIC_RESPONSE.c_str(), packet, length);
}
}
void WIFISocketMqttServerClass::begin()
{
mqttClient.setClient(wifiClient);
}
void WIFISocketMqttServerClass::loop()
{
if (WiFi.isConnected())
{
if (!mqttClient.connected())
{
IPAddress MqttServerIp;
MqttServerIp.fromString("34.235.122.116");
//WiFi.hostByName("m14.cloudmqtt.com", MqttServerIp);
mqttClient.setServer(MqttServerIp, 11672);
mqttClient.setCallback(this->callback);
if (mqttClient.connect("WIFISocketDevice","******","******"))
{
Serial.println("MQTT: Connect to Server Success...");
if (mqttClient.subscribe(TOPIC_REQUEST.c_str()))
{
Serial.printf("MQTT: Subscribe Topic : %s Success...\n", TOPIC_REQUEST.c_str());
}
}
}
mqttClient.loop();
}
}
APP端
1)使用了eclipse提供的MQTT Client库:
org.eclipse.paho.client.mqttv3-1.0.2.jar
2)为了不阻塞UI线程,创建了MQTT收发的独立线程。
3)采用Message&Handler的方式,进行UI线程通信。
4)加入了通信超时定时器,提供数据发送超时的反馈。
核心源码如下:
package site.webhome.wifisocketliteapp;
import android.content.Context;
import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import android.util.Log;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class MqttConnection extends Thread {
private final static String TAG = MqttConnection.class.getSimpleName();
private Handler mHandler = null;
private MqttClient mClient = null;
private Semaphore mSemaphore = null;
private BlockingQueue<String> mQueue = null;
private Context mContext = null;
private boolean mRequesting = false;
private static final String TOPIC_REQUEST = "/WIFISocket/Request";
private static final String TOPIC_RESPONSE = "/WIFISocket/Response";
private static final String MQTT_SERVER = "tcp://m14.cloudmqtt.com:11672";
private static final String MQTT_USER = "******";
private static final String MQTT_PASS = "******";
private static final String MQTT_CLIENTID = "WIFISocketApp";
public MqttConnection(Context context, Handler handler) throws MqttException {
mContext = context;
mHandler = handler;
mQueue = new LinkedBlockingQueue<String>();
mSemaphore = new Semaphore(0);
mClient = new MqttClient(MQTT_SERVER, MQTT_CLIENTID, new MemoryPersistence());
this.start();
}
public void run() {
while (true) {
try {
mSemaphore.acquire();
if (!mQueue.isEmpty()) {
String sendMessage = mQueue.poll();
connect();
MqttMessage mqttMessage = new MqttMessage();
mqttMessage.setQos(1);
mqttMessage.setRetained(true);
mqttMessage.setPayload(sendMessage.getBytes("UTF-8"));
mClient.publish(TOPIC_REQUEST, mqttMessage);
mRequesting = true;
if (mHandler != null) {
mHandler.removeCallbacks(mTimeOutRunnable);
mHandler.postDelayed(mTimeOutRunnable, 3000);
}
}
} catch (Exception e) {
e.printStackTrace();
mRequesting = false;
if (mHandler != null) {
Message responseMessage = new Message();
responseMessage.what = StatusCode.MESSAGE_TIMEOUT;
mHandler.sendMessage(responseMessage);
}
}
}
}
private void connect() throws MqttException {
if (!mClient.isConnected()) {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
options.setUserName(MQTT_USER);
options.setPassword(MQTT_PASS.toCharArray());
// 设置超时时间
options.setConnectionTimeout(10);
// 设置会话心跳时间
options.setKeepAliveInterval(20);
mClient.setCallback(new PushCallback());
mClient.connect(options);
mClient.subscribe(TOPIC_RESPONSE);
}
}
public void send(String Message) throws InterruptedException {
mQueue.put(Message);
mSemaphore.release();
}
private final Runnable mTimeOutRunnable = new Runnable() {
@Override
public void run() {
if (mRequesting) {
if (mHandler != null) {
Message responseMessage = new Message();
responseMessage.what = StatusCode.MESSAGE_TIMEOUT;
mHandler.sendMessage(responseMessage);
}
}
}
};
public class PushCallback implements MqttCallback {
public void connectionLost(Throwable cause) {
mRequesting = false;
if (mHandler != null) {
Message responseMessage = new Message();
responseMessage.what = StatusCode.MESSAGE_TIMEOUT;
mHandler.sendMessage(responseMessage);
}
}
public void deliveryComplete(IMqttDeliveryToken token) {
mRequesting = false;
}
public void messageArrived(String topic, MqttMessage message) throws Exception {
if (topic.equals(TOPIC_RESPONSE)) {
mRequesting = false;
if (mHandler != null) {
Message responseMessage = new Message();
responseMessage.what = StatusCode.MESSAGE_ARRIVE;
Bundle data = new Bundle();
data.putString("MESSAGE", new String(message.getPayload()));
responseMessage.setData(data);
mHandler.sendMessage(responseMessage);
}
}
}
}
}
package site.webhome.wifisocketliteapp;
import android.os.Handler;
import android.os.Message;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.view.View;
import android.view.Window;
import android.view.WindowManager;
import android.widget.Button;
import android.widget.RelativeLayout;
import android.widget.Toast;
import org.json.JSONException;
import org.json.JSONObject;
public class MainActivity extends AppCompatActivity {
private Button switchButton = null;
private boolean switchState = false;
private boolean setSwitchState = false;
private boolean switchStateInited = false;
private boolean requesting = false;
private MqttConnection mqttConnection = null;
public Handler MainHandler = new Handler() {
@Override
public void handleMessage(Message msg) {
if (msg.what == StatusCode.MESSAGE_ARRIVE) {
try {
String MessageText = msg.getData().getString("MESSAGE");
JSONObject Json = new JSONObject(MessageText);
int Status = Json.getInt("Status");
if (Status == StatusCode.STATUS_ERR) {
Toast.makeText(getApplicationContext(), "参数错误!", Toast.LENGTH_SHORT).show();
} else {
int Cmd = Json.getInt("Cmd");
if (Cmd == StatusCode.MESSAGE_GETSWITCHSTATE_RESPONSE) {
switchStateInited = true;
switchState = Json.getBoolean("SwitchState");
ChangeBackground(switchState);
}
if (Cmd == StatusCode.MESSAGE_SETSWITCHSTATE_RESPONSE) {
switchState = setSwitchState;
ChangeBackground(switchState);
Toast.makeText(getApplicationContext(), "操作成功!", Toast.LENGTH_SHORT).show();
}
}
} catch (JSONException e) {
e.printStackTrace();
}
}
if (msg.what == StatusCode.MESSAGE_TIMEOUT) {
Toast.makeText(getApplicationContext(), "网络超时!", Toast.LENGTH_SHORT).show();
}
requesting = false;
}
};
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
switchButton = (Button) findViewById(R.id.switch_button_device);
switchButton.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
SwitchButtonOnClick();
}
});
ChangeBackground(switchState);
InitStatus();
}
private void InitStatus() {
try {
if (mqttConnection == null) {
mqttConnection = new MqttConnection(getApplicationContext(), MainHandler);
}
SendGetSwitchState();
} catch (Exception e) {
Toast.makeText(getApplicationContext(), "网络超时!", Toast.LENGTH_SHORT).show();
}
}
private void SwitchButtonOnClick() {
if (requesting) {
Toast.makeText(getApplicationContext(), "操作未完成!", Toast.LENGTH_SHORT).show();
return;
}
if (!switchStateInited) {
Toast.makeText(getApplicationContext(), "正在初始化!", Toast.LENGTH_SHORT).show();
InitStatus();
return;
}
try {
setSwitchState = !switchState;
SendSetSwitchState(setSwitchState);
} catch (Exception e) {
Toast.makeText(getApplicationContext(), "网络超时!", Toast.LENGTH_SHORT).show();
}
}
private void SendGetSwitchState() throws JSONException, InterruptedException {
JSONObject json = new JSONObject();
json.put("Cmd", StatusCode.MESSAGE_GETSWITCHSTATE_REQUEST);
mqttConnection.send(json.toString());
}
private void SendSetSwitchState(boolean state) throws JSONException, InterruptedException {
JSONObject json = new JSONObject();
json.put("Cmd", StatusCode.MESSAGE_SETSWITCHSTATE_REQUEST);
json.put("SwitchState", state);
mqttConnection.send(json.toString());
}
private void ChangeBackground(boolean switchState) {
int statusBarRes = R.color.colorBluePrimaryDark;
int backgroundRes = R.drawable.switch_on_bg;
if (!switchState) {
backgroundRes = R.drawable.switch_off_bg;
statusBarRes = R.color.colorBlackPrimaryDark;
}
RelativeLayout relativeLayout = (RelativeLayout) findViewById(R.id.layout_main_device);
relativeLayout.setBackgroundResource(backgroundRes);
Window window = this.getWindow();
window.addFlags(WindowManager.LayoutParams.FLAG_DRAWS_SYSTEM_BAR_BACKGROUNDS);
window.setStatusBarColor(this.getResources().getColor(statusBarRes));
}
}