Jim Flynn
Added function deleteLocations to delete the location history.
Dependents: WNCInterface_M2Xdemo ATT_WNCInterface_Info WNCInterface_HTTP_example Public_IoT_M2X_Cellular_Demo
Fork of
M2XStreamClient
by 
AT&T M2X Team
Embed:
(wiki syntax)
Show/hide line numbers
M2XStreamClient.h
00001 #ifndef M2XStreamClient_h 00002 #define M2XStreamClient_h 00003 00004 #if (!defined(ARDUINO_PLATFORM)) && (!defined(ESP8266_PLATFORM)) && (!defined(MBED_PLATFORM)) 00005 #error "Platform definition is missing!" 00006 #endif 00007 00008 #define M2X_VERSION "2.2.0" 00009 00010 #ifdef ARDUINO_PLATFORM 00011 #include "m2x-arduino.h" 00012 #endif /* ARDUINO_PLATFORM */ 00013 00014 #ifdef ESP8266_PLATFORM 00015 #include "m2x-esp8266.h" 00016 #endif /* ESP8266_PLATFORM */ 00017 00018 #ifdef MBED_PLATFORM 00019 #include "m2x-mbed.h" 00020 #endif /* MBED_PLATFORM */ 00021 00022 /* If we don't have DBG defined, provide dump implementation */ 00023 #ifndef DBG 00024 #define DBG(fmt_, data_) 00025 #define DBGLN(fmt_, data_) 00026 #define DBGLNEND 00027 #endif /* DBG */ 00028 00029 #define MIN(a, b) (((a) > (b))?(b):(a)) 00030 #define TO_HEX(t_) ((char) (((t_) > 9) ? ((t_) - 10 + 'A') : ((t_) + '0'))) 00031 #define MAX_DOUBLE_DIGITS 7 00032 00033 /* For tolower */ 00034 #include <ctype.h> 00035 00036 static const int E_OK = 0; 00037 static const int E_NOCONNECTION = -1; 00038 static const int E_DISCONNECTED = -2; 00039 static const int E_NOTREACHABLE = -3; 00040 static const int E_INVALID = -4; 00041 static const int E_JSON_INVALID = -5; 00042 static const int E_BUFFER_TOO_SMALL = -6; 00043 static const int E_TIMESTAMP_ERROR = -8; 00044 00045 static const char* DEFAULT_M2X_HOST = "api-m2x.att.com"; 00046 static const int DEFAULT_M2X_PORT = 80; 00047 00048 static inline bool m2x_status_is_success(int status) { 00049 return (status == E_OK) || (status >= 200 && status <= 299); 00050 } 00051 00052 static inline bool m2x_status_is_client_error(int status) { 00053 return status >= 400 && status <= 499; 00054 } 00055 00056 static inline bool m2x_status_is_server_error(int status) { 00057 return status >= 500 && status <= 599; 00058 } 00059 00060 static inline bool m2x_status_is_error(int status) { 00061 return m2x_status_is_client_error(status) || 00062 m2x_status_is_server_error(status); 00063 } 00064 00065 // Null Print class used to calculate length to print 00066 class NullPrint : public Print { 00067 public: 00068 size_t counter; 00069 00070 virtual size_t write(uint8_t b) { 00071 counter++; 00072 return 1; 00073 } 00074 00075 virtual size_t write(const uint8_t* buf, size_t size) { 00076 counter += size; 00077 return size; 00078 } 00079 }; 00080 00081 // Encodes and prints string using Percent-encoding specified 00082 // in RFC 1738, Section 2.2 00083 static inline int print_encoded_string(Print* print, const char* str) { 00084 int bytes = 0; 00085 for (int i = 0; str[i] != 0; i++) { 00086 if (((str[i] >= 'A') && (str[i] <= 'Z')) || 00087 ((str[i] >= 'a') && (str[i] <= 'z')) || 00088 ((str[i] >= '0') && (str[i] <= '9')) || 00089 (str[i] == '-') || (str[i] == '_') || 00090 (str[i] == '.') || (str[i] == '~')) { 00091 bytes += print->print(str[i]); 00092 } else { 00093 // Encode all other characters 00094 bytes += print->print('%'); 00095 bytes += print->print(TO_HEX(str[i] / 16)); 00096 bytes += print->print(TO_HEX(str[i] % 16)); 00097 } 00098 } 00099 return bytes; 00100 } 00101 00102 #ifdef M2X_ENABLE_READER 00103 /* 00104 * +type+ indicates the value type: 1 for string, 2 for number 00105 * NOTE that the value type here only contains a hint on how 00106 * you can use the value. Even though 2 is returned, the value 00107 * is still stored in (const char *), and atoi/atof is needed to 00108 * get the actual value 00109 */ 00110 typedef void (*stream_value_read_callback)(const char* at, 00111 const char* value, 00112 int index, 00113 void* context, 00114 int type); 00115 00116 typedef void (*location_read_callback)(const char* name, 00117 double latitude, 00118 double longitude, 00119 double elevation, 00120 const char* timestamp, 00121 int index, 00122 void* context); 00123 00124 typedef void (*m2x_command_read_callback)(const char* id, 00125 const char* name, 00126 int index, 00127 void *context); 00128 #endif /* M2X_ENABLE_READER */ 00129 00130 typedef void (*m2x_fill_data_callback)(Print *print, void *context); 00131 00132 class M2XStreamClient { 00133 public: 00134 M2XStreamClient(Client* client, 00135 const char* key, 00136 void (* idlefunc)(void) = NULL, 00137 int case_insensitive = 1, 00138 const char* host = DEFAULT_M2X_HOST, 00139 int port = DEFAULT_M2X_PORT, 00140 const char* path_prefix = NULL); 00141 00142 // Push data stream value using PUT request, returns the HTTP status code 00143 // NOTE: if you want to update by a serial, use "serial/<serial ID>" as 00144 // the device ID here. 00145 template <class T> 00146 int updateStreamValue(const char* deviceId, const char* streamName, T value); 00147 00148 // Post multiple values to M2X all at once. 00149 // +deviceId+ - id of the device to post values 00150 // +streamNum+ - Number of streams to post 00151 // +names+ - Array of stream names, the length of the array should 00152 // be exactly +streamNum+ 00153 // +counts+ - Array of +streamNum+ length, each item in this array 00154 // containing the number of values we want to post for each stream 00155 // +ats+ - Timestamps for each value, the length of this array should 00156 // be the some of all values in +counts+, for the first +counts[0]+ 00157 // items, the values belong to the first stream, for the following 00158 // +counts[1]+ number of items, the values belong to the second stream, 00159 // etc. Notice that timestamps are required here: you must provide 00160 // a timestamp for each value posted. 00161 // +values+ - Values to post. This works the same way as +ats+, the 00162 // first +counts[0]+ number of items contain values to post to the first 00163 // stream, the succeeding +counts[1]+ number of items contain values 00164 // for the second stream, etc. The length of this array should be 00165 // the sum of all values in +counts+ array. 00166 // NOTE: if you want to update by a serial, use "serial/<serial ID>" as 00167 // the device ID here. 00168 template <class T> 00169 int postDeviceUpdates(const char* deviceId, int streamNum, 00170 const char* names[], const int counts[], 00171 const char* ats[], T values[]); 00172 00173 // Post multiple values of a single device at once. 00174 // +deviceId+ - id of the device to post values 00175 // +streamNum+ - Number of streams to post 00176 // +names+ - Array of stream names, the length of the array should 00177 // be exactly +streamNum+ 00178 // +values+ - Array of values to post, the length of the array should 00179 // be exactly +streamNum+. Notice that the array of +values+ should 00180 // match the array of +names+, and that the ith value in +values+ is 00181 // exactly the value to post for the ith stream name in +names+ 00182 // NOTE: if you want to update by a serial, use "serial/<serial ID>" as 00183 // the device ID here. 00184 template <class T> 00185 int postDeviceUpdate(const char* deviceId, int streamNum, 00186 const char* names[], T values[], 00187 const char* at = NULL); 00188 00189 #ifdef M2X_ENABLE_READER 00190 // Fetch values for a particular data stream. Since memory is 00191 // very limited on an Arduino, we cannot parse and get all the 00192 // data points in memory. Instead, we use callbacks here: whenever 00193 // a new data point is parsed, we call the callback using the values, 00194 // after that, the values will be thrown away to make space for new 00195 // values. 00196 // Note that you can also pass in a user-specified context in this 00197 // function, this context will be passed to the callback function 00198 // each time we get a data point. 00199 // For each data point, the callback will be called once. The HTTP 00200 // status code will be returned. And the content is only parsed when 00201 // the status code is 200. 00202 int listStreamValues(const char* deviceId, const char* streamName, 00203 stream_value_read_callback callback, void* context, 00204 const char* query = NULL); 00205 #endif /* M2X_ENABLE_READER */ 00206 00207 // Update datasource location 00208 // NOTE: On an Arduino Uno and other ATMEGA based boards, double has 00209 // 4-byte (32 bits) precision, which is the same as float. So there's 00210 // no natural double-precision floating number on these boards. With 00211 // a float value, we have a precision of roughly 7 digits, that means 00212 // either 5 or 6 digits after the floating point. According to wikipedia, 00213 // a difference of 0.00001 will give us ~1.1132m distance. If this 00214 // precision is good for you, you can use the double-version we provided 00215 // here. Otherwise, you may need to use the string-version and do the 00216 // actual conversion by yourselves. 00217 // However, with an Arduino Due board, double has 8-bytes (64 bits) 00218 // precision, which means you are free to use the double-version only 00219 // without any precision problems. 00220 // Returned value is the http status code. 00221 // NOTE: if you want to update by a serial, use "serial/<serial ID>" as 00222 // the device ID here. 00223 template <class T> 00224 int updateLocation(const char* deviceId, const char* name, 00225 T latitude, T longitude, T elevation); 00226 00227 #ifdef M2X_ENABLE_READER 00228 // Read location information for a device. Also used callback to process 00229 // data points for memory reasons. The HTTP status code is returned, 00230 // response is only parsed when the HTTP status code is 200 00231 int readLocation(const char* deviceId, location_read_callback callback, 00232 void* context); 00233 00234 // Delete location information for a device. The HTTP status code is 00235 // returned,response is only parsed when the HTTP status code is 200 00236 int deleteLocations(const char* deviceId, 00237 const char* from, const char* end); 00238 00239 #endif /* M2X_ENABLE_READER */ 00240 00241 // Delete values from a data stream 00242 // You will need to provide from and end date/time strings in the ISO8601 00243 // format "yyyy-mm-ddTHH:MM:SS.SSSZ" where 00244 // yyyy: the year 00245 // mm: the month 00246 // dd: the day 00247 // HH: the hour (24 hour format) 00248 // MM: the minute 00249 // SS.SSS: the seconds (to the millisecond) 00250 // NOTE: the time is given in Zulu (GMT) 00251 // M2X will delete all values within the from to end date/time range. 00252 // The status code is 204 on success and 400 on a bad request (e.g. the 00253 // timestamp is not in ISO8601 format or the from timestamp is not less than 00254 // or equal to the end timestamp. 00255 int deleteValues(const char* deviceId, const char* streamName, 00256 const char* from, const char* end); 00257 00258 #ifdef M2X_ENABLE_READER 00259 // Fetch commands available for this device, notice that for memory constraints, 00260 // we only keep ID and name of the command received here. 00261 // You can tweak the command receiving via the query parameter. 00262 int listCommands(const char* deviceId, 00263 m2x_command_read_callback callback, void* context, 00264 const char* query = NULL); 00265 #endif /* M2X_ENABLE_READER */ 00266 00267 // Mark a command as processed. 00268 // Link: https://m2x.att.com/developer/documentation/v2/commands#Device-Marks-a-Command-as-Processed 00269 // To make sure the minimal amount of memory is needed, this API works with 00270 // a callback function. The callback function is then used to fill the request 00271 // data, note that in order to correctly set the content length in HTTP header, 00272 // the callback function will be called twice, the caller must make sure both 00273 // calls fill the Print object with exactly the same data. 00274 // If you have a pre-allocated buffer filled with the data to post, you can 00275 // use markCommandProcessedWithData API below. 00276 int markCommandProcessed(const char* deviceId, const char* commandId, 00277 m2x_fill_data_callback callback, void *context); 00278 00279 // Mark a command as processed with a data buffer 00280 // Link: https://m2x.att.com/developer/documentation/v2/commands#Device-Marks-a-Command-as-Processed 00281 // This is exactly like markCommandProcessed, except that a buffer is use to 00282 // contain the data to post as the request body. 00283 int markCommandProcessedWithData(const char* deviceId, const char* commandId, 00284 const char* data); 00285 00286 // Mark a command as rejected. 00287 // Link: https://m2x.att.com/developer/documentation/v2/commands#Device-Marks-a-Command-as-Rejected 00288 // To make sure the minimal amount of memory is needed, this API works with 00289 // a callback function. The callback function is then used to fill the request 00290 // data, note that in order to correctly set the content length in HTTP header, 00291 // the callback function will be called twice, the caller must make sure both 00292 // calls fill the Print object with exactly the same data. 00293 // If you have a pre-allocated buffer filled with the data to post, you can 00294 // use markCommandRejectedWithData API below. 00295 int markCommandRejected(const char* deviceId, const char* commandId, 00296 m2x_fill_data_callback callback, void *context); 00297 00298 // Mark a command as rejected with a data buffer 00299 // Link: https://m2x.att.com/developer/documentation/v2/commands#Device-Marks-a-Command-as-Rejected 00300 // This is exactly like markCommandRejected, except that a buffer is use to 00301 // contain the data to post as the request body. 00302 int markCommandRejectedWithData(const char* deviceId, const char* commandId, 00303 const char* data); 00304 00305 // Fetches current timestamp in seconds from M2X server. Since we 00306 // are using signed 32-bit integer as return value, this will only 00307 // return valid results before 03:14:07 UTC on 19 January 2038. If 00308 // the device is supposed to work after that, this function should 00309 // not be used. 00310 // 00311 // The returned value will contain the status code(positive values) 00312 // or the error code(negative values). 00313 // In case of success, the current timestamp will be filled in the 00314 // +ts+ pointer passed in as argument. 00315 // 00316 // NOTE: although returning uint32_t can give us a larger space, 00317 // we prefer to cope with the unix convention here. 00318 int getTimestamp32(int32_t* ts); 00319 00320 // Fetches current timestamp in seconds from M2X server. 00321 // This function will return the timestamp as an integer literal 00322 // in the provided buffer. Hence there's no problem working after 00323 // 03:14:07 UTC on 19 January 2038. The drawback part here, is that 00324 // you will have to work with 64-bit integer, which is not available 00325 // on certain platform(such as Arduino), a bignum library or alike 00326 // is needed in this case. 00327 // 00328 // Notice +bufferLength+ is supposed to contain the length of the 00329 // buffer when calling this function. It is also the caller's 00330 // responsibility to ensure the buffer is big enough, otherwise 00331 // the library will return an error indicating the buffer is too 00332 // small. 00333 // While this is not accurate all the time, one trick here is to 00334 // pass in 0 as the bufferLength, in which case we will always return 00335 // the buffer-too-small error. However, the correct buffer length 00336 // can be found this way so a secound execution is most likely to work 00337 // (unless we are at the edge of the buffer length increasing, for 00338 // example, when the timestamp jumps from 9999999999 to 10000000000, 00339 // which is highly unlikely to happend). However, given that the 00340 // maximum 64-bit integer can be stored in 19 bytes, there's not 00341 // much need to use this trick.) 00342 // 00343 // The returned value will contain the status code(positive values) 00344 // or the error code(negative values). 00345 // In case of success, the current timestamp will be filled in the 00346 // passed +buffer+ pointer, and the actual used buffer length will 00347 // be returned in +bufferLength+ pointer. 00348 // NOTE: as long as we can read the returned buffer length, it will 00349 // be used to fill in the +bufferLength+ variable even though other 00350 // errors occur(buffer is not enough, network is shutdown before 00351 // reading the whole buffer, etc.) 00352 int getTimestamp(char* buffer, int* bufferLength); 00353 private: 00354 Client* _client; 00355 const char* _key; 00356 int _case_insensitive; 00357 const char* _host; 00358 int _port; 00359 void (* _idlefunc)(void); 00360 const char* _path_prefix; 00361 NullPrint _null_print; 00362 00363 // Writes the HTTP header part for updating a stream value 00364 void writePutHeader(const char* deviceId, 00365 const char* streamName, 00366 int contentLength); 00367 // Writes the HTTP header part for deleting stream values 00368 void writeDeleteHeader(const char* deviceId, 00369 const char* streamName, 00370 int contentLength); 00371 void writeDeleteLocationHeader(const char* deviceId, 00372 int contentLength); 00373 00374 // Writes HTTP header lines including M2X API Key, host, content 00375 // type and content length(if the body exists) 00376 void writeHttpHeader(int contentLength); 00377 // Parses HTTP response header and return the content length. 00378 // Note that this function does not parse all http headers, as long 00379 // as the content length is found, this function will return 00380 int readContentLength(); 00381 // Skips all HTTP response header part. Return minus value in case 00382 // the connection is closed before we got all headers 00383 int skipHttpHeader(); 00384 // Parses and returns the HTTP status code, note this function will 00385 // return immediately once it gets the status code 00386 int readStatusCode(bool closeClient); 00387 // Waits for a certain string pattern in the HTTP header, and returns 00388 // once the pattern is found. In the pattern, you can use '*' to denote 00389 // any character 00390 int waitForString(const char* str); 00391 // Closes the connection 00392 void close(); 00393 00394 #ifdef M2X_ENABLE_READER 00395 // Parses JSON response of stream value API, and calls callback function 00396 // once we get a data point 00397 int readStreamValue(stream_value_read_callback callback, void* context); 00398 // Parses JSON response of location API, and calls callback function once 00399 // we get a data point 00400 int readLocation(location_read_callback callback, void* context); 00401 // Parses JSON response of command API, and calls callback function once 00402 // we get a data point 00403 int readCommand(m2x_command_read_callback callback, void* context); 00404 #endif /* M2X_ENABLE_READER */ 00405 }; 00406 00407 00408 // A ISO8601 timestamp generation service for M2X. 00409 // It uses the Time API provided by the M2X server to initialize 00410 // clock, then uses millis() function provided by Arduino to calculate 00411 // time advancements so as to reduce API query times. 00412 // 00413 // Right now, this service only works with 32-bit timestamp, meaning that 00414 // this service won't work after 03:14:07 UTC on 19 January 2038. However, 00415 // a similar service that uses 64-bit timestamp can be implemented following 00416 // the logic here. 00417 class TimeService { 00418 public: 00419 TimeService(M2XStreamClient* client); 00420 00421 // Initialize the time service. Notice the TimeService instance is only 00422 // working after calling this function successfully. 00423 int init(); 00424 00425 // Reset the internal recorded time by calling M2X Time API again. Normally, 00426 // you don't need to call this manually. TimeService will handle Arduino clock 00427 // overflow automatically 00428 int reset(); 00429 00430 // Fills ISO8601 formatted timestamp into the buffer provided. +length+ should 00431 // contains the maximum supported length of the buffer when calling. For now, 00432 // the buffer should be able to store 25 characters for a full ISO8601 formatted 00433 // timestamp, otherwise, an error will be returned. 00434 int getTimestamp(char* buffer, int* length); 00435 private: 00436 M2XStreamClient* _client; 00437 int32_t _server_timestamp; 00438 uint32_t _local_last_milli; 00439 M2XTimer _timer; 00440 }; 00441 00442 00443 // Implementations 00444 M2XStreamClient::M2XStreamClient(Client* client, 00445 const char* key, 00446 void (* idlefunc)(void), 00447 int case_insensitive, 00448 const char* host, 00449 int port, 00450 const char* path_prefix) : _client(client), 00451 _key(key), 00452 _idlefunc(idlefunc), 00453 _case_insensitive(case_insensitive), 00454 _host(host), 00455 _port(port), 00456 _path_prefix(path_prefix), 00457 _null_print() { 00458 } 00459 00460 template <class T> 00461 int M2XStreamClient::updateStreamValue(const char* deviceId, const char* streamName, T value) { 00462 if (_client->connect(_host, _port)) { 00463 DBGLN("%s", "Connected to M2X server!"); 00464 writePutHeader(deviceId, streamName, 00465 // for {"value": and } 00466 _null_print.print(value) + 12); 00467 _client->print("{\"value\":\""); 00468 _client->print(value); 00469 _client->print("\"}"); 00470 } else { 00471 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00472 return E_NOCONNECTION; 00473 } 00474 00475 return readStatusCode(true); 00476 } 00477 00478 template <class T> 00479 inline int write_multiple_values(Print* print, int streamNum, 00480 const char* names[], const int counts[], 00481 const char* ats[], T values[]) { 00482 int bytes = 0, value_index = 0; 00483 bytes += print->print("{\"values\":{"); 00484 for (int i = 0; i < streamNum; i++) { 00485 bytes += print->print("\""); 00486 bytes += print->print(names[i]); 00487 bytes += print->print("\":["); 00488 for (int j = 0; j < counts[i]; j++) { 00489 bytes += print->print("{\"timestamp\": \""); 00490 bytes += print->print(ats[value_index]); 00491 bytes += print->print("\",\"value\": \""); 00492 bytes += print->print(values[value_index]); 00493 bytes += print->print("\"}"); 00494 if (j < counts[i] - 1) { bytes += print->print(","); } 00495 value_index++; 00496 } 00497 bytes += print->print("]"); 00498 if (i < streamNum - 1) { bytes += print->print(","); } 00499 } 00500 bytes += print->print("}}"); 00501 return bytes; 00502 } 00503 00504 template <class T> 00505 int M2XStreamClient::postDeviceUpdates(const char* deviceId, int streamNum, 00506 const char* names[], const int counts[], 00507 const char* ats[], T values[]) { 00508 if (_client->connect(_host, _port)) { 00509 DBGLN("%s", "Connected to M2X server!"); 00510 int length = write_multiple_values(&_null_print, streamNum, names, 00511 counts, ats, values); 00512 _client->print("POST "); 00513 if (_path_prefix) { _client->print(_path_prefix); } 00514 _client->print("/v2/devices/"); 00515 _client->print(deviceId); 00516 _client->println("/updates HTTP/1.1"); 00517 writeHttpHeader(length); 00518 write_multiple_values(_client, streamNum, names, counts, ats, values); 00519 } else { 00520 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00521 return E_NOCONNECTION; 00522 } 00523 return readStatusCode(true); 00524 } 00525 00526 template <class T> 00527 inline int write_single_device_values(Print* print, int streamNum, 00528 const char* names[], T values[], 00529 const char* at) { 00530 int bytes = 0; 00531 bytes += print->print("{\"values\":{"); 00532 for (int i = 0; i < streamNum; i++) { 00533 bytes += print->print("\""); 00534 bytes += print->print(names[i]); 00535 bytes += print->print("\": \""); 00536 bytes += print->print(values[i]); 00537 bytes += print->print("\""); 00538 if (i < streamNum - 1) { bytes += print->print(","); } 00539 } 00540 bytes += print->print("}"); 00541 if (at != NULL) { 00542 bytes += print->print(",\"timestamp\":\""); 00543 bytes += print->print(at); 00544 bytes += print->print("\""); 00545 } 00546 bytes += print->print("}"); 00547 return bytes; 00548 } 00549 00550 template <class T> 00551 int M2XStreamClient::postDeviceUpdate(const char* deviceId, int streamNum, 00552 const char* names[], T values[], 00553 const char* at) { 00554 if (_client->connect(_host, _port)) { 00555 DBGLN("%s", "Connected to M2X server!"); 00556 int length = write_single_device_values(&_null_print, streamNum, names, 00557 values, at); 00558 _client->print("POST "); 00559 if (_path_prefix) { _client->print(_path_prefix); } 00560 _client->print("/v2/devices/"); 00561 _client->print(deviceId); 00562 _client->println("/update HTTP/1.1"); 00563 writeHttpHeader(length); 00564 write_single_device_values(_client, streamNum, names, values, at); 00565 } else { 00566 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00567 return E_NOCONNECTION; 00568 } 00569 return readStatusCode(true); 00570 } 00571 00572 template <class T> 00573 static int write_location_data(Print* print, const char* name, 00574 T latitude, T longitude, 00575 T elevation) { 00576 int bytes = 0; 00577 bytes += print->print("{\"name\":\""); 00578 bytes += print->print(name); 00579 bytes += print->print("\",\"latitude\":\""); 00580 bytes += print->print(latitude); 00581 bytes += print->print("\",\"longitude\":\""); 00582 bytes += print->print(longitude); 00583 bytes += print->print("\",\"elevation\":\""); 00584 bytes += print->print(elevation); 00585 bytes += print->print("\"}"); 00586 return bytes; 00587 } 00588 00589 static int write_location_data(Print* print, const char* name, 00590 double latitude, double longitude, 00591 double elevation) { 00592 int bytes = 0; 00593 bytes += print->print("{\"name\":\""); 00594 bytes += print->print(name); 00595 bytes += print->print("\",\"latitude\":\""); 00596 bytes += print->print(latitude, MAX_DOUBLE_DIGITS); 00597 bytes += print->print("\",\"longitude\":\""); 00598 bytes += print->print(longitude, MAX_DOUBLE_DIGITS); 00599 bytes += print->print("\",\"elevation\":\""); 00600 bytes += print->print(elevation); 00601 bytes += print->print("\"}"); 00602 return bytes; 00603 } 00604 00605 template <class T> 00606 int M2XStreamClient::updateLocation(const char* deviceId, 00607 const char* name, 00608 T latitude, 00609 T longitude, 00610 T elevation) { 00611 if (_client->connect(_host, _port)) { 00612 DBGLN("%s", "Connected to M2X server!"); 00613 00614 int length = write_location_data(&_null_print, name, latitude, longitude, 00615 elevation); 00616 _client->print("PUT "); 00617 if (_path_prefix) { _client->print(_path_prefix); } 00618 _client->print("/v2/devices/"); 00619 _client->print(deviceId); 00620 _client->println("/location HTTP/1.1"); 00621 00622 writeHttpHeader(length); 00623 write_location_data(_client, name, latitude, longitude, elevation); 00624 } else { 00625 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00626 return E_NOCONNECTION; 00627 } 00628 return readStatusCode(true); 00629 } 00630 00631 static inline int write_delete_values(Print* print, const char* from, 00632 const char* end) { 00633 int bytes = 0; 00634 bytes += print->print("{\"from\":\""); 00635 bytes += print->print(from); 00636 bytes += print->print("\",\"end\":\""); 00637 bytes += print->print(end); 00638 bytes += print->print("\"}"); 00639 return bytes; 00640 } 00641 00642 int M2XStreamClient::deleteValues(const char* deviceId, const char* streamName, 00643 const char* from, const char* end) { 00644 if (_client->connect(_host, _port)) { 00645 DBGLN("%s", "Connected to M2X server!"); 00646 int length = write_delete_values(&_null_print, from, end); 00647 writeDeleteHeader(deviceId, streamName, length); 00648 write_delete_values(_client, from, end); 00649 } else { 00650 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00651 return E_NOCONNECTION; 00652 } 00653 00654 return readStatusCode(true); 00655 } 00656 00657 int M2XStreamClient::markCommandProcessed(const char* deviceId, 00658 const char* commandId, 00659 m2x_fill_data_callback callback, 00660 void *context) { 00661 if (_client->connect(_host, _port)) { 00662 DBGLN("%s", "Connected to M2X server!"); 00663 _null_print.counter = 0; 00664 callback(&_null_print, context); 00665 int length = _null_print.counter; 00666 _client->print("POST "); 00667 if (_path_prefix) { _client->print(_path_prefix); } 00668 _client->print("/v2/devices/"); 00669 _client->print(deviceId); 00670 _client->print("/commands/"); 00671 _client->print(commandId); 00672 _client->print("/process"); 00673 _client->println(" HTTP/1.1"); 00674 writeHttpHeader(length); 00675 callback(_client, context); 00676 } else { 00677 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00678 return E_NOCONNECTION; 00679 } 00680 return readStatusCode(true); 00681 } 00682 00683 static void m2x_fixed_buffer_filling_callback(Print *print, void *context) { 00684 if (context) { 00685 print->print((const char *) context); 00686 } 00687 } 00688 00689 int M2XStreamClient::markCommandProcessedWithData(const char* deviceId, 00690 const char* commandId, 00691 const char* data) { 00692 return markCommandProcessed(deviceId, commandId, 00693 m2x_fixed_buffer_filling_callback, (void *) data); 00694 } 00695 00696 int M2XStreamClient::markCommandRejected(const char* deviceId, 00697 const char* commandId, 00698 m2x_fill_data_callback callback, 00699 void *context) { 00700 if (_client->connect(_host, _port)) { 00701 DBGLN("%s", "Connected to M2X server!"); 00702 _null_print.counter = 0; 00703 callback(&_null_print, context); 00704 int length = _null_print.counter; 00705 _client->print("POST "); 00706 if (_path_prefix) { _client->print(_path_prefix); } 00707 _client->print("/v2/devices/"); 00708 _client->print(deviceId); 00709 _client->print("/commands/"); 00710 _client->print(commandId); 00711 _client->print("/reject"); 00712 _client->println(" HTTP/1.1"); 00713 writeHttpHeader(length); 00714 callback(_client, context); 00715 } else { 00716 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00717 return E_NOCONNECTION; 00718 } 00719 return readStatusCode(true); 00720 } 00721 00722 int M2XStreamClient::markCommandRejectedWithData(const char* deviceId, 00723 const char* commandId, 00724 const char* data) { 00725 return markCommandRejected(deviceId, commandId, 00726 m2x_fixed_buffer_filling_callback, (void *) data); 00727 } 00728 00729 int M2XStreamClient::getTimestamp32(int32_t *ts) { 00730 // The maximum value of signed 64-bit integer is 0x7fffffffffffffff, 00731 // which is 9223372036854775807. It consists of 19 characters, so a 00732 // buffer of 20 is definitely enough here 00733 int length = 20; 00734 char buffer[20]; 00735 int status = getTimestamp(buffer, &length); 00736 if (status == 200) { 00737 int32_t result = 0; 00738 for (int i = 0; i < length; i++) { 00739 result = result * 10 + (buffer[i] - '0'); 00740 } 00741 if (ts != NULL) { *ts = result; } 00742 } 00743 return status; 00744 } 00745 00746 int M2XStreamClient::getTimestamp(char* buffer, int *bufferLength) { 00747 if (bufferLength == NULL) { return E_INVALID; } 00748 if (_client->connect(_host, _port)) { 00749 DBGLN("%s", "Connected to M2X server!"); 00750 _client->print("GET "); 00751 if (_path_prefix) { _client->print(_path_prefix); } 00752 _client->println("/v2/time/seconds HTTP/1.1"); 00753 00754 writeHttpHeader(-1); 00755 } else { 00756 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 00757 return E_NOCONNECTION; 00758 } 00759 int status = readStatusCode(false); 00760 if (status == 200) { 00761 int length = readContentLength(); 00762 if (length < 0) { 00763 close(); 00764 return length; 00765 } 00766 if (*bufferLength < length) { 00767 *bufferLength = length; 00768 return E_BUFFER_TOO_SMALL; 00769 } 00770 *bufferLength = length; 00771 int index = skipHttpHeader(); 00772 if (index != E_OK) { 00773 close(); 00774 return index; 00775 } 00776 index = 0; 00777 while (index < length) { 00778 DBG("%s", "Received Data: "); 00779 while ((index < length) && _client->available()) { 00780 buffer[index++] = _client->read(); 00781 DBG("%c", buffer[index - 1]); 00782 } 00783 DBGLNEND; 00784 00785 if ((!_client->connected()) && 00786 (index < length)) { 00787 close(); 00788 return E_NOCONNECTION; 00789 } 00790 00791 if (_idlefunc!=NULL) { 00792 _idlefunc(); 00793 } else { 00794 delay(200); 00795 } 00796 } 00797 } 00798 close(); 00799 return status; 00800 } 00801 00802 00803 void M2XStreamClient::writePutHeader(const char* deviceId, 00804 const char* streamName, 00805 int contentLength) { 00806 _client->print("PUT "); 00807 if (_path_prefix) { _client->print(_path_prefix); } 00808 _client->print("/v2/devices/"); 00809 _client->print(deviceId); 00810 _client->print("/streams/"); 00811 print_encoded_string(_client, streamName); 00812 _client->println("/value HTTP/1.1"); 00813 00814 writeHttpHeader(contentLength); 00815 } 00816 00817 void M2XStreamClient::writeDeleteHeader(const char* deviceId, 00818 const char* streamName, 00819 int contentLength) { 00820 _client->print("DELETE "); 00821 if (_path_prefix) { _client->print(_path_prefix); } 00822 _client->print("/v2/devices/"); 00823 _client->print(deviceId); 00824 _client->print("/streams/"); 00825 print_encoded_string(_client, streamName); 00826 _client->print("/values"); 00827 _client->println(" HTTP/1.1"); 00828 00829 writeHttpHeader(contentLength); 00830 } 00831 00832 void M2XStreamClient::writeDeleteLocationHeader(const char* deviceId, 00833 int contentLength) { 00834 _client->print("DELETE "); 00835 if (_path_prefix) { _client->print(_path_prefix); } 00836 _client->print("/v2/devices/"); 00837 _client->print(deviceId); 00838 _client->print("/location/waypoints/"); 00839 _client->println(" HTTP/1.1"); 00840 00841 writeHttpHeader(contentLength); 00842 } 00843 00844 00845 void M2XStreamClient::writeHttpHeader(int contentLength) { 00846 _client->println(USER_AGENT); 00847 _client->print("X-M2X-KEY: "); 00848 _client->println(_key); 00849 00850 _client->print("Host: "); 00851 print_encoded_string(_client, _host); 00852 if (_port != DEFAULT_M2X_PORT) { 00853 _client->print(":"); 00854 // port is an integer, does not need encoding 00855 _client->print(_port); 00856 } 00857 _client->println(); 00858 00859 if (contentLength > 0) { 00860 _client->println("Content-Type: application/json"); 00861 DBG("%s", "Content Length: "); 00862 DBGLN("%d", contentLength); 00863 00864 _client->print("Content-Length: "); 00865 _client->println(contentLength); 00866 } 00867 _client->println(); 00868 } 00869 00870 int M2XStreamClient::waitForString(const char* str) { 00871 int currentIndex = 0; 00872 if (str[currentIndex] == '\0') return E_OK; 00873 00874 while (true) { 00875 while (_client->available()) { 00876 char c = _client->read(); 00877 DBG("%c", c); 00878 00879 int cmp; 00880 if (_case_insensitive) { 00881 cmp = tolower(c) - tolower(str[currentIndex]); 00882 } else { 00883 cmp = c - str[currentIndex]; 00884 } 00885 00886 if ((str[currentIndex] == '*') || (cmp == 0)) { 00887 currentIndex++; 00888 if (str[currentIndex] == '\0') { 00889 return E_OK; 00890 } 00891 } else { 00892 // start from the beginning 00893 currentIndex = 0; 00894 } 00895 } 00896 00897 if (!_client->connected()) { 00898 DBGLN("%s", "ERROR: The client is disconnected from the server!"); 00899 00900 close(); 00901 return E_DISCONNECTED; 00902 } 00903 if (_idlefunc!=NULL) 00904 _idlefunc(); 00905 else 00906 delay(200); 00907 } 00908 // never reached here 00909 return E_NOTREACHABLE; 00910 } 00911 00912 int M2XStreamClient::readStatusCode(bool closeClient) { 00913 int responseCode = 0; 00914 int ret = waitForString("HTTP/*.* "); 00915 if (ret != E_OK) { 00916 if (closeClient) close(); 00917 return ret; 00918 } 00919 00920 // ret is not needed from here(since it must be E_OK), so we can use it 00921 // as a regular variable now. 00922 ret = 0; 00923 while (true) { 00924 while (_client->available()) { 00925 char c = _client->read(); 00926 DBG("%c", c); 00927 00928 responseCode = responseCode * 10 + (c - '0'); 00929 ret++; 00930 if (ret == 3) { 00931 if (closeClient) close(); 00932 return responseCode; 00933 } 00934 } 00935 00936 if (!_client->connected()) { 00937 DBGLN("%s", "ERROR: The client is disconnected from the server!"); 00938 00939 if (closeClient) close(); 00940 return E_DISCONNECTED; 00941 } 00942 if (_idlefunc!=NULL) 00943 _idlefunc(); 00944 else 00945 delay(200); 00946 } 00947 00948 // never reached here 00949 return E_NOTREACHABLE; 00950 } 00951 00952 int M2XStreamClient::readContentLength() { 00953 int ret = waitForString("Content-Length: "); 00954 if (ret != E_OK) { 00955 return ret; 00956 } 00957 00958 // From now on, ret is not needed, we can use it 00959 // to keep the final result 00960 ret = 0; 00961 while (true) { 00962 while (_client->available()) { 00963 char c = _client->read(); 00964 DBG("%c", c); 00965 00966 if ((c == '\r') || (c == '\n')) { 00967 return (ret == 0) ? (E_INVALID) : (ret); 00968 } else { 00969 ret = ret * 10 + (c - '0'); 00970 } 00971 } 00972 00973 if (!_client->connected()) { 00974 DBGLN("%s", "ERROR: The client is disconnected from the server!"); 00975 00976 return E_DISCONNECTED; 00977 } 00978 if (_idlefunc!=NULL) 00979 _idlefunc(); 00980 else 00981 delay(200); 00982 } 00983 00984 // never reached here 00985 return E_NOTREACHABLE; 00986 } 00987 00988 int M2XStreamClient::skipHttpHeader() { 00989 return waitForString("\n\r\n"); 00990 } 00991 00992 void M2XStreamClient::close() { 00993 // Eats up buffered data before closing 00994 _client->flush(); 00995 _client->stop(); 00996 } 00997 00998 static inline int fill_iso8601_timestamp(int32_t seconds, int32_t milli, 00999 char* buffer, int* length); 01000 01001 TimeService::TimeService(M2XStreamClient* client) : _client(client) { 01002 } 01003 01004 int TimeService::init() { 01005 _timer.start(); 01006 return reset(); 01007 } 01008 01009 int TimeService::reset() { 01010 int32_t ts; 01011 int status = _client->getTimestamp32(&ts); 01012 01013 if (m2x_status_is_success(status)) { 01014 _server_timestamp = ts; 01015 _local_last_milli = _timer.read_ms(); 01016 } 01017 01018 return status; 01019 } 01020 01021 int TimeService::getTimestamp(char* buffer, int* length) { 01022 uint32_t now = _timer.read_ms(); 01023 if (now < _local_last_milli) { 01024 // In case of a timestamp overflow(happens once every 50 days on 01025 // Arduino), we reset the server timestamp recorded. 01026 // NOTE: while on an Arduino this might be okay, the situation is worse 01027 // on mbed, see the notes in m2x-mbed.h for details 01028 int status = reset(); 01029 if (!m2x_status_is_success(status)) { return status; } 01030 now = _timer.read_ms(); 01031 } 01032 if (now < _local_last_milli) { 01033 // We have already reseted the timestamp, so this cannot happen 01034 // (an HTTP request can take longer than 50 days to finished? You 01035 // must be kidding here). Something else must be wrong here 01036 return E_TIMESTAMP_ERROR; 01037 } 01038 uint32_t diff = now - _local_last_milli; 01039 _local_last_milli = now; 01040 _server_timestamp += (int32_t) (diff / 1000); // Milliseconds to seconds 01041 return fill_iso8601_timestamp(_server_timestamp, (int32_t) (diff % 1000), 01042 buffer, length); 01043 } 01044 01045 #define SIZE_ISO_8601 25 01046 static inline bool is_leap_year(int16_t y) { 01047 return ((1970 + y) > 0) && 01048 !((1970 + y) % 4) && 01049 (((1970 + y) % 100) || !((1970 + y) % 400)); 01050 } 01051 static inline int32_t days_in_year(int16_t y) { 01052 return is_leap_year(y) ? 366 : 365; 01053 } 01054 static const uint8_t MONTH_DAYS[]={31,28,31,30,31,30,31,31,30,31,30,31}; 01055 01056 static inline int fill_iso8601_timestamp(int32_t timestamp, int32_t milli, 01057 char* buffer, int* length) { 01058 int16_t year; 01059 int8_t month, month_length; 01060 int32_t day; 01061 int8_t hour, minute, second; 01062 01063 if (*length < SIZE_ISO_8601) { 01064 *length = SIZE_ISO_8601; 01065 return E_BUFFER_TOO_SMALL; 01066 } 01067 01068 second = timestamp % 60; 01069 timestamp /= 60; // now it is minutes 01070 01071 minute = timestamp % 60; 01072 timestamp /= 60; // now it is hours 01073 01074 hour = timestamp % 24; 01075 timestamp /= 24; // now it is days 01076 01077 year = 0; 01078 day = 0; 01079 while ((day += days_in_year(year)) <= timestamp) { 01080 year++; 01081 } 01082 day -= days_in_year(year); 01083 timestamp -= day; // now it is days in this year, starting at 0 01084 01085 day = 0; 01086 month_length = 0; 01087 for (month = 0; month < 12; month++) { 01088 if (month == 1) { 01089 // February 01090 month_length = is_leap_year(year) ? 29 : 28; 01091 } else { 01092 month_length = MONTH_DAYS[month]; 01093 } 01094 01095 if (timestamp >= month_length) { 01096 timestamp -= month_length; 01097 } else { 01098 break; 01099 } 01100 } 01101 year = 1970 + year; 01102 month++; // offset by 1 01103 day = timestamp + 1; 01104 01105 int i = 0, j = 0; 01106 01107 // NOTE: It seems the snprintf implementation in Arduino has bugs, 01108 // we have to manually piece the string together here. 01109 #define INT_TO_STR(v_, width_) \ 01110 for (j = 0; j < (width_); j++) { \ 01111 buffer[i + (width_) - 1 - j] = '0' + ((v_) % 10); \ 01112 (v_) /= 10; \ 01113 } \ 01114 i += (width_) 01115 01116 INT_TO_STR(year, 4); 01117 buffer[i++] = '-'; 01118 INT_TO_STR(month, 2); 01119 buffer[i++] = '-'; 01120 INT_TO_STR(day, 2); 01121 buffer[i++] = 'T'; 01122 INT_TO_STR(hour, 2); 01123 buffer[i++] = ':'; 01124 INT_TO_STR(minute, 2); 01125 buffer[i++] = ':'; 01126 INT_TO_STR(second, 2); 01127 buffer[i++] = '.'; 01128 INT_TO_STR(milli, 3); 01129 buffer[i++] = 'Z'; 01130 buffer[i++] = '\0'; 01131 01132 #undef INT_TO_STR 01133 01134 *length = i; 01135 return E_OK; 01136 } 01137 01138 /* Reader functions */ 01139 #ifdef M2X_ENABLE_READER 01140 01141 // Data structures and functions used to parse stream values 01142 01143 #define STREAM_BUF_LEN 32 01144 01145 typedef struct { 01146 uint8_t state; 01147 char at_str[STREAM_BUF_LEN + 1]; 01148 char value_str[STREAM_BUF_LEN + 1]; 01149 int index; 01150 01151 stream_value_read_callback callback; 01152 void* context; 01153 } stream_parsing_context_state; 01154 01155 #define WAITING_AT 0x1 01156 #define GOT_AT 0x2 01157 #define WAITING_VALUE 0x4 01158 #define GOT_VALUE 0x8 01159 01160 #define GOT_STREAM (GOT_AT | GOT_VALUE) 01161 #define TEST_GOT_STREAM(state_) (((state_) & GOT_STREAM) == GOT_STREAM) 01162 01163 #define TEST_IS_AT(state_) (((state_) & (WAITING_AT | GOT_AT)) == WAITING_AT) 01164 #define TEST_IS_VALUE(state_) (((state_) & (WAITING_VALUE | GOT_VALUE)) == \ 01165 WAITING_VALUE) 01166 01167 static void on_stream_key_found(jsonlite_callback_context* context, 01168 jsonlite_token* token) 01169 { 01170 stream_parsing_context_state* state = 01171 (stream_parsing_context_state*) context->client_state; 01172 if (strncmp((const char*) token->start, "timestamp", 9) == 0) { 01173 state->state |= WAITING_AT; 01174 } else if ((strncmp((const char*) token->start, "value", 5) == 0) && 01175 (token->start[5] != 's')) { // get rid of "values" 01176 state->state |= WAITING_VALUE; 01177 } 01178 } 01179 01180 static void on_stream_value_found(jsonlite_callback_context* context, 01181 jsonlite_token* token, 01182 int type) 01183 { 01184 stream_parsing_context_state* state = 01185 (stream_parsing_context_state*) context->client_state; 01186 01187 if (TEST_IS_AT(state->state)) { 01188 strncpy(state->at_str, (const char*) token->start, 01189 MIN(token->end - token->start, STREAM_BUF_LEN)); 01190 state->at_str[MIN(token->end - token->start, STREAM_BUF_LEN)] = '\0'; 01191 state->state |= GOT_AT; 01192 } else if (TEST_IS_VALUE(state->state)) { 01193 strncpy(state->value_str, (const char*) token->start, 01194 MIN(token->end - token->start, STREAM_BUF_LEN)); 01195 state->value_str[MIN(token->end - token->start, STREAM_BUF_LEN)] = '\0'; 01196 state->state |= GOT_VALUE; 01197 } 01198 01199 if (TEST_GOT_STREAM(state->state)) { 01200 state->callback(state->at_str, state->value_str, 01201 state->index++, state->context, type); 01202 state->state = 0; 01203 } 01204 } 01205 01206 static void on_stream_string_found(jsonlite_callback_context* context, 01207 jsonlite_token* token) 01208 { 01209 on_stream_value_found(context, token, 1); 01210 } 01211 01212 static void on_stream_number_found(jsonlite_callback_context* context, 01213 jsonlite_token* token) 01214 { 01215 on_stream_value_found(context, token, 2); 01216 } 01217 01218 // Data structures and functions used to parse locations 01219 01220 #define LOCATION_BUF_LEN 20 01221 01222 typedef struct { 01223 uint16_t state; 01224 char name_str[LOCATION_BUF_LEN + 1]; 01225 double latitude; 01226 double longitude; 01227 double elevation; 01228 char timestamp_str[LOCATION_BUF_LEN + 1]; 01229 int index; 01230 01231 location_read_callback callback; 01232 void* context; 01233 } location_parsing_context_state; 01234 01235 #define WAITING_NAME 0x1 01236 #define WAITING_LATITUDE 0x2 01237 #define WAITING_LONGITUDE 0x4 01238 #define WAITING_ELEVATION 0x8 01239 #define WAITING_TIMESTAMP 0x10 01240 01241 #define GOT_NAME 0x20 01242 #define GOT_LATITUDE 0x40 01243 #define GOT_LONGITUDE 0x80 01244 #define GOT_ELEVATION 0x100 01245 #define GOT_TIMESTAMP 0x200 01246 01247 #define GOT_LOCATION (GOT_NAME | GOT_LATITUDE | GOT_LONGITUDE | GOT_ELEVATION | GOT_TIMESTAMP) 01248 #define TEST_GOT_LOCATION(state_) (((state_) & GOT_LOCATION) == GOT_LOCATION) 01249 01250 #define TEST_IS_NAME(state_) (((state_) & (WAITING_NAME | GOT_NAME)) == WAITING_NAME) 01251 #define TEST_IS_LATITUDE(state_) (((state_) & (WAITING_LATITUDE | GOT_LATITUDE)) \ 01252 == WAITING_LATITUDE) 01253 #define TEST_IS_LONGITUDE(state_) (((state_) & (WAITING_LONGITUDE | GOT_LONGITUDE)) \ 01254 == WAITING_LONGITUDE) 01255 #define TEST_IS_ELEVATION(state_) (((state_) & (WAITING_ELEVATION | GOT_ELEVATION)) \ 01256 == WAITING_ELEVATION) 01257 #define TEST_IS_TIMESTAMP(state_) (((state_) & (WAITING_TIMESTAMP | GOT_TIMESTAMP)) \ 01258 == WAITING_TIMESTAMP) 01259 01260 static void on_location_key_found(jsonlite_callback_context* context, 01261 jsonlite_token* token) { 01262 location_parsing_context_state* state = 01263 (location_parsing_context_state*) context->client_state; 01264 if (strncmp((const char*) token->start, "waypoints", 9) == 0) { 01265 // only parses those locations in waypoints, skip the outer one 01266 state->state = 0; 01267 } else if (strncmp((const char*) token->start, "name", 4) == 0) { 01268 state->state |= WAITING_NAME; 01269 } else if (strncmp((const char*) token->start, "latitude", 8) == 0) { 01270 state->state |= WAITING_LATITUDE; 01271 } else if (strncmp((const char*) token->start, "longitude", 9) == 0) { 01272 state->state |= WAITING_LONGITUDE; 01273 } else if (strncmp((const char*) token->start, "elevation", 9) == 0) { 01274 state->state |= WAITING_ELEVATION; 01275 } else if (strncmp((const char*) token->start, "timestamp", 9) == 0) { 01276 state->state |= WAITING_TIMESTAMP; 01277 } 01278 } 01279 01280 static void on_location_string_found(jsonlite_callback_context* context, 01281 jsonlite_token* token) { 01282 location_parsing_context_state* state = 01283 (location_parsing_context_state*) context->client_state; 01284 01285 if (TEST_IS_NAME(state->state)) { 01286 strncpy(state->name_str, (const char*) token->start, 01287 MIN(token->end - token->start, LOCATION_BUF_LEN)); 01288 state->name_str[MIN(token->end - token->start, LOCATION_BUF_LEN)] = '\0'; 01289 state->state |= GOT_NAME; 01290 } else if (TEST_IS_LATITUDE(state->state)) { 01291 state->latitude = atof((const char*) token->start); 01292 state->state |= GOT_LATITUDE; 01293 } else if (TEST_IS_LONGITUDE(state->state)) { 01294 state->longitude = atof((const char*) token->start); 01295 state->state |= GOT_LONGITUDE; 01296 } else if (TEST_IS_ELEVATION(state->state)) { 01297 state->elevation = atof((const char*) token->start); 01298 state->state |= GOT_ELEVATION; 01299 } else if (TEST_IS_TIMESTAMP(state->state)) { 01300 strncpy(state->timestamp_str, (const char*) token->start, 01301 MIN(token->end - token->start, LOCATION_BUF_LEN)); 01302 state->timestamp_str[MIN(token->end - token->start, LOCATION_BUF_LEN)] = '\0'; 01303 state->state |= GOT_TIMESTAMP; 01304 } 01305 01306 if (TEST_GOT_LOCATION(state->state)) { 01307 state->callback(state->name_str, state->latitude, state->longitude, 01308 state->elevation, state->timestamp_str, state->index++, 01309 state->context); 01310 state->state = 0; 01311 } 01312 } 01313 01314 #ifndef M2X_COMMAND_BUF_LEN 01315 #define M2X_COMMAND_BUF_LEN 40 01316 #endif /* M2X_COMMAND_BUF_LEN */ 01317 01318 typedef struct { 01319 uint8_t state; 01320 char id_str[M2X_COMMAND_BUF_LEN + 1]; 01321 char name_str[M2X_COMMAND_BUF_LEN + 1]; 01322 int index; 01323 01324 m2x_command_read_callback callback; 01325 void* context; 01326 } m2x_command_parsing_context_state; 01327 01328 #define M2X_COMMAND_WAITING_ID 0x1 01329 #define M2X_COMMAND_GOT_ID 0x2 01330 #define M2X_COMMAND_WAITING_NAME 0x4 01331 #define M2X_COMMAND_GOT_NAME 0x8 01332 01333 #define M2X_COMMAND_GOT_COMMAND (M2X_COMMAND_GOT_ID | M2X_COMMAND_GOT_NAME) 01334 #define M2X_COMMAND_TEST_GOT_COMMAND(state_) \ 01335 (((state_) & M2X_COMMAND_GOT_COMMAND) == M2X_COMMAND_GOT_COMMAND) 01336 01337 #define M2X_COMMAND_TEST_IS_ID(state_) \ 01338 (((state_) & (M2X_COMMAND_WAITING_ID | M2X_COMMAND_GOT_ID)) == \ 01339 M2X_COMMAND_WAITING_ID) 01340 #define M2X_COMMAND_TEST_IS_NAME(state_) \ 01341 (((state_) & (M2X_COMMAND_WAITING_NAME | M2X_COMMAND_GOT_NAME)) == \ 01342 M2X_COMMAND_WAITING_NAME) 01343 01344 static void m2x_on_command_key_found(jsonlite_callback_context* context, 01345 jsonlite_token* token) 01346 { 01347 m2x_command_parsing_context_state* state = 01348 (m2x_command_parsing_context_state*) context->client_state; 01349 if (strncmp((const char*) token->start, "id", 2) == 0) { 01350 state->state |= M2X_COMMAND_WAITING_ID; 01351 } else if (strncmp((const char*) token->start, "name", 4) == 0) { 01352 state->state |= M2X_COMMAND_WAITING_NAME; 01353 } 01354 } 01355 01356 static void m2x_on_command_value_found(jsonlite_callback_context* context, 01357 jsonlite_token* token) 01358 { 01359 m2x_command_parsing_context_state* state = 01360 (m2x_command_parsing_context_state*) context->client_state; 01361 if (M2X_COMMAND_TEST_IS_ID(state->state)) { 01362 strncpy(state->id_str, (const char*) token->start, 01363 MIN(token->end - token->start, M2X_COMMAND_BUF_LEN)); 01364 state->id_str[MIN(token->end - token->start, M2X_COMMAND_BUF_LEN)] = '\0'; 01365 state->state |= M2X_COMMAND_GOT_ID; 01366 } else if (M2X_COMMAND_TEST_IS_NAME(state->state)) { 01367 strncpy(state->name_str, (const char*) token->start, 01368 MIN(token->end - token->start, M2X_COMMAND_BUF_LEN)); 01369 state->name_str[MIN(token->end - token->start, M2X_COMMAND_BUF_LEN)] = '\0'; 01370 state->state |= M2X_COMMAND_GOT_NAME; 01371 } 01372 01373 if (M2X_COMMAND_TEST_GOT_COMMAND(state->state)) { 01374 state->callback(state->id_str, state->name_str, 01375 state->index++, state->context); 01376 state->state = 0; 01377 } 01378 } 01379 01380 int M2XStreamClient::listStreamValues(const char* deviceId, const char* streamName, 01381 stream_value_read_callback callback, void* context, 01382 const char* query) { 01383 if (_client->connect(_host, _port)) { 01384 DBGLN("%s", "Connected to M2X server!"); 01385 _client->print("GET "); 01386 if (_path_prefix) { _client->print(_path_prefix); } 01387 _client->print("/v2/devices/"); 01388 _client->print(deviceId); 01389 _client->print("/streams/"); 01390 print_encoded_string(_client, streamName); 01391 _client->print("/values"); 01392 01393 if (query) { 01394 if (query[0] != '?') { 01395 _client->print('?'); 01396 } 01397 _client->print(query); 01398 } 01399 01400 _client->println(" HTTP/1.1"); 01401 writeHttpHeader(-1); 01402 } else { 01403 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 01404 return E_NOCONNECTION; 01405 } 01406 int status = readStatusCode(false); 01407 if (status == 200) { 01408 readStreamValue(callback, context); 01409 } 01410 01411 close(); 01412 return status; 01413 } 01414 01415 int M2XStreamClient::readLocation(const char* deviceId, 01416 location_read_callback callback, 01417 void* context) { 01418 if (_client->connect(_host, _port)) { 01419 DBGLN("%s", "Connected to M2X server!"); 01420 _client->print("GET "); 01421 if (_path_prefix) { _client->print(_path_prefix); } 01422 _client->print("/v2/devices/"); 01423 _client->print(deviceId); 01424 _client->println("/location HTTP/1.1"); 01425 01426 writeHttpHeader(-1); 01427 } else { 01428 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 01429 return E_NOCONNECTION; 01430 } 01431 int status = readStatusCode(false); 01432 if (status == 200) { 01433 readLocation(callback, context); 01434 } 01435 01436 close(); 01437 return status; 01438 } 01439 01440 int M2XStreamClient::deleteLocations(const char* deviceId, 01441 const char* from, const char* end) { 01442 if (_client->connect(_host, _port)) { 01443 DBGLN("%s", "Connected to M2X server!"); 01444 int length = write_delete_values(&_null_print, from, end); 01445 writeDeleteLocationHeader(deviceId, length); 01446 write_delete_values(_client, from, end); 01447 } else { 01448 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 01449 return E_NOCONNECTION; 01450 } 01451 01452 return readStatusCode(true); 01453 } 01454 01455 int M2XStreamClient::listCommands(const char* deviceId, 01456 m2x_command_read_callback callback, 01457 void* context, 01458 const char* query) { 01459 if (_client->connect(_host, _port)) { 01460 DBGLN("%s", "Connected to M2X server!"); 01461 _client->print("GET "); 01462 if (_path_prefix) { _client->print(_path_prefix); } 01463 _client->print("/v2/devices/"); 01464 _client->print(deviceId); 01465 _client->print("/commands"); 01466 01467 if (query) { 01468 if (query[0] != '?') { 01469 _client->print('?'); 01470 } 01471 _client->print(query); 01472 } 01473 01474 _client->println(" HTTP/1.1"); 01475 writeHttpHeader(-1); 01476 } else { 01477 DBGLN("%s", "ERROR: Cannot connect to M2X server!"); 01478 return E_NOCONNECTION; 01479 } 01480 int status = readStatusCode(false); 01481 if (status == 200) { 01482 readCommand(callback, context); 01483 } 01484 01485 close(); 01486 return status; 01487 } 01488 01489 01490 int M2XStreamClient::readStreamValue(stream_value_read_callback callback, 01491 void* context) { 01492 const int BUF_LEN = 64; 01493 char buf[BUF_LEN]; 01494 01495 int length = readContentLength(); 01496 if (length < 0) { 01497 close(); 01498 return length; 01499 } 01500 01501 int index = skipHttpHeader(); 01502 if (index != E_OK) { 01503 close(); 01504 return index; 01505 } 01506 index = 0; 01507 01508 stream_parsing_context_state state; 01509 state.state = state.index = 0; 01510 state.callback = callback; 01511 state.context = context; 01512 01513 jsonlite_parser_callbacks cbs = jsonlite_default_callbacks; 01514 cbs.key_found = on_stream_key_found; 01515 cbs.number_found = on_stream_number_found; 01516 cbs.string_found = on_stream_string_found; 01517 cbs.context.client_state = &state; 01518 01519 jsonlite_parser p = jsonlite_parser_init(jsonlite_parser_estimate_size(5)); 01520 jsonlite_parser_set_callback(p, &cbs); 01521 01522 jsonlite_result result = jsonlite_result_unknown; 01523 while (index < length) { 01524 int i = 0; 01525 01526 DBG("%s", "Received Data: "); 01527 while ((i < BUF_LEN) && _client->available()) { 01528 buf[i++] = _client->read(); 01529 DBG("%c", buf[i - 1]); 01530 } 01531 DBGLNEND; 01532 01533 if ((!_client->connected()) && 01534 (!_client->available()) && 01535 ((index + i) < length)) { 01536 jsonlite_parser_release(p); 01537 close(); 01538 return E_NOCONNECTION; 01539 } 01540 01541 result = jsonlite_parser_tokenize(p, buf, i); 01542 if ((result != jsonlite_result_ok) && 01543 (result != jsonlite_result_end_of_stream)) { 01544 jsonlite_parser_release(p); 01545 close(); 01546 return E_JSON_INVALID; 01547 } 01548 01549 index += i; 01550 } 01551 01552 jsonlite_parser_release(p); 01553 close(); 01554 return (result == jsonlite_result_ok) ? (E_OK) : (E_JSON_INVALID); 01555 } 01556 01557 int M2XStreamClient::readLocation(location_read_callback callback, 01558 void* context) { 01559 const int BUF_LEN = 40; 01560 char buf[BUF_LEN]; 01561 01562 int length = readContentLength(); 01563 if (length < 0) { 01564 close(); 01565 return length; 01566 } 01567 01568 int index = skipHttpHeader(); 01569 if (index != E_OK) { 01570 close(); 01571 return index; 01572 } 01573 index = 0; 01574 01575 location_parsing_context_state state; 01576 state.state = state.index = 0; 01577 state.callback = callback; 01578 state.context = context; 01579 01580 jsonlite_parser_callbacks cbs = jsonlite_default_callbacks; 01581 cbs.key_found = on_location_key_found; 01582 cbs.string_found = on_location_string_found; 01583 cbs.context.client_state = &state; 01584 01585 jsonlite_parser p = jsonlite_parser_init(jsonlite_parser_estimate_size(5)); 01586 jsonlite_parser_set_callback(p, &cbs); 01587 01588 jsonlite_result result = jsonlite_result_unknown; 01589 while (index < length) { 01590 int i = 0; 01591 01592 DBG("%s", "Received Data: "); 01593 while ((i < BUF_LEN) && _client->available()) { 01594 buf[i++] = _client->read(); 01595 DBG("%c", buf[i - 1]); 01596 } 01597 DBGLNEND; 01598 01599 if ((!_client->connected()) && 01600 (!_client->available()) && 01601 ((index + i) < length)) { 01602 jsonlite_parser_release(p); 01603 close(); 01604 return E_NOCONNECTION; 01605 } 01606 01607 result = jsonlite_parser_tokenize(p, buf, i); 01608 if ((result != jsonlite_result_ok) && 01609 (result != jsonlite_result_end_of_stream)) { 01610 jsonlite_parser_release(p); 01611 close(); 01612 return E_JSON_INVALID; 01613 } 01614 01615 index += i; 01616 } 01617 01618 jsonlite_parser_release(p); 01619 close(); 01620 return (result == jsonlite_result_ok) ? (E_OK) : (E_JSON_INVALID); 01621 } 01622 01623 int M2XStreamClient::readCommand(m2x_command_read_callback callback, 01624 void* context) { 01625 const int BUF_LEN = 60; 01626 char buf[BUF_LEN]; 01627 01628 int length = readContentLength(); 01629 if (length < 0) { 01630 close(); 01631 return length; 01632 } 01633 01634 int index = skipHttpHeader(); 01635 if (index != E_OK) { 01636 close(); 01637 return index; 01638 } 01639 index = 0; 01640 01641 m2x_command_parsing_context_state state; 01642 state.state = 0; 01643 state.index = 0; 01644 state.callback = callback; 01645 state.context = context; 01646 01647 jsonlite_parser_callbacks cbs = jsonlite_default_callbacks; 01648 cbs.key_found = m2x_on_command_key_found; 01649 cbs.string_found = m2x_on_command_value_found; 01650 cbs.context.client_state = &state; 01651 01652 jsonlite_parser p = jsonlite_parser_init(jsonlite_parser_estimate_size(5)); 01653 jsonlite_parser_set_callback(p, &cbs); 01654 01655 jsonlite_result result = jsonlite_result_unknown; 01656 while (index < length) { 01657 int i = 0; 01658 01659 DBG("%s", "Received Data: "); 01660 while ((i < BUF_LEN) && _client->available()) { 01661 buf[i++] = _client->read(); 01662 DBG("%c", buf[i - 1]); 01663 } 01664 DBGLNEND; 01665 01666 if ((!_client->connected()) && 01667 (!_client->available()) && 01668 ((index + i) < length)) { 01669 jsonlite_parser_release(p); 01670 close(); 01671 return E_NOCONNECTION; 01672 } 01673 01674 result = jsonlite_parser_tokenize(p, buf, i); 01675 if ((result != jsonlite_result_ok) && 01676 (result != jsonlite_result_end_of_stream)) { 01677 jsonlite_parser_release(p); 01678 close(); 01679 return E_JSON_INVALID; 01680 } 01681 01682 index += i; 01683 } 01684 01685 jsonlite_parser_release(p); 01686 close(); 01687 return (result == jsonlite_result_ok) ? (E_OK) : (E_JSON_INVALID); 01688 } 01689 01690 #endif /* M2X_ENABLE_READER */ 01691 01692 #endif /* M2XStreamClient_h */
Generated on Sun Jul 24 2022 18:17:44 by
1.7.2