1 //******************************************************************
3 // Copyright 2015 Intel Mobile Communications GmbH All Rights Reserved.
5 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
7 // Licensed under the Apache License, Version 2.0 (the "License");
8 // you may not use this file except in compliance with the License.
9 // You may obtain a copy of the License at
11 // http://www.apache.org/licenses/LICENSE-2.0
13 // Unless required by applicable law or agreed to in writing, software
14 // distributed under the License is distributed on an "AS IS" BASIS,
15 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 // See the License for the specific language governing permissions and
17 // limitations under the License.
19 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
21 #include "telegesis_socket.h"
27 #include <sys/select.h>
30 #include "twsocketlist.h"
31 #include "oic_string.h"
32 #include "oic_malloc.h"
35 #define TAG "Telegesis_Socket"
38 * New thread's main() ftn.
40 void * readForever(/*PIPlugin_Zigbee */void * plugin);
42 * Just grabs the next char in the buffer. Called by readBufferLine() multiple times.
44 char readBufferChar(int fd, ssize_t * readDataBytes);
46 * Calls readBufferChar() until line is formed.
48 const char * readBufferLine(int fd);
50 * Calls readBufferLine() until a full TWEntry is formed.
52 TWEntry * readEntry(int fd);
54 * Posts the TWEntry to the queue.
56 TWResultCode TWEnqueueEntry(PIPlugin_Zigbee * plugin, TWEntry * entry);
59 * Defines the mapping relationships between Telegesis AT response/prompt tags and
60 * how many lines we should expect with each following the tag.
64 const char *resultTxt;
66 TWEntryType entryType;
69 static TWEntryTypePair TWEntryTypePairArray[] =
71 {"+N=", 1, TW_NETWORK_INFO},
72 {"JPAN:", 1, TW_JPAN},
77 {"MatchDesc:", 1, TW_MATCHDESC},
78 {"SimpleDesc:", 6, TW_SIMPLEDESC},
79 {"InCluster:", 1, TW_INCLUSTER},
80 {"WRITEATTR:", 1, TW_WRITEATTR},
81 {"RESPATTR:", 1, TW_RESPATTR},
82 {"TEMPERATURE:", 1, TW_RESPATTR},
83 {"DFTREP", 1, TW_DFTREP},
84 {"DRLOCRSP:", 1, TW_DRLOCRSP},
85 {"DRUNLOCKRSP:", 1, TW_DRUNLOCKRSP},
87 {"NACK:", 1, TW_NACK},
89 {"ZENROLLREQ:", 1, TW_ZENROLLREQ},
90 {"ENROLLED:", 1, TW_ENROLLED},
91 {"ZONESTATUS:", 1, TW_ZONESTATUS},
92 {"AddrResp:", 1, TW_ADDRESS_RESPONSE},
93 {"REPORTATTR:", 1, TW_NONE},
94 {"Unknown:", 0, TW_NONE},
95 {"Unknown:", 1, TW_MAX_ENTRY}
98 TWEntry * TWEntryList = NULL;
100 TWEntryTypePair getEntryTypePair(const char * bufferLine)
102 size_t count = sizeof(TWEntryTypePairArray)/sizeof(TWEntryTypePairArray[0]);
105 return TWEntryTypePairArray[count-1];
107 size_t bufferLength = strlen(bufferLine);
108 for (size_t i = 0; i < count; i++)
110 if (!TWEntryTypePairArray[i].resultTxt)
112 return TWEntryTypePairArray[count-1];
114 size_t resultTxtLength = strlen(TWEntryTypePairArray[i].resultTxt);
115 if ((bufferLength >= resultTxtLength) &&
116 strncmp(bufferLine, TWEntryTypePairArray[i].resultTxt, resultTxtLength) == 0)
118 return TWEntryTypePairArray[i];
121 return TWEntryTypePairArray[count-1];
124 TWResultCode TWWait(pthread_cond_t * cond, pthread_mutex_t * mutex, uint8_t timeout)
127 // This is a blocking call which hold the calling thread until an entry
128 // has been enqueued or until the specified timeout has surpassed.
129 struct timespec abs_time;
130 clock_gettime(CLOCK_REALTIME , &abs_time);
131 abs_time.tv_sec += timeout;
132 ret = pthread_cond_timedwait(cond, mutex, &abs_time);
139 OIC_LOG(INFO, TAG, "Timed out waiting for CV. Non-error. Please try again.");
142 OIC_LOG(ERROR, TAG, "Cond or Mutex is invalid. OR timeout value for CV is invalid.");
145 OIC_LOG(ERROR, TAG, "Cannot use CV because the current thread does not own the CV.");
149 return TW_RESULT_ERROR;
152 TWResultCode TWGrabMutex(pthread_mutex_t * mutex)
154 int ret = pthread_mutex_lock(mutex);
161 OIC_LOG(ERROR, TAG, "Mutex was not initialized.");
164 OIC_LOG(INFO, TAG, "Timed out waiting for mutex. Non-error. Please try again.");
167 OIC_LOG(ERROR, TAG, "Maximum number of locks for mutex have been exceeded.");
170 OIC_LOG(ERROR, TAG, "Deadlock OR this thread already owns the mutex.");
173 return TW_RESULT_ERROR;
176 TWResultCode TWReleaseMutex(pthread_mutex_t * mutex)
178 int ret = pthread_mutex_unlock(mutex);
185 OIC_LOG(ERROR, TAG, "Mutex was not initialized.");
188 OIC_LOG(ERROR, TAG, "Maximum number of locks for mutex have been exceeded.");
191 OIC_LOG(ERROR, TAG, "Cannot unlock because the current thread does not own the mutex.");
194 return TW_RESULT_ERROR;
197 char readBufferChar(int fd, ssize_t * readDataBytes)
199 // Performs read operation on fd for one character at a time.
207 *readDataBytes = read(fd, &byte, sizeof(byte));
208 if(*readDataBytes < 0)
210 OIC_LOG_V(ERROR, TAG, "\tCould not read from port. Errno is: %d\n", errno);
216 bool isLineIgnored(const char * line, size_t length)
220 if(line[0] == 'N' && line[1] == 'A' && line[2] == 'C' && line[3] == 'K')
227 if(line[0] == 'S' && line[1] == 'E' && line[2] == 'Q')
231 else if(line[0] == 'A' && line[1] == 'C' && line[2] == 'K')
238 if(line[0] == 'A' && line[1] == 'T')
240 // If the line starts with "AT", then this is an echo. We ignore echos.
243 else if(line[0] == 'O' && line[1] == 'K')
245 //If this line is "OK", we ignore success codes. But we do end TWEntry's on "OK".
246 // (FYI, error codes are handled in readEntry() which invokes this function.)
253 const char * readBufferLine(int fd)
255 char bufferChar = '\0';
256 size_t bufferLoc = 0;
257 ssize_t readDataBytes = 0;
258 char * bufferLineHold = NULL;
259 char * bufferLine = NULL;
260 bool endOfLine1 = false;
261 bool endOfLine2 = false;
265 if(!endOfLine1 || !endOfLine2)
267 bufferChar = readBufferChar(fd, &readDataBytes);
268 if(bufferChar == '\r')
273 if(bufferChar == '\n')
279 if(readDataBytes != 0 && (!endOfLine1 && !endOfLine2))
281 size_t bufferLineSize = sizeof(bufferChar)*(bufferLoc+2);
282 bufferLine = (char *) OICRealloc(bufferLineHold, bufferLineSize);
285 OIC_LOG(ERROR, TAG, "Ran out of memory.");
288 bufferLine[bufferLoc] = '\0';
289 OICStrcat(bufferLine, bufferLineSize+2, &bufferChar);
291 bufferLineHold = bufferLine;
293 OIC_LOG_V(DEBUG, TAG, "Incoming: %s", bufferLine);
301 size_t bufferLineSize = sizeof(bufferChar)*(bufferLoc+2);
302 bufferLine = (char *) OICRealloc(bufferLineHold, bufferLineSize);
305 OIC_LOG(ERROR, TAG, "Ran out of memory.");
308 bufferLine[bufferLoc] = '\0';
310 bufferLineHold = bufferLine;
311 ignore = isLineIgnored(bufferLine, strlen(bufferLine));
315 return readBufferLine(fd);
317 if(endOfLine1 && endOfLine2)
329 TWResultCode TWAddLineToEntry(const char * line, TWEntry * entry)
333 OIC_LOG(ERROR, TAG, "Invalid/NULL parameter(s) received.");
334 return TW_RESULT_ERROR_INVALID_PARAMS;
336 TWLine * twLine = (TWLine *) OICCalloc(1, sizeof(TWLine));
339 OIC_LOG(ERROR, TAG, "Ran out of memory.");
340 return TW_RESULT_ERROR_NO_MEMORY;
342 size_t lineLength = strlen(line);
344 twLine->length = lineLength;
345 size_t errorLength = strlen(TW_ENDCONTROL_ERROR_STRING);
346 size_t maxLength = (lineLength > errorLength) ? errorLength : lineLength;
348 if((errorLength == lineLength) &&
349 strncmp(line, TW_ENDCONTROL_ERROR_STRING, maxLength) == 0)
351 entry->atErrorCode[0] = line[errorLength];
352 entry->atErrorCode[1] = line[errorLength + 1];
356 entry->atErrorCode[0] = '0';
357 entry->atErrorCode[1] = '0';
360 // Null terminate the string.
361 entry->atErrorCode[2] = '\0';
365 entry->lines = twLine;
369 entry->lines[entry->count] = *twLine;
376 TWEntry * readEntry(int fd)
378 // Calls readBufferLine().
379 // Forms TWEntry from 1-n lines based on the response type.
381 TWEntry * entry = (TWEntry *) OICCalloc(1, sizeof(TWEntry));
384 OIC_LOG(ERROR, TAG, "Ran out of memory.");
389 const char * bufferLine = NULL;
390 TWEntryTypePair entryTypePair = { .resultTxt = NULL,
392 .entryType = TW_NONE };
394 while(numLines == 0 || bufferLine)
398 bufferLine = readBufferLine(fd);
403 entryTypePair = getEntryTypePair(bufferLine);
404 if(entryTypePair.entryType == TW_NONE)
411 bufferLine = readBufferLine(fd);
413 if(bufferLine != NULL)
415 entry->type = entryTypePair.entryType;
416 TWAddLineToEntry(bufferLine, entry);
418 entry->count = numLines;
421 if(entryTypePair.numLines != numLines)
423 entry->resultCode = TW_RESULT_ERROR_LINE_COUNT;
427 entry->resultCode = TW_RESULT_OK;
436 TWResultCode TWRetrieveEUI(PIPlugin_Zigbee * plugin, TWSock * twSock)
438 if(!plugin || !twSock)
440 OIC_LOG(ERROR, TAG, "Invalid param.");
441 return TW_RESULT_ERROR_INVALID_PARAMS;
443 if(twSock->isActive == false)
445 OIC_LOG(ERROR, TAG, "Tried to retrieve Zigbee EUI on an uninitialized socket.");
446 return TW_RESULT_ERROR;
450 char hideBuffer[1] = "";
455 p = read(twSock->fd, hideBuffer, 1);
458 OIC_LOG_V(ERROR, TAG, "\tCould not read from port. Errno is: %d\n", errno);
459 return TW_RESULT_ERROR;
463 TWEntry * entry = NULL;
464 TWResultCode deleteResult = TW_RESULT_OK;
465 TWResultCode result = TWIssueATCommand(plugin, AT_CMD_GET_LOCAL_EUI);
466 if(result != TW_RESULT_OK)
470 result = TWGrabMutex(&twSock->mutex);
471 if(result != TW_RESULT_OK)
475 entry = readEntry(twSock->fd);
478 result = TWReleaseMutex(&twSock->mutex);
479 if(result != TW_RESULT_OK)
484 twSock->eui = (char *) OICMalloc(strlen(entry->lines[0].line)+1);
487 result = TWReleaseMutex(&twSock->mutex);
488 if(result != TW_RESULT_OK)
492 result = TW_RESULT_ERROR_NO_MEMORY;
496 if(SIZE_EUI != (strlen(entry->lines[0].line)+1))
498 result = TWReleaseMutex(&twSock->mutex);
499 if(result != TW_RESULT_OK)
503 OICFree(twSock->eui);
504 result = TW_RESULT_ERROR;
508 OICStrcpy(twSock->eui, SIZE_EUI, entry->lines[0].line);
510 result = TWReleaseMutex(&twSock->mutex);
511 if(result != TW_RESULT_OK)
513 OICFree(twSock->eui);
517 deleteResult = TWDeleteEntry(plugin, entry);
518 if(deleteResult != TW_RESULT_OK)
525 TWResultCode TWStartSock(PIPlugin_Zigbee * plugin, const char * fileLoc)
527 TWSock * sock = TWGetSock((PIPlugin_Zigbee *)plugin);
528 if(sock && sock->isActive == true)
530 // Ignore requests to start up the same socket.
535 sock = (TWSock *) OICCalloc(1, sizeof(TWSock));
538 return TW_RESULT_ERROR_NO_MEMORY;
541 TWResultCode ret = TWAddTWSock(sock, plugin, fileLoc);
547 ret = TWRetrieveEUI((PIPlugin_Zigbee *)plugin, sock);
548 if(ret != TW_RESULT_OK)
550 OIC_LOG(ERROR, TAG, "Unable to retrieve Zigbee Radio's EUI.");
554 int result = pthread_create(&(sock->threadHandle),
560 OIC_LOG_V(ERROR, TAG, "Thread start failed with error %d", result);
561 ret = TW_RESULT_ERROR;
568 TWDeleteTWSock(sock);
572 static void sigHandler(int signal)
574 pthread_t tid = pthread_self();
577 OIC_LOG_V(INFO, TAG, "Received signal on thread: %lu\n", tid);
578 OIC_LOG_V(INFO, TAG, "Signal is: %d", signal);
581 void * readForever(/*PIPlugin*/ void * plugin)
583 TWResultCode result = TW_RESULT_OK;
584 TWEntry * entry = NULL;
585 TWSock * twSock = TWGetSock((PIPlugin_Zigbee *)plugin);
588 OIC_LOG(ERROR, TAG, "Unable to retrieve associated socket.");
592 pthread_t tid = pthread_self();
594 OIC_LOG_V(INFO, TAG, "ReadForever Telegesis ThreadId: %lu", tid);
596 struct sigaction action = { .sa_handler = 0 };
600 action.sa_handler = sigHandler;
602 sigemptyset(&action.sa_mask);
603 sigaction(EINTR, &action, NULL);
607 FD_SET(twSock->fd, &readFDS);
608 bool haveMutex = false;
612 // 'sigmask' is needed to catch intterupts.
613 // This interrupt happens after call to pthread_exit(..., EINTR).
614 // Once a signal handler is registered, pselect will handle interrupts by returning
615 // with '-1' and setting errno appropriately.
616 int ret = pselect(twSock->fd+1, &readFDS, NULL, NULL, NULL, &sigmask);
624 // This EINTR signal is not for us. Do not handle it.
626 // Notify other threads waiting for a response that the stack is going down.
627 pthread_cond_signal(&twSock->queueCV);
628 OIC_LOG(DEBUG, TAG, "Thread has been joined. Exiting thread.");
629 pthread_exit(PTHREAD_CANCELED);
634 OIC_LOG_V(ERROR, TAG, "Unaccounted error occurred. Exiting thread."
635 "Errno is: %d", errno);
641 ret = FD_ISSET(twSock->fd, &readFDS);
644 // Valid data on valid socket.
645 // Grab & parse, then pass up to upper layers.
646 if(haveMutex != true)
648 result = TWGrabMutex(&twSock->mutex);
649 if(result != TW_RESULT_OK)
651 OIC_LOG(ERROR, TAG, "Unable to grab mutex.");
656 entry = readEntry(twSock->fd);
659 result = TWReleaseMutex(&twSock->mutex);
660 if(result != TW_RESULT_OK)
662 OIC_LOG(ERROR, TAG, "Unable to release mutex.");
666 // This is most likely a parsing error of the received
667 // response. Not necessarily fatal.
670 result = TWEnqueueEntry((PIPlugin_Zigbee *)plugin, entry);
671 if(result != TW_RESULT_OK)
673 OIC_LOG_V(ERROR, TAG, "Could not add TWEntry to queue for"
674 "consumption by the application"
675 "layer. Error is: %d", result);
676 TWDeleteEntry((PIPlugin_Zigbee *)plugin, entry);
677 // This is most likely a FATAL error, such as out of memory.
681 // Notify other threads waiting for a response that an entry has been enqueued.
682 pthread_cond_signal(&twSock->queueCV);
684 result = TWReleaseMutex(&twSock->mutex);
689 // Unrelated data waiting elsewhere. Continue the loop.
698 TWResultCode TWIssueATCommand(PIPlugin_Zigbee * plugin, const char * command)
700 if(!plugin || !command)
702 return TW_RESULT_ERROR_INVALID_PARAMS;
704 OIC_LOG_V(INFO, TAG, "Attempt to write %s.", command);
705 TWSock * twSock = TWGetSock(plugin);
708 return TW_RESULT_ERROR;
711 if(twSock->isActive == false)
713 return TW_RESULT_ERROR;
716 TWResultCode result = TW_RESULT_OK;
717 TWResultCode mutexResult = TW_RESULT_OK;
718 mutexResult = TWGrabMutex(&twSock->mutex);
719 if(mutexResult != TW_RESULT_OK)
723 size_t sendCommandSize = (strlen(command) + strlen("\r") + 1) * sizeof(char);
724 char * sendCommand = (char *) OICCalloc(1, sendCommandSize);
727 return TW_RESULT_ERROR_NO_MEMORY;
729 char * temp = OICStrcpy(sendCommand, sendCommandSize, command);
730 if(temp != sendCommand)
732 result = TW_RESULT_ERROR;
735 temp = OICStrcat(sendCommand, sendCommandSize, "\r");
736 if(temp != sendCommand)
738 result = TW_RESULT_ERROR;
741 size_t expectedWrittenBytes = strlen(sendCommand);
743 size_t actualWrittenBytes = write(twSock->fd, sendCommand, expectedWrittenBytes);
744 if(actualWrittenBytes <= 0)
746 OIC_LOG_V(ERROR, TAG, "Could not write to port. Errno is: %d\n", errno);
747 result = TW_RESULT_ERROR;
750 if(actualWrittenBytes != expectedWrittenBytes)
752 OIC_LOG(ERROR, TAG, "Telegesis Adapter did not receive expected command. Unknown state.");
753 result = TW_RESULT_ERROR;
757 mutexResult = TWReleaseMutex(&twSock->mutex);
758 if(mutexResult != TW_RESULT_OK)
762 OICFree(sendCommand);
766 TWResultCode TWEnqueueEntry(PIPlugin_Zigbee * plugin, TWEntry * entry)
768 if(!plugin || !entry)
770 return TW_RESULT_ERROR_INVALID_PARAMS;
772 TWSock * twSock = TWGetSock(plugin);
775 return TW_RESULT_ERROR;
778 if(twSock->isActive == false)
780 return TW_RESULT_ERROR;
782 LL_APPEND(twSock->queue, entry);
786 TWResultCode TWDequeueEntry(PIPlugin_Zigbee * plugin, TWEntry ** entry, TWEntryType type)
788 if(!plugin || !entry)
790 return TW_RESULT_ERROR_INVALID_PARAMS;
792 TWSock * twSock = TWGetSock(plugin);
795 return TW_RESULT_ERROR;
798 if(twSock->isActive == false)
800 return TW_RESULT_ERROR;
803 TWResultCode ret = TW_RESULT_OK;
805 // If no entry is found, then this code path returns immediately.
806 ret = TWGrabMutex(&twSock->mutex);
807 if(ret != TW_RESULT_OK)
814 struct timespec abs_time;
815 clock_gettime(CLOCK_REALTIME , &abs_time);
816 abs_time.tv_sec += TIME_OUT_10_SECONDS;
819 // Wait for up to 10 seconds for the entry to put into the queue.
820 ret = TWWait(&twSock->queueCV, &twSock->mutex, TIME_OUT_10_SECONDS);
821 if(ret != TW_RESULT_OK)
825 if(twSock->isActive == false)
829 TWEntry * out = NULL;
830 TWEntry * temp = NULL;
831 LL_FOREACH_SAFE(twSock->queue, out, temp)
833 if(out->type == type)
839 struct timespec cur_time;
840 clock_gettime(CLOCK_REALTIME, &cur_time);
841 if(cur_time.tv_sec >= abs_time.tv_sec)
849 *entry = twSock->queue;
853 LL_DELETE(twSock->queue, *entry);
855 return TWReleaseMutex(&twSock->mutex);
858 TWResultCode TWFreeQueue(PIPlugin_Zigbee * plugin)
862 return TW_RESULT_ERROR_INVALID_PARAMS;
864 TWResultCode ret = TW_RESULT_OK;
865 TWEntry * entry = NULL;
868 ret = TWDequeueEntry(plugin, &entry, TW_NONE);
869 if(ret != TW_RESULT_OK)
877 ret = TWDeleteEntry(plugin, entry);
878 if(ret != TW_RESULT_OK)
886 TWResultCode TWDeleteEntry(PIPlugin_Zigbee * plugin, TWEntry * entry)
888 if(!plugin || !entry)
890 return TW_RESULT_ERROR_INVALID_PARAMS;
893 TWSock * twSock = TWGetSock(plugin);
896 return TW_RESULT_ERROR;
899 if(twSock->isActive == false)
901 return TW_RESULT_ERROR;
904 TWResultCode ret = TWGrabMutex(&twSock->mutex);
905 if(ret != TW_RESULT_OK)
909 TWEntry * out = NULL;
910 TWEntry * tmp = NULL;
911 LL_FOREACH_SAFE(twSock->queue, out, tmp)
915 OIC_LOG(ERROR, TAG, "Tried to delete an entry that is still in the queue. \
916 Please dequeue the entry first.");
917 return TW_RESULT_ERROR;
920 ret = TWReleaseMutex(&twSock->mutex);
927 TWResultCode TWGetEUI(PIPlugin_Zigbee * plugin, char ** eui)
931 return TW_RESULT_ERROR_INVALID_PARAMS;
933 TWSock * twSock = TWGetSock(plugin);
936 return TW_RESULT_ERROR;
939 if(twSock->isActive == false)
941 return TW_RESULT_ERROR;
944 *eui = OICStrdup(twSock->eui);
950 TWResultCode TWStopSock(PIPlugin_Zigbee * plugin)
954 return TW_RESULT_ERROR_INVALID_PARAMS;
956 TWSock * twSock = TWGetSock(plugin);
959 return TW_RESULT_ERROR;
962 if(twSock->isActive == false)
964 return TW_RESULT_ERROR;
967 TWResultCode ret = TWFreeQueue(plugin);
968 if(ret != TW_RESULT_OK)
973 twSock->isActive = false;
975 void * retVal = NULL;
976 int pthreadRet = pthread_kill(twSock->threadHandle, EINTR);
979 return TW_RESULT_ERROR;
982 pthreadRet = pthread_join(twSock->threadHandle, &retVal);
988 OIC_LOG(ERROR, TAG, "A deadlock has occurred.");
991 OIC_LOG(ERROR, TAG, "Thread is not joinable or another thread has already joined.");
994 OIC_LOG(ERROR, TAG, "No thread with the ID could be found.");
997 OIC_LOG_V(ERROR, TAG, "Unknown error occurred when joining thread: %d", pthreadRet);
999 return TW_RESULT_ERROR;
1001 if(retVal != PTHREAD_CANCELED)
1003 return TW_RESULT_ERROR;
1005 ret = TWDeleteTWSock(twSock);
1006 if(ret != TW_RESULT_OK)