Han Yan / Mbed OS ms2b

not running

Dependencies:   TextLCD MQTT

Communication.cpp@1:19c3299ea83a, 2019-12-10 (annotated)

Committer:
hyan99
Date:
Tue Dec 10 23:40:25 2019 +0000
Revision:
1:19c3299ea83a
Parent:
0:3b4906b8a747
Child:
2:16b3bd337db2
cannot subscribe outside in main

Who changed what in which revision?

UserRevisionLine numberNew contents of line
hyan99 0:3b4906b8a747 1 #include "Communication.h"
hyan99 0:3b4906b8a747 2
hyan99 0:3b4906b8a747 3 Communication* Communication::singleton = NULL;
hyan99 0:3b4906b8a747 4 int Communication::speeds[5] = {0, 0, 0, 0, 0};
hyan99 0:3b4906b8a747 5 int Communication::sync = 0;
hyan99 0:3b4906b8a747 6 int Communication::flags[5] = {0x01, 0x02, 0x04, 0x08, 0x10};
hyan99 0:3b4906b8a747 7 EventFlags Communication::control_flags;
hyan99 0:3b4906b8a747 8 EventFlags Communication::sync_flags;
hyan99 0:3b4906b8a747 9
hyan99 0:3b4906b8a747 10 Serial pc_comm(USBTX, USBRX);
hyan99 0:3b4906b8a747 11
hyan99 0:3b4906b8a747 12 Communication* Communication::getInstance() {
hyan99 0:3b4906b8a747 13 if (singleton == NULL) {
hyan99 1:19c3299ea83a 14 singleton = new Communication();
hyan99 0:3b4906b8a747 15 }
hyan99 0:3b4906b8a747 16 return singleton;
hyan99 0:3b4906b8a747 17 }
hyan99 0:3b4906b8a747 18
hyan99 0:3b4906b8a747 19 Communication::Communication()
hyan99 0:3b4906b8a747 20 {
hyan99 1:19c3299ea83a 21 // singleton = this;
hyan99 0:3b4906b8a747 22 setup_wifi();
hyan99 0:3b4906b8a747 23 if (wifi == NULL) {
hyan99 0:3b4906b8a747 24 pc_comm.printf("Failed to set up Wifi.");
hyan99 0:3b4906b8a747 25 }
hyan99 0:3b4906b8a747 26 MQTTNetwork network(wifi);
hyan99 0:3b4906b8a747 27 setup_mqtt(network);
hyan99 0:3b4906b8a747 28 if (client == NULL) {
hyan99 0:3b4906b8a747 29 pc_comm.printf("Failed to set up Client.");
hyan99 0:3b4906b8a747 30 }
hyan99 1:19c3299ea83a 31 subscribe_to_topic_sync("Rahman/Sync/Send/1");
hyan99 0:3b4906b8a747 32 }
hyan99 0:3b4906b8a747 33
hyan99 0:3b4906b8a747 34 void Communication::setup_wifi() {
hyan99 0:3b4906b8a747 35 // Get a handle to the WiFi module
hyan99 0:3b4906b8a747 36 wifi = WiFiInterface::get_default_instance();
hyan99 0:3b4906b8a747 37
hyan99 0:3b4906b8a747 38 // Connect the module to the wifi, based on the SSID and password
hyan99 0:3b4906b8a747 39 // specified in the mbed_app.json configuration file
hyan99 0:3b4906b8a747 40 // If you are using AirPennNet-Device, this will not succeed until the MAC
hyan99 0:3b4906b8a747 41 // address (printed shortly after this) is registered
hyan99 0:3b4906b8a747 42 pc_comm.printf("Connecting to wifi\r\n");
hyan99 0:3b4906b8a747 43 int rc = wifi->connect(MBED_CONF_APP_WIFI_SSID, MBED_CONF_APP_WIFI_PASSWORD, NSAPI_SECURITY_WPA_WPA2);
hyan99 1:19c3299ea83a 44 pc_comm.printf("Finished connected to wifi\r\n");
hyan99 0:3b4906b8a747 45 // Print out the MAC address of the wifi module. The MAC address is
hyan99 0:3b4906b8a747 46 // needed to register the device with AirPennNet-Device, so that you
hyan99 0:3b4906b8a747 47 // can use the campus wifi
hyan99 0:3b4906b8a747 48 pc_comm.printf("MAC Address: ");
hyan99 0:3b4906b8a747 49 pc_comm.printf(wifi->get_mac_address());
hyan99 0:3b4906b8a747 50 pc_comm.printf("\r\n");
hyan99 0:3b4906b8a747 51
hyan99 0:3b4906b8a747 52 if (rc != 0) {
hyan99 0:3b4906b8a747 53 pc_comm.printf("Problem connecting to wifi\r\n");
hyan99 0:3b4906b8a747 54 return;
hyan99 0:3b4906b8a747 55 } else {
hyan99 0:3b4906b8a747 56 pc_comm.printf("Wifi connected\r\n");
hyan99 0:3b4906b8a747 57 }
hyan99 0:3b4906b8a747 58 }
hyan99 0:3b4906b8a747 59
hyan99 0:3b4906b8a747 60
hyan99 0:3b4906b8a747 61 void Communication::setup_mqtt(MQTTNetwork &network) {
hyan99 0:3b4906b8a747 62 // the hostname and port point to a Google Cloud MQTT server we setup for
hyan99 0:3b4906b8a747 63 // this project
hyan99 0:3b4906b8a747 64 const char* hostname = "34.68.206.11";
hyan99 0:3b4906b8a747 65 int port = 1883;
hyan99 0:3b4906b8a747 66
hyan99 0:3b4906b8a747 67 // Create the underlying network connection to the MQTT server
hyan99 0:3b4906b8a747 68 pc_comm.printf("Connecting to %s:%d\r\n", hostname, port);
hyan99 0:3b4906b8a747 69 int rc = network.connect(hostname, port);
hyan99 0:3b4906b8a747 70 if (rc != 0) {
hyan99 0:3b4906b8a747 71 pc_comm.printf("There was an error with the TCP connect: %d\r\n", rc);
hyan99 0:3b4906b8a747 72 return;
hyan99 0:3b4906b8a747 73 }
hyan99 0:3b4906b8a747 74
hyan99 0:3b4906b8a747 75 pc_comm.printf("Connected to %s:%d\r\n", hostname, port);
hyan99 0:3b4906b8a747 76
hyan99 0:3b4906b8a747 77 // Connect the MQTT client to the server
hyan99 0:3b4906b8a747 78 client = new MQTT::Client<MQTTNetwork, Countdown>(network);
hyan99 0:3b4906b8a747 79 rc = client->connect();
hyan99 0:3b4906b8a747 80 if (rc != 0) {
hyan99 0:3b4906b8a747 81 pc_comm.printf("There was an error with the MQTT connect: %d\r\n", rc);
hyan99 0:3b4906b8a747 82 return;
hyan99 0:3b4906b8a747 83 }
hyan99 0:3b4906b8a747 84
hyan99 0:3b4906b8a747 85 pc_comm.printf("MQTT connect successful!\r\n");
hyan99 0:3b4906b8a747 86 }
hyan99 0:3b4906b8a747 87
hyan99 0:3b4906b8a747 88 void Communication::message_arrived(MQTT::MessageData& md)
hyan99 0:3b4906b8a747 89 {
hyan99 0:3b4906b8a747 90 MQTT::Message &message = md.message;
hyan99 0:3b4906b8a747 91 pc_comm.printf("Message arrived: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id);
hyan99 0:3b4906b8a747 92 pc_comm.printf("Payload %.*s\r\n", message.payloadlen, (char*)message.payload);
hyan99 0:3b4906b8a747 93 }
hyan99 0:3b4906b8a747 94
hyan99 0:3b4906b8a747 95 void Communication::control_arrived(MQTT::MessageData& md) {
hyan99 0:3b4906b8a747 96 MQTT::Message &message = md.message;
hyan99 0:3b4906b8a747 97
hyan99 0:3b4906b8a747 98 char* payload = (char*) message.payload;
hyan99 0:3b4906b8a747 99
hyan99 0:3b4906b8a747 100 int id = (int) payload[0];
hyan99 0:3b4906b8a747 101 int speed = (int) payload[1];
hyan99 0:3b4906b8a747 102
hyan99 0:3b4906b8a747 103 speeds[id-1] = speed;
hyan99 0:3b4906b8a747 104 control_flags.set(flags[id-1]);
hyan99 0:3b4906b8a747 105 }
hyan99 0:3b4906b8a747 106
hyan99 0:3b4906b8a747 107 void Communication::sync_arrived(MQTT::MessageData& md) {
hyan99 0:3b4906b8a747 108 MQTT::Message &message = md.message;
hyan99 0:3b4906b8a747 109
hyan99 0:3b4906b8a747 110 char* payload = (char*) message.payload;
hyan99 0:3b4906b8a747 111 sync = (int) payload[0];
hyan99 0:3b4906b8a747 112 sync_flags.set(0x01);
hyan99 0:3b4906b8a747 113 }
hyan99 0:3b4906b8a747 114
hyan99 0:3b4906b8a747 115 int Communication::send_message(char* topic) {
hyan99 0:3b4906b8a747 116 pc_comm.printf("Sending a message!\r\n");
hyan99 0:3b4906b8a747 117 MQTT::Message message;
hyan99 0:3b4906b8a747 118
hyan99 0:3b4906b8a747 119 char buf[100];
hyan99 0:3b4906b8a747 120 sprintf(buf, "Hello World! This is a test message!\r\n");
hyan99 0:3b4906b8a747 121 int rc = client->publish(topic, (char*) buf, strlen(buf)+1, MQTT::QOS1);
hyan99 0:3b4906b8a747 122
hyan99 0:3b4906b8a747 123 if (rc != 0) {
hyan99 0:3b4906b8a747 124 return -1;
hyan99 0:3b4906b8a747 125 } else {
hyan99 0:3b4906b8a747 126 pc_comm.printf("Message sent!\r\n");
hyan99 0:3b4906b8a747 127 return 0;
hyan99 0:3b4906b8a747 128 }
hyan99 0:3b4906b8a747 129 }
hyan99 0:3b4906b8a747 130
hyan99 0:3b4906b8a747 131 int Communication::subscribe_to_topic_position(char* topic) {
hyan99 1:19c3299ea83a 132 pc_comm.printf("subcribing: %s", topic);
hyan99 0:3b4906b8a747 133 int rc = client->subscribe(topic, MQTT::QOS1, message_arrived);
hyan99 0:3b4906b8a747 134 if (rc != 0) {
hyan99 0:3b4906b8a747 135 pc_comm.printf("There was a problem subscribing: %d\r\n", rc);
hyan99 0:3b4906b8a747 136
hyan99 0:3b4906b8a747 137 // client->disconnect();
hyan99 0:3b4906b8a747 138 return -1;
hyan99 0:3b4906b8a747 139 } else {
hyan99 0:3b4906b8a747 140 pc_comm.printf("Subscribed!\r\n");
hyan99 0:3b4906b8a747 141 }
hyan99 0:3b4906b8a747 142 return rc;
hyan99 0:3b4906b8a747 143 }
hyan99 0:3b4906b8a747 144
hyan99 0:3b4906b8a747 145 int Communication::subscribe_to_topic_control(char* topic) {
hyan99 1:19c3299ea83a 146 pc_comm.printf("subcribing: %s\r\n", topic);
hyan99 0:3b4906b8a747 147 int rc = client->subscribe(topic, MQTT::QOS1, control_arrived);
hyan99 0:3b4906b8a747 148 if (rc != 0) {
hyan99 0:3b4906b8a747 149 pc_comm.printf("There was a problem subscribing: %d\r\n", rc);
hyan99 0:3b4906b8a747 150
hyan99 0:3b4906b8a747 151 // client->disconnect();
hyan99 0:3b4906b8a747 152 return -1;
hyan99 0:3b4906b8a747 153 } else {
hyan99 0:3b4906b8a747 154 pc_comm.printf("Subscribed!\r\n");
hyan99 0:3b4906b8a747 155 }
hyan99 0:3b4906b8a747 156 return rc;
hyan99 0:3b4906b8a747 157 }
hyan99 0:3b4906b8a747 158
hyan99 0:3b4906b8a747 159 int Communication::subscribe_to_topic_sync(char* topic) {
hyan99 1:19c3299ea83a 160 pc_comm.printf("subcribing: %s\r\n", topic);
hyan99 1:19c3299ea83a 161 int rc = client->subscribe(topic, MQTT::QOS1, sync_arrived);
hyan99 0:3b4906b8a747 162 if (rc != 0) {
hyan99 0:3b4906b8a747 163 pc_comm.printf("There was a problem subscribing: %d\r\n", rc);
hyan99 0:3b4906b8a747 164
hyan99 0:3b4906b8a747 165 // client->disconnect();
hyan99 0:3b4906b8a747 166 return -1;
hyan99 0:3b4906b8a747 167 } else {
hyan99 0:3b4906b8a747 168 pc_comm.printf("Subscribed!\r\n");
hyan99 0:3b4906b8a747 169 }
hyan99 0:3b4906b8a747 170 return rc;
hyan99 0:3b4906b8a747 171 }
hyan99 0:3b4906b8a747 172
hyan99 0:3b4906b8a747 173 void Communication::publish_car(char* topic, int speed, int position) {
hyan99 0:3b4906b8a747 174 message_t *message = mpool.alloc();
hyan99 0:3b4906b8a747 175 message->topic = topic;
hyan99 0:3b4906b8a747 176 message->speed = speed;
hyan99 0:3b4906b8a747 177 message->position = position;
hyan99 0:3b4906b8a747 178 q_car.put(message);
hyan99 0:3b4906b8a747 179 }
hyan99 0:3b4906b8a747 180
hyan99 0:3b4906b8a747 181 void Communication::publish_road(char* topic, int num) {
hyan99 0:3b4906b8a747 182 message_t *message = mpool.alloc();
hyan99 0:3b4906b8a747 183 message->topic = topic;
hyan99 0:3b4906b8a747 184 message->speed = -1; // indicate this is for sync
hyan99 0:3b4906b8a747 185 message->position = num;
hyan99 0:3b4906b8a747 186 q_car.put(message);
hyan99 0:3b4906b8a747 187 }
hyan99 0:3b4906b8a747 188
hyan99 0:3b4906b8a747 189 void Communication::start() {
hyan99 0:3b4906b8a747 190 while(1) {
hyan99 0:3b4906b8a747 191 osEvent evt = q_car.get();
hyan99 0:3b4906b8a747 192 if (evt.status == osEventMessage) {
hyan99 0:3b4906b8a747 193 message_t *message = (message_t*)evt.value.p;
hyan99 0:3b4906b8a747 194 char* topic = message->topic;
hyan99 0:3b4906b8a747 195
hyan99 0:3b4906b8a747 196 if (message->speed == -1) {
hyan99 0:3b4906b8a747 197 char buf[1];
hyan99 0:3b4906b8a747 198 buf[0] = (char) message->position; // update number
hyan99 0:3b4906b8a747 199
hyan99 0:3b4906b8a747 200 int rc = client->publish(topic, (char*) buf, strlen(buf)+1, MQTT::QOS1);
hyan99 0:3b4906b8a747 201 if (rc != 0) {
hyan99 0:3b4906b8a747 202 pc_comm.printf("Failed to publish Road info: %s", topic);
hyan99 0:3b4906b8a747 203 } else {
hyan99 0:3b4906b8a747 204 pc_comm.printf("Message sent!\r\n");
hyan99 0:3b4906b8a747 205 }
hyan99 0:3b4906b8a747 206 } else {
hyan99 0:3b4906b8a747 207 char buf[2];
hyan99 0:3b4906b8a747 208 buf[0] = (char) message->position;
hyan99 0:3b4906b8a747 209 buf[1] = (char) message->speed;
hyan99 0:3b4906b8a747 210
hyan99 0:3b4906b8a747 211 int rc = client->publish(topic, (char*) buf, strlen(buf)+1, MQTT::QOS1);
hyan99 0:3b4906b8a747 212 if (rc != 0) {
hyan99 0:3b4906b8a747 213 pc_comm.printf("Failed to publish AccCar info: %s", topic);
hyan99 0:3b4906b8a747 214 } else {
hyan99 0:3b4906b8a747 215 pc_comm.printf("Message sent!\r\n");
hyan99 0:3b4906b8a747 216 }
hyan99 0:3b4906b8a747 217 }
hyan99 0:3b4906b8a747 218 mpool.free(message);
hyan99 0:3b4906b8a747 219 client->yield(100); // wait for 100ms
hyan99 0:3b4906b8a747 220 }
hyan99 0:3b4906b8a747 221 }
hyan99 0:3b4906b8a747 222 }
hyan99 0:3b4906b8a747 223
hyan99 0:3b4906b8a747 224 void Communication::reset() {
hyan99 0:3b4906b8a747 225 control_flags.clear();
hyan99 0:3b4906b8a747 226
hyan99 0:3b4906b8a747 227 if (thread != NULL) {
hyan99 0:3b4906b8a747 228 thread->terminate();
hyan99 0:3b4906b8a747 229 }
hyan99 0:3b4906b8a747 230
hyan99 0:3b4906b8a747 231 thread = new Thread();
hyan99 0:3b4906b8a747 232 thread->start( callback(this, &Communication::start) );
hyan99 0:3b4906b8a747 233 }
hyan99 0:3b4906b8a747 234
hyan99 0:3b4906b8a747 235 void Communication::stop() {
hyan99 0:3b4906b8a747 236 if (thread != NULL) {
hyan99 0:3b4906b8a747 237 thread->terminate();
hyan99 0:3b4906b8a747 238 }
hyan99 0:3b4906b8a747 239 }