source/MQTTOperation.c
/**
 * @brief publish sensor data, get sensor data, or
 *        yield mqtt client to check subscriptions
 *
 * @param[in] pvParameters UNUSED/PASSED THROUGH
 *
 * @return NONE
 */
@brief publish sensor data, get sensor data, or
yield mqtt client to check subscriptions

@param[in] pvParameters UNUSED/PASSED THROUGH

@return NONE
@brief publish sensor data get sensor data or yield mqtt client to check subscriptions @param [ in ] pvParameters UNUSED / PASSED THROUGH @return NONE
static void MQTTOperation_ClientPublish(void) {
	LOG_AT_INFO(("MQTTOperation: Start publishing ...\r\n"));
	semaphoreAssetBuffer = xSemaphoreCreateBinary();
	xSemaphoreGive(semaphoreAssetBuffer);
	semaphoreSensorBuffer = xSemaphoreCreateBinary();
	xSemaphoreGive(semaphoreSensorBuffer);
	commandQueue = xQueueCreate(4, SIZE_XSMALL_BUF);
	Retcode_T retcode = RETCODE_OK;
	memset(sensorStreamBuffer.data, 0x00, sizeof(sensorStreamBuffer.data));
	sensorStreamBuffer.length = NUMBER_UINT32_ZERO;
	memset(assetStreamBuffer.data, 0x00, sizeof(assetStreamBuffer.data));
	assetStreamBuffer.length = NUMBER_UINT32_ZERO;
	timerHandleAsset = xTimerCreate((const char * const ) "Asset Update Timer", 
			MILLISECONDS(1000), 
			pdTRUE, 
			NULL, 
			MQTTOperation_AssetUpdate 
			);
	xTimerStart(timerHandleAsset, UINT32_C(0xffff));
	timerHandleSensor = xTimerCreate(
			(const char * const ) "Sensor Update Timer", 
			pdMS_TO_TICKS(tickRateMS), 
			pdTRUE, 
			NULL, 
			MQTTOperation_SensorUpdate 
			);
	xTimerStart(timerHandleSensor, UINT32_C(0xffff));
	char readbuffer[strlen(BOOT_PENDING)+1]; 
	MQTTStorage_Flash_ReadBootStatus((uint8_t *) readbuffer);
	LOG_AT_DEBUG(("MQTTOperation: Reading boot status: [%s]\r\n", readbuffer));
	if ((strncmp(readbuffer, BOOT_PENDING, strlen(BOOT_PENDING)) == 0)) {
		if (xQueueSend(commandQueue,(char *) "511,DUMMY,restartConfirm", 0) != pdTRUE) {
			LOG_AT_ERROR(("MQTTOperation: Could not buffer command!\r\n"));
		}
	}
	uint32_t measurementCounter = 0;
	char commandBuffer[SIZE_XSMALL_BUF] = { 0 };
	BaseType_t semaphoreResult;
	while (1) {
		retcode = MQTTOperation_ValidateWLANConnectivity();
		if (assetStreamBuffer.length > NUMBER_UINT32_ZERO) {
			if (RETCODE_OK == retcode) {
				semaphoreResult = xSemaphoreTake(
						semaphoreAssetBuffer, pdMS_TO_TICKS(SEMAPHORE_TIMEOUT));
				if (pdPASS == semaphoreResult) {
					if (logging_enabled) {
						LOG_AT_DEBUG(
								("MQTTOperation: Publishing asset data: length [%ld], content:\r\n%s", assetStreamBuffer.length, assetStreamBuffer.data));
					}
					MqttPublishAssetInfo.Payload = assetStreamBuffer.data;
					MqttPublishAssetInfo.PayloadLength =
							assetStreamBuffer.length;
					retcode = MQTT_PublishToTopic_Z(&MqttPublishAssetInfo,
					MQTT_PUBLISH_TIMEOUT_IN_MS);
					memset(assetStreamBuffer.data, 0x00,
							assetStreamBuffer.length);
					assetStreamBuffer.length = NUMBER_UINT32_ZERO;
				}
				xSemaphoreGive(semaphoreAssetBuffer);
				if (RETCODE_OK != retcode) {
					LOG_AT_ERROR(("MQTTOperation: MQTT publish failed \r\n"));
					Retcode_RaiseError(retcode);
					errorCountPublish++;
				}
				if (assetUpdateProcess == APP_ASSET_PUBLISHED
						&& RETCODE_OK == retcode) {
					vTaskDelay(pdMS_TO_TICKS(1000));
					retcode = MQTTOperation_SubscribeTopics();
					if (RETCODE_OK != retcode) {
						LOG_AT_ERROR(
								("MQTTOperation: MQTT subscription failed!\r\n"));
						retcode = MQTTOperation_ValidateWLANConnectivity();
					} else {
						assetUpdateProcess = APP_ASSET_COMPLETED;
					}
				}
			}
		}
		if (sensorStreamBuffer.length > NUMBER_UINT32_ZERO) {
			AppController_SetAppStatus(APP_STATUS_OPERATING_STARTED);
			if (RETCODE_OK == retcode) {
				   semaphoreResult = xSemaphoreTake(
						semaphoreSensorBuffer,
						pdMS_TO_TICKS(SEMAPHORE_TIMEOUT));
				if (pdPASS == semaphoreResult) {
					measurementCounter++;
					if (logging_enabled) {
						LOG_AT_DEBUG(
								("MQTTOperation: Publishing sensor data: length [%ld], message [%lu], content:\r\n%s", sensorStreamBuffer.length, measurementCounter, sensorStreamBuffer.data));
					}
					MqttPublishDataInfo.Payload = sensorStreamBuffer.data;
					MqttPublishDataInfo.PayloadLength =
							sensorStreamBuffer.length;
					retcode = MQTT_PublishToTopic_Z(&MqttPublishDataInfo,
							MQTT_PUBLISH_TIMEOUT_IN_MS);
					memset(sensorStreamBuffer.data, 0x00,
							sensorStreamBuffer.length);
					sensorStreamBuffer.length = NUMBER_UINT32_ZERO;
					if (RETCODE_OK != retcode) {
						LOG_AT_ERROR(
								("MQTTOperation: MQTT publish failed trying to ignore\r\n"));
						retcode = RETCODE_OK;
					}
				}
				xSemaphoreGive(semaphoreSensorBuffer);
				if (RETCODE_OK != retcode) {
					LOG_AT_ERROR(("MQTTOperation: MQTT publish failed \r\n"));
					Retcode_RaiseError(retcode);
					errorCountPublish++;
				}
			} else {
				memset(sensorStreamBuffer.data, 0x00,
						sensorStreamBuffer.length);
				sensorStreamBuffer.length = NUMBER_UINT32_ZERO;
			}
		}
		if (uxQueueMessagesWaiting(commandQueue) > 0
				&& commandProgress == DEVICE_OPERATION_WAITING) {
			memset(commandBuffer, 0x00, SIZE_XSMALL_BUF);
			if (xQueueReceive(commandQueue, &commandBuffer, 0) != pdTRUE)
				LOG_AT_ERROR(
						("MQTTOperation: Could read command from buffer!\r\n"));
			else {
				commandProgress = DEVICE_OPERATION_BEFORE_EXECUTING;
				LOG_AT_DEBUG(
						("MQTTOperation: Execute command from buffer: [%s]!\r\n", commandBuffer));
				MQTTOperation_ExecuteCommand(commandBuffer);
			}
		}
		vTaskDelay(pdMS_TO_TICKS(MINIMAL_SPEED));
	}
}
static void MQTTOperation_ClientPublish ( void ) { LOG_AT_INFO ( ( " \r \n " ) ) ; semaphoreAssetBuffer = xSemaphoreCreateBinary ( ) ; xSemaphoreGive ( semaphoreAssetBuffer ) ; semaphoreSensorBuffer = xSemaphoreCreateBinary ( ) ; xSemaphoreGive ( semaphoreSensorBuffer ) ; commandQueue = xQueueCreate ( 4 , SIZE_XSMALL_BUF ) ; Retcode_T retcode = RETCODE_OK ; memset ( sensorStreamBuffer . data , 0x00 , sizeof ( sensorStreamBuffer . data ) ) ; sensorStreamBuffer . length = NUMBER_UINT32_ZERO ; memset ( assetStreamBuffer . data , 0x00 , sizeof ( assetStreamBuffer . data ) ) ; assetStreamBuffer . length = NUMBER_UINT32_ZERO ; timerHandleAsset = xTimerCreate ( ( const char * const ) " " , MILLISECONDS ( 1000 ) , pdTRUE , NULL , MQTTOperation_AssetUpdate ) ; xTimerStart ( timerHandleAsset , UINT32_C ( 0xffff ) ) ; timerHandleSensor = xTimerCreate ( ( const char * const ) " " , pdMS_TO_TICKS ( tickRateMS ) , pdTRUE , NULL , MQTTOperation_SensorUpdate ) ; xTimerStart ( timerHandleSensor , UINT32_C ( 0xffff ) ) ; char readbuffer [ strlen ( BOOT_PENDING ) + 1 ] ; MQTTStorage_Flash_ReadBootStatus ( ( uint8_t * ) readbuffer ) ; LOG_AT_DEBUG ( ( " \r \n " , readbuffer ) ) ; if ( ( strncmp ( readbuffer , BOOT_PENDING , strlen ( BOOT_PENDING ) ) == 0 ) ) { if ( xQueueSend ( commandQueue , ( char * ) " " , 0 ) != pdTRUE ) { LOG_AT_ERROR ( ( " \r \n " ) ) ; } } uint32_t measurementCounter = 0 ; char commandBuffer [ SIZE_XSMALL_BUF ] = { 0 } ; BaseType_t semaphoreResult ; while ( 1 ) { retcode = MQTTOperation_ValidateWLANConnectivity ( ) ; if ( assetStreamBuffer . length > NUMBER_UINT32_ZERO ) { if ( RETCODE_OK == retcode ) { semaphoreResult = xSemaphoreTake ( semaphoreAssetBuffer , pdMS_TO_TICKS ( SEMAPHORE_TIMEOUT ) ) ; if ( pdPASS == semaphoreResult ) { if ( logging_enabled ) { LOG_AT_DEBUG ( ( " \r \n " , assetStreamBuffer . length , assetStreamBuffer . data ) ) ; } MqttPublishAssetInfo . Payload = assetStreamBuffer . data ; MqttPublishAssetInfo . PayloadLength = assetStreamBuffer . length ; retcode = MQTT_PublishToTopic_Z ( & MqttPublishAssetInfo , MQTT_PUBLISH_TIMEOUT_IN_MS ) ; memset ( assetStreamBuffer . data , 0x00 , assetStreamBuffer . length ) ; assetStreamBuffer . length = NUMBER_UINT32_ZERO ; } xSemaphoreGive ( semaphoreAssetBuffer ) ; if ( RETCODE_OK != retcode ) { LOG_AT_ERROR ( ( " \r \n " ) ) ; Retcode_RaiseError ( retcode ) ; errorCountPublish ++ ; } if ( assetUpdateProcess == APP_ASSET_PUBLISHED && RETCODE_OK == retcode ) { vTaskDelay ( pdMS_TO_TICKS ( 1000 ) ) ; retcode = MQTTOperation_SubscribeTopics ( ) ; if ( RETCODE_OK != retcode ) { LOG_AT_ERROR ( ( " \r \n " ) ) ; retcode = MQTTOperation_ValidateWLANConnectivity ( ) ; } else { assetUpdateProcess = APP_ASSET_COMPLETED ; } } } } if ( sensorStreamBuffer . length > NUMBER_UINT32_ZERO ) { AppController_SetAppStatus ( APP_STATUS_OPERATING_STARTED ) ; if ( RETCODE_OK == retcode ) { semaphoreResult = xSemaphoreTake ( semaphoreSensorBuffer , pdMS_TO_TICKS ( SEMAPHORE_TIMEOUT ) ) ; if ( pdPASS == semaphoreResult ) { measurementCounter ++ ; if ( logging_enabled ) { LOG_AT_DEBUG ( ( " \r \n " , sensorStreamBuffer . length , measurementCounter , sensorStreamBuffer . data ) ) ; } MqttPublishDataInfo . Payload = sensorStreamBuffer . data ; MqttPublishDataInfo . PayloadLength = sensorStreamBuffer . length ; retcode = MQTT_PublishToTopic_Z ( & MqttPublishDataInfo , MQTT_PUBLISH_TIMEOUT_IN_MS ) ; memset ( sensorStreamBuffer . data , 0x00 , sensorStreamBuffer . length ) ; sensorStreamBuffer . length = NUMBER_UINT32_ZERO ; if ( RETCODE_OK != retcode ) { LOG_AT_ERROR ( ( " \r \n " ) ) ; retcode = RETCODE_OK ; } } xSemaphoreGive ( semaphoreSensorBuffer ) ; if ( RETCODE_OK != retcode ) { LOG_AT_ERROR ( ( " \r \n " ) ) ; Retcode_RaiseError ( retcode ) ; errorCountPublish ++ ; } } else { memset ( sensorStreamBuffer . data , 0x00 , sensorStreamBuffer . length ) ; sensorStreamBuffer . length = NUMBER_UINT32_ZERO ; } } if ( uxQueueMessagesWaiting ( commandQueue ) > 0 && commandProgress == DEVICE_OPERATION_WAITING ) { memset ( commandBuffer , 0x00 , SIZE_XSMALL_BUF ) ; if ( xQueueReceive ( commandQueue , & commandBuffer , 0 ) != pdTRUE ) LOG_AT_ERROR ( ( " \r \n " ) ) ; else { commandProgress = DEVICE_OPERATION_BEFORE_EXECUTING ; LOG_AT_DEBUG ( ( " \r \n " , commandBuffer ) ) ; MQTTOperation_ExecuteCommand ( commandBuffer ) ; } } vTaskDelay ( pdMS_TO_TICKS ( MINIMAL_SPEED ) ) ; } }
@brief publish sensor data, get sensor data, or
        yield mqtt client to check subscriptions
@brief publish sensor data get sensor data or yield mqtt client to check subscriptions
// initialize buffers // used only for debugging purposes // timer period //Autoreload pdTRUE or pdFALSE - should the timer start again after it expired? // optional identifier // static callback function // used only for debugging purposes // timer period //Autoreload pdTRUE or pdFALSE - should the timer start again after it expired? // optional identifier // static callback function // ckeck if reboot process is pending to be confirmed /* Temporary buffer for write file */ /* A function that implements a task must not exit or attempt to return to
	 its caller function as there is nothing to return to. */ /* Check whether the WLAN network connection is available */ // only log measurements when loggin is enabled // wait an extra tick rate until topic are created in Cumulocity // topics are only created after the device is created // ignore previous measurements in order to prevent buffer overrun // test if some commands are pending