messaging.c 9.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305
  1. /**
  2. *
  3. */
  4. #include <stdlib.h> // Required for libtelnet.h
  5. #include <esp_log.h>
  6. #include "stdbool.h"
  7. #include <lwip/def.h>
  8. #include <lwip/sockets.h>
  9. #include <errno.h>
  10. #include <string.h>
  11. #include "esp_app_trace.h"
  12. #include "esp_attr.h"
  13. #include "config.h"
  14. #include "nvs_utilities.h"
  15. #include "platform_esp32.h"
  16. #include "messaging.h"
  17. #include "tools.h"
  18. /************************************
  19. * Globals
  20. */
  21. const static char tag[] = "messaging";
  22. typedef struct {
  23. struct messaging_list_t * next;
  24. char * subscriber_name;
  25. size_t max_count;
  26. RingbufHandle_t buf_handle;
  27. } messaging_list_t;
  28. static messaging_list_t top;
  29. #define MSG_LENGTH_AVG 1024
  30. messaging_list_t * get_struct_ptr(messaging_handle_t handle){
  31. return (messaging_list_t *)handle;
  32. }
  33. messaging_handle_t get_handle_ptr(messaging_list_t * handle){
  34. return (messaging_handle_t )handle;
  35. }
  36. RingbufHandle_t messaging_create_ring_buffer(uint8_t max_count){
  37. RingbufHandle_t buf_handle = NULL;
  38. StaticRingbuffer_t *buffer_struct = malloc_init_external(sizeof(StaticRingbuffer_t));
  39. if (buffer_struct != NULL) {
  40. size_t buf_size = (size_t )(sizeof(single_message_t)+8+MSG_LENGTH_AVG)*(size_t )(max_count>0?max_count:5); // no-split buffer requires an additional 8 bytes
  41. buf_size = buf_size - (buf_size % 4);
  42. uint8_t *buffer_storage = (uint8_t *)heap_caps_malloc(buf_size, MALLOC_CAP_SPIRAM | MALLOC_CAP_32BIT);
  43. if (buffer_storage== NULL) {
  44. ESP_LOGE(tag,"buff alloc failed");
  45. }
  46. else {
  47. buf_handle = xRingbufferCreateStatic(buf_size, RINGBUF_TYPE_NOSPLIT, buffer_storage, buffer_struct);
  48. }
  49. }
  50. else {
  51. ESP_LOGE(tag,"ringbuf alloc failed");
  52. }
  53. return buf_handle;
  54. }
  55. void messaging_fill_messages(messaging_list_t * target_subscriber){
  56. single_message_t * message=NULL;
  57. UBaseType_t uxItemsWaiting;
  58. vRingbufferGetInfo(top.buf_handle, NULL, NULL, NULL, NULL, &uxItemsWaiting);
  59. for(size_t i=0;i<uxItemsWaiting;i++){
  60. message= messaging_retrieve_message(top.buf_handle);
  61. if(message){
  62. //re-post to original queue so it is available to future subscribers
  63. messaging_post_to_queue(get_handle_ptr(&top), message, message->msg_size);
  64. // post to new subscriber
  65. messaging_post_to_queue(get_handle_ptr(target_subscriber) , message, message->msg_size);
  66. FREE_AND_NULL(message);
  67. }
  68. }
  69. }
  70. messaging_handle_t messaging_register_subscriber(uint8_t max_count, char * name){
  71. messaging_list_t * cur=&top;
  72. while(cur->next){
  73. cur = get_struct_ptr(cur->next);
  74. }
  75. cur->next=malloc_init_external(sizeof(messaging_list_t));
  76. if(!cur->next){
  77. ESP_LOGE(tag,"subscriber alloc failed");
  78. return NULL;
  79. }
  80. memset(cur->next,0x00,sizeof(messaging_list_t));
  81. cur = get_struct_ptr(cur->next);
  82. cur->max_count=max_count;
  83. cur->subscriber_name=strdup_psram(name);
  84. cur->buf_handle = messaging_create_ring_buffer(max_count);
  85. if(cur->buf_handle){
  86. messaging_fill_messages(cur);
  87. }
  88. return cur->buf_handle;
  89. }
  90. void messaging_service_init(){
  91. size_t max_count=15;
  92. top.buf_handle = messaging_create_ring_buffer(max_count);
  93. if(!top.buf_handle){
  94. ESP_LOGE(tag, "messaging service init failed.");
  95. }
  96. else {
  97. top.max_count = max_count;
  98. top.subscriber_name = strdup_psram("messaging");
  99. }
  100. return;
  101. }
  102. const char * messaging_get_type_desc(messaging_types msg_type){
  103. switch (msg_type) {
  104. CASE_TO_STR(MESSAGING_INFO);
  105. CASE_TO_STR(MESSAGING_WARNING);
  106. CASE_TO_STR(MESSAGING_ERROR);
  107. default:
  108. return "Unknown";
  109. break;
  110. }
  111. }
  112. const char * messaging_get_class_desc(messaging_classes msg_class){
  113. switch (msg_class) {
  114. CASE_TO_STR(MESSAGING_CLASS_OTA);
  115. CASE_TO_STR(MESSAGING_CLASS_SYSTEM);
  116. CASE_TO_STR(MESSAGING_CLASS_STATS);
  117. CASE_TO_STR(MESSAGING_CLASS_CFGCMD);
  118. CASE_TO_STR(MESSAGING_CLASS_BT);
  119. default:
  120. return "Unknown";
  121. break;
  122. }
  123. }
  124. cJSON * messaging_retrieve_messages(RingbufHandle_t buf_handle){
  125. single_message_t * message=NULL;
  126. cJSON * json_messages=cJSON_CreateArray();
  127. cJSON * json_message=NULL;
  128. size_t item_size;
  129. UBaseType_t uxItemsWaiting;
  130. vRingbufferGetInfo(buf_handle, NULL, NULL, NULL, NULL, &uxItemsWaiting);
  131. for(int i = 0;i<uxItemsWaiting;i++){
  132. message = (single_message_t *)xRingbufferReceive(buf_handle, &item_size, pdMS_TO_TICKS(50));
  133. //Check received data
  134. if (message== NULL) {
  135. ESP_LOGE(tag,"received null ptr");
  136. }
  137. else {
  138. json_message = cJSON_CreateObject();
  139. cJSON_AddStringToObject(json_message, "message", message->message);
  140. cJSON_AddStringToObject(json_message, "type", messaging_get_type_desc(message->type));
  141. cJSON_AddStringToObject(json_message, "class", messaging_get_class_desc(message->msg_class));
  142. cJSON_AddNumberToObject(json_message,"sent_time",message->sent_time);
  143. cJSON_AddNumberToObject(json_message,"current_time",esp_timer_get_time() / 1000);
  144. cJSON_AddItemToArray(json_messages,json_message);
  145. vRingbufferReturnItem(buf_handle, (void *)message);
  146. }
  147. }
  148. return json_messages;
  149. }
  150. single_message_t * messaging_retrieve_message(RingbufHandle_t buf_handle){
  151. single_message_t * message=NULL;
  152. single_message_t * message_copy=NULL;
  153. size_t item_size;
  154. UBaseType_t uxItemsWaiting;
  155. vRingbufferGetInfo(buf_handle, NULL, NULL, NULL, NULL, &uxItemsWaiting);
  156. if(uxItemsWaiting>0){
  157. message = (single_message_t *)xRingbufferReceive(buf_handle, &item_size, pdMS_TO_TICKS(50));
  158. message_copy = clone_obj_psram(message,item_size);
  159. vRingbufferReturnItem(buf_handle, (void *)message);
  160. }
  161. return message_copy;
  162. }
  163. esp_err_t messaging_post_to_queue(messaging_handle_t subscriber_handle, single_message_t * message, size_t message_size){
  164. size_t item_size=0;
  165. messaging_list_t * subscriber=get_struct_ptr(subscriber_handle);
  166. if(!subscriber->buf_handle){
  167. ESP_LOGE(tag,"post failed: null buffer for %s", str_or_unknown(subscriber->subscriber_name));
  168. return ESP_FAIL;
  169. }
  170. void * pItem=NULL;
  171. UBaseType_t res=pdFALSE;
  172. while(1){
  173. ESP_LOGD(tag,"Attempting to reserve %d bytes for %s",message_size, str_or_unknown(subscriber->subscriber_name));
  174. res = xRingbufferSendAcquire(subscriber->buf_handle, &pItem, message_size, pdMS_TO_TICKS(50));
  175. if(res == pdTRUE && pItem){
  176. ESP_LOGD(tag,"Reserving complete for %s", str_or_unknown(subscriber->subscriber_name));
  177. memcpy(pItem,message,message_size);
  178. xRingbufferSendComplete(subscriber->buf_handle, pItem);
  179. break;
  180. }
  181. ESP_LOGD(tag,"Dropping for %s",str_or_unknown(subscriber->subscriber_name));
  182. single_message_t * dummy = (single_message_t *)xRingbufferReceive(subscriber->buf_handle, &item_size, pdMS_TO_TICKS(50));
  183. if (dummy== NULL) {
  184. ESP_LOGE(tag,"Dropping message failed");
  185. break;
  186. }
  187. else {
  188. ESP_LOGD(tag,"Dropping message of %d bytes for %s",item_size, str_or_unknown(subscriber->subscriber_name));
  189. vRingbufferReturnItem(subscriber->buf_handle, (void *)dummy);
  190. }
  191. }
  192. if (res != pdTRUE) {
  193. ESP_LOGE(tag,"post to %s failed",str_or_unknown(subscriber->subscriber_name));
  194. return ESP_FAIL;
  195. }
  196. return ESP_OK;
  197. }
  198. esp_err_t messaging_type_to_err_type(messaging_types type){
  199. switch (type) {
  200. case MESSAGING_INFO:
  201. return ESP_LOG_INFO;
  202. break;
  203. case MESSAGING_ERROR:
  204. return ESP_LOG_ERROR;
  205. break;
  206. case MESSAGING_WARNING:
  207. return ESP_LOG_WARN;
  208. break;
  209. default:
  210. return ESP_LOG_DEBUG;
  211. break;
  212. }
  213. return ESP_LOG_DEBUG;
  214. }
  215. void messaging_post_message(messaging_types type,messaging_classes msg_class, const char *fmt, ...){
  216. single_message_t * message=NULL;
  217. size_t msg_size=0;
  218. size_t ln =0;
  219. messaging_list_t * cur=&top;
  220. va_list va;
  221. va_start(va, fmt);
  222. ln = vsnprintf(NULL, 0, fmt, va)+1;
  223. msg_size = sizeof(single_message_t)+ln;
  224. message = (single_message_t *)malloc_init_external(msg_size);
  225. vsprintf(message->message, fmt, va);
  226. va_end(va);
  227. message->msg_size = msg_size;
  228. message->type = type;
  229. message->msg_class = msg_class;
  230. message->sent_time = esp_timer_get_time() / 1000;
  231. if(type==MESSAGING_WARNING) {
  232. ESP_LOGW(tag,"%s",message->message);
  233. }
  234. else if(type==MESSAGING_ERROR) {
  235. ESP_LOGE(tag,"%s",message->message);
  236. }
  237. else {
  238. ESP_LOGD(tag,"Post: %s",message->message);
  239. }
  240. while(cur){
  241. messaging_post_to_queue(get_handle_ptr(cur), message, msg_size);
  242. cur = get_struct_ptr(cur->next);
  243. }
  244. FREE_AND_NULL(message);
  245. return;
  246. }
  247. char * messaging_alloc_format_string(const char *fmt, ...) {
  248. va_list va;
  249. va_start(va, fmt);
  250. size_t ln = vsnprintf(NULL, 0, fmt, va)+1;
  251. char * message_txt = malloc_init_external(ln);
  252. if(message_txt){
  253. vsprintf(message_txt, fmt, va);
  254. va_end(va);
  255. }
  256. else{
  257. ESP_LOGE(tag, "Memory allocation failed while sending message");
  258. }
  259. return message_txt;
  260. }
  261. void log_send_messaging(messaging_types msgtype,const char *fmt, ...) {
  262. va_list va;
  263. va_start(va, fmt);
  264. size_t ln = vsnprintf(NULL, 0, fmt, va)+1;
  265. char * message_txt = malloc_init_external(ln);
  266. if(message_txt){
  267. vsprintf(message_txt, fmt, va);
  268. va_end(va);
  269. ESP_LOG_LEVEL_LOCAL(messaging_type_to_err_type(msgtype),tag, "%s",message_txt);
  270. messaging_post_message(msgtype, MESSAGING_CLASS_SYSTEM, message_txt );
  271. free(message_txt);
  272. }
  273. else{
  274. ESP_LOGE(tag, "Memory allocation failed while sending message");
  275. }
  276. }
  277. void cmd_send_messaging(const char * cmdname,messaging_types msgtype, const char *fmt, ...){
  278. va_list va;
  279. va_start(va, fmt);
  280. size_t cmd_len = strlen(cmdname)+1;
  281. size_t ln = vsnprintf(NULL, 0, fmt, va)+1;
  282. char * message_txt = malloc_init_external(ln+cmd_len);
  283. if(message_txt){
  284. strcpy(message_txt,cmdname);
  285. strcat(message_txt,"\n");
  286. vsprintf((message_txt+cmd_len), fmt, va);
  287. va_end(va);
  288. ESP_LOG_LEVEL_LOCAL(messaging_type_to_err_type(msgtype),tag, "%s",message_txt);
  289. messaging_post_message(msgtype, MESSAGING_CLASS_CFGCMD, message_txt );
  290. free(message_txt);
  291. }
  292. else{
  293. ESP_LOGE(tag, "Memory allocation failed while sending message");
  294. }
  295. }