Skip to content

Commit

Permalink
Prep for MQTT binary file transfer
Browse files Browse the repository at this point in the history
  • Loading branch information
@arendst @yvesdm3000
arendst authored and yvesdm3000 committed May 16, 2021
1 parent 44577c7 commit 75c783d
Show file tree
Hide file tree
Showing 3 changed files with 325 additions and 273 deletions.
31 changes: 19 additions & 12 deletions tasmota/support_command.ino
View file
Original file line number Diff line number Diff line change
Expand Up @@ -192,11 +192,6 @@ void CommandHandler(char* topicBuf, char* dataBuf, uint32_t data_len)
ShowFreeMem(PSTR("CommandHandler"));
#endif

while (*dataBuf && isspace(*dataBuf)) {
dataBuf++; // Skip leading spaces in data
data_len--;
}

bool grpflg = false;
uint32_t real_index = SET_MQTT_GRP_TOPIC;
for (uint32_t i = 0; i < MAX_GROUP_TOPICS; i++) {
Expand Down Expand Up @@ -237,20 +232,32 @@ void CommandHandler(char* topicBuf, char* dataBuf, uint32_t data_len)
type[i] = '\0';
}

AddLog_P(LOG_LEVEL_DEBUG, PSTR("CMD: " D_GROUP " %d, " D_INDEX " %d, " D_COMMAND " \"%s\", " D_DATA " \"%s\""), grpflg, index, type, dataBuf);
bool binary_data = (index > 199); // Suppose binary data on topic index > 199
if (!binary_data) {
while (*dataBuf && isspace(*dataBuf)) {
dataBuf++; // Skip leading spaces in data
data_len--;
}
}

AddLog_P(LOG_LEVEL_DEBUG, PSTR("CMD: " D_GROUP " %d, " D_INDEX " %d, " D_COMMAND " \"%s\", " D_DATA " \"%s\""),
grpflg, index, type, (binary_data) ? PSTR("Binary") : dataBuf);

if (type != nullptr) {
Response_P(PSTR("{\"" D_JSON_COMMAND "\":\"" D_JSON_ERROR "\"}"));

if (Settings.ledstate &0x02) { TasmotaGlobal.blinks++; }

if (!strcmp(dataBuf,"?")) { data_len = 0; }
int32_t payload = -99;
if (!binary_data) {
if (!strcmp(dataBuf,"?")) { data_len = 0; }

char *p;
int32_t payload = strtol(dataBuf, &p, 0); // decimal, octal (0) or hex (0x)
if (p == dataBuf) { payload = -99; }
int temp_payload = GetStateNumber(dataBuf);
if (temp_payload > -1) { payload = temp_payload; }
char *p;
payload = strtol(dataBuf, &p, 0); // decimal, octal (0) or hex (0x)
if (p == dataBuf) { payload = -99; }
int temp_payload = GetStateNumber(dataBuf);
if (temp_payload > -1) { payload = temp_payload; }
}

DEBUG_CORE_LOG(PSTR("CMD: Payload %d"), payload);

Expand Down
290 changes: 290 additions & 0 deletions tasmota/xdrv_02_mqtt_1_file.ino
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,290 @@
/*
xdrv_02_mqtt_1_file.ino - mqtt file support for Tasmota
Copyright (C) 2021 Theo Arends
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/

#define USE_MQTT_FILE

#ifdef USE_MQTT_FILE
/*********************************************************************************************\
* MQTT file transfer
*
* Supports base64 encoded binary data transfer
\*********************************************************************************************/

#include <base64.hpp>

struct FMQTT {
uint32_t file_pos = 0; // MQTT file position during upload/download
uint32_t file_size = 0; // MQTT total file size
uint32_t file_type = 0; // MQTT File type (See UploadTypes)
uint8_t* file_buffer = nullptr; // MQTT file buffer
MD5Builder md5; // MQTT md5
String file_md5; // MQTT received file md5 (32 chars)
uint16_t topic_size; // MQTT topic length with terminating <null>
uint8_t file_id = 0; // MQTT unique file id during upload/download
} FMqtt;

void MqttTopicSize(uint32_t topic_size) {
FMqtt.topic_size = topic_size +1;
}

/*
The download chunk size is the data size before it is encoded to base64.
It is smaller than the upload chunksize as it is bound by MESSZ
The download buffer with length MESSZ (1042) contains
- Payload ({"Id":117,"Data":"<base64 encoded mqtt_file_chuck_size>"}<null>)
*/
const uint32_t FileTransferHeaderSize = 21; // {"Id":116,"Data":""}<null>
const uint32_t mqtt_file_chuck_size = (((MESSZ - FileTransferHeaderSize) / 4) * 3) -2;

uint32_t FileUploadChunckSize(void) {
/*
The upload chunk size is the data size before it is encoded to base64.
It can be larger than the download chunksize which is bound by MESSZ
The PubSubClient upload buffer with length MQTT_MAX_PACKET_SIZE (1200) contains
- Header of 5 bytes (MQTT_MAX_HEADER_SIZE)
- Topic string terminated with a zero (stat/demo/FILEUPLOAD<null>)
- Payload ({"Id":116,"Data":"<base64 encoded FileUploadChunckSize>"}<null>)
*/
const uint32_t PubSubClientHeaderSize = 5; // MQTT_MAX_HEADER_SIZE

return (((MQTT_MAX_PACKET_SIZE - PubSubClientHeaderSize - FMqtt.topic_size - FileTransferHeaderSize) / 4) * 3) -2;
}

void CmndFileUpload(void) {
/*
Upload (binary) max 700 bytes chunks of data base64 encoded with MD5 hash over base64 decoded data
FileUpload 0 - Abort current upload
FileUpload {"File":"Config_wemos10_9.4.0.3.dmp","Id":116,"Type":2,"Size":4096}
FileUpload {"Id":116,"Data":"CRJcTQ9fYGF ... OT1BRUlNUVVZXWFk="}
FileUpload {"Id":116,"Data":" ... "}
FileUpload {"Id":116,"Md5":"496fcbb433bbca89833063174d2c5747"}
*/
const char* base64_data = nullptr;
uint32_t rcv_id = 0;

char* dataBuf = (char*)XdrvMailbox.data;
if (strlen(dataBuf) > 8) { // Workaround exception if empty JSON like {} - Needs checks
JsonParser parser((char*) dataBuf);
JsonParserObject root = parser.getRootObject();
if (root) {
JsonParserToken val = root[PSTR("ID")];
if (val) { rcv_id = val.getUInt(); }
val = root[PSTR("TYPE")];
if (val) { FMqtt.file_type = val.getUInt(); }
val = root[PSTR("SIZE")];
if (val) { FMqtt.file_size = val.getUInt(); }
val = root[PSTR("MD5")];
if (val) { FMqtt.file_md5 = val.getStr(); }
val = root[PSTR("DATA")];
if (val) { base64_data = val.getStr(); }
}
}

if ((0 == FMqtt.file_id) && (rcv_id > 0) && (FMqtt.file_size > 0) && (FMqtt.file_type > 0)) {
// Init upload buffer
FMqtt.file_buffer = nullptr;

if (UPL_TASMOTA == FMqtt.file_type) {
if (Update.begin(FMqtt.file_size)) {
FMqtt.file_buffer = &FMqtt.file_id; // Dummy buffer
}
}
else if (UPL_SETTINGS == FMqtt.file_type) {
if (SettingsConfigBackup()) {
FMqtt.file_buffer = settings_buffer;
}
}

if (!FMqtt.file_buffer) {
ResponseCmndChar(PSTR(D_JSON_INVALID_FILE_TYPE));
} else {
FMqtt.file_id = rcv_id;
FMqtt.file_pos = 0;

FMqtt.md5 = MD5Builder();
FMqtt.md5.begin();

ResponseCmndChar(PSTR(D_JSON_STARTED));
MqttPublishPrefixTopic_P(STAT, XdrvMailbox.command); // Enforce stat/wemos10/FILEUPLOAD
}
}
else if ((FMqtt.file_id > 0) && (FMqtt.file_id != rcv_id)) {
// Error receiving data

if (UPL_TASMOTA == FMqtt.file_type) {
Update.end(true);
}
else if (UPL_SETTINGS == FMqtt.file_type) {
SettingsBufferFree();
}

FMqtt.file_buffer = nullptr;
ResponseCmndChar(PSTR(D_JSON_ABORTED));
}

if (FMqtt.file_buffer) {
if ((FMqtt.file_pos < FMqtt.file_size) && base64_data) {
// Save upload into buffer - Handle possible buffer overflows
uint32_t rcvd_bytes = decode_base64_length((unsigned char*)base64_data);
unsigned char decode_output[rcvd_bytes];
decode_base64((unsigned char*)base64_data, (unsigned char*)decode_output);

uint32_t bytes_left = FMqtt.file_size - FMqtt.file_pos;
uint32_t read_bytes = (bytes_left < rcvd_bytes) ? bytes_left : rcvd_bytes;
FMqtt.md5.add(decode_output, read_bytes);

if (UPL_TASMOTA == FMqtt.file_type) {
Update.write(decode_output, read_bytes);
} else {
uint8_t* buffer = FMqtt.file_buffer + FMqtt.file_pos;
memcpy(buffer, decode_output, read_bytes);
}

FMqtt.file_pos += read_bytes;

if ((FMqtt.file_pos > rcvd_bytes) && ((FMqtt.file_pos % 102400) <= rcvd_bytes)) {
TasmotaGlobal.masterlog_level = LOG_LEVEL_NONE; // Enable logging
AddLog(LOG_LEVEL_DEBUG, PSTR(D_LOG_UPLOAD "Progress %d kB"), FMqtt.file_pos / 1024);
TasmotaGlobal.masterlog_level = LOG_LEVEL_DEBUG_MORE; // Hide upload data logging
}
}

if ((FMqtt.file_pos < FMqtt.file_size) || (FMqtt.file_md5.length() != 32)) {
TasmotaGlobal.masterlog_level = LOG_LEVEL_DEBUG_MORE; // Hide upload data logging

// {"Id":116,"MaxSize":"765"}
Response_P(PSTR("{\"Id\":%d,\"MaxSize\":%d}"), FMqtt.file_id, FileUploadChunckSize());
} else {
FMqtt.md5.calculate();
if (strcasecmp(FMqtt.file_md5.c_str(), FMqtt.md5.toString().c_str())) {
ResponseCmndChar(PSTR(D_JSON_MD5_MISMATCH));
} else {
// Process upload data en free buffer
ResponseCmndDone();

if (UPL_TASMOTA == FMqtt.file_type) {
if (!Update.end(true)) {
ResponseCmndFailed();
} else {
TasmotaGlobal.restart_flag = 2; // Always restart to re-enable disabled features during update
}
}
else if (UPL_SETTINGS == FMqtt.file_type) {
if (!SettingsConfigRestore()) {
ResponseCmndFailed();
} else {
TasmotaGlobal.restart_flag = 2; // Always restart to re-enable disabled features during update
}
}

}
FMqtt.file_buffer = nullptr;
}
}

if (!FMqtt.file_buffer) {
TasmotaGlobal.masterlog_level = LOG_LEVEL_NONE; // Enable logging
FMqtt.file_id = 0;
FMqtt.file_size = 0;
FMqtt.file_type = 0;
FMqtt.file_md5 = (const char*) nullptr; // Force deallocation of the String internal memory
}
MqttPublishPrefixTopic_P(STAT, XdrvMailbox.command); // Enforce stat/wemos10/FILEUPLOAD
ResponseClear();
}

void CmndFileDownload(void) {
/*
Download (binary) max 700 bytes chunks of data base64 encoded with MD5 hash over base64 decoded data
Currently supports Settings (file type 2)
Filedownload 0 - Abort current download
FileDownload 2 - Start download of settings file
FileDownload - Continue downloading data until reception of MD5 hash
*/

if (FMqtt.file_id && FMqtt.file_buffer) {
bool finished = false;

if (0 == XdrvMailbox.payload) { // Abort file download
ResponseCmndChar(PSTR(D_JSON_ABORTED));
finished = true;
}
else if (FMqtt.file_pos < FMqtt.file_size) {
uint32_t bytes_left = FMqtt.file_size - FMqtt.file_pos;
uint32_t write_bytes = (bytes_left < mqtt_file_chuck_size) ? bytes_left : mqtt_file_chuck_size;

uint8_t* buffer = FMqtt.file_buffer + FMqtt.file_pos;
FMqtt.md5.add(buffer, write_bytes);

// {"Id":117,"Data":"CRJcTQ9fYGF ... OT1BRUlNUVVZXWFk="}
Response_P(PSTR("{\"Id\":%d,\"Data\":\""), FMqtt.file_id); // FileTransferHeaderSize
char base64_data[encode_base64_length(write_bytes)];
encode_base64((unsigned char*)buffer, write_bytes, (unsigned char*)base64_data);
ResponseAppend_P(base64_data);
ResponseAppend_P("\"}");

FMqtt.file_pos += write_bytes;
} else {
FMqtt.md5.calculate();

// {"Id":117,"Md5":"496fcbb433bbca89833063174d2c5747"}
Response_P(PSTR("{\"Id\":%d,\"Md5\":\"%s\"}"), FMqtt.file_id, FMqtt.md5.toString().c_str());
finished = true;
}

if (finished) {
if (UPL_SETTINGS == FMqtt.file_type) {
SettingsBufferFree();
}

FMqtt.file_id = 0;
}
}
else if (XdrvMailbox.data_len) {
FMqtt.file_buffer = nullptr;
FMqtt.file_id = (UtcTime() & 0xFE) +1; // Odd id between 1 and 255

if (UPL_SETTINGS == XdrvMailbox.payload) {
uint32_t len = SettingsConfigBackup();
if (len) {
FMqtt.file_type = UPL_SETTINGS;
FMqtt.file_buffer = settings_buffer;
FMqtt.file_size = len;

// {"File":"Config_wemos10_9.4.0.3.dmp","Id":117,"Type":2,"Size":4096}
Response_P(PSTR("{\"File\":\"%s\",\"Id\":%d,\"Type\":%d,\"Size\":%d}"),
SettingsConfigFilename().c_str(), FMqtt.file_id, FMqtt.file_type, len);
}
}

if (FMqtt.file_buffer) {
FMqtt.file_pos = 0;

FMqtt.md5 = MD5Builder();
FMqtt.md5.begin();
} else {
FMqtt.file_id = 0;
ResponseCmndFailed();
}
}
MqttPublishPrefixTopic_P(STAT, XdrvMailbox.command);
ResponseClear();
}

#endif // USE_MQTT_FILE
Loading

0 comments on commit 75c783d

Please sign in to comment.