-
Notifications
You must be signed in to change notification settings - Fork 547
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
MQTT proof-of-concept for outgoing connections
- Loading branch information
Showing
15 changed files
with
537 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,276 @@ | ||
| /* | ||
| * MqttClient.cpp | ||
| */ | ||
| #include "MqttClient.h" | ||
|
|
||
| #if SUPPORT_MQTT | ||
|
|
||
| #include <Platform/Platform.h> | ||
| #include <Networking/Network.h> | ||
|
|
||
| #include "Socket.h" | ||
|
|
||
| #include "mqtt.h" | ||
|
|
||
| MqttClient::MqttClient(NetworkResponder *n, NetworkClient *c) noexcept : NetworkClient(n, c), | ||
| sendBuf(nullptr), | ||
| recvBuf(nullptr), | ||
| client(nullptr), | ||
| username(nullptr), | ||
| password(nullptr), | ||
| clientId(nullptr), | ||
| publishTopic(nullptr), | ||
| willClient(nullptr), | ||
| willTopic(nullptr), | ||
| willMessage(nullptr), | ||
| willMessageSize(0), | ||
| keepAlive(DefaultKeepAlive), | ||
| qos(QOS::QOS0), | ||
| retain(false), | ||
| dup(false) | ||
| { | ||
| client = new mqtt_client; | ||
| publishMutex.Create("MqttPublishBuffer"); | ||
|
|
||
| sendBuf = new uint8_t[SendBufferSize]; | ||
| recvBuf = new uint8_t[ReceiveBufferSize]; | ||
|
|
||
| mqtt_init(client, skt, sendBuf, SendBufferSize, recvBuf, ReceiveBufferSize, PublishCallback); | ||
| } | ||
|
|
||
| bool MqttClient::Subscribe(const char *topic, QOS maxQos) | ||
| { | ||
| enum MQTTErrors err = mqtt_subscribe(client, topic, static_cast<int>(maxQos)); | ||
| return err == MQTT_OK; | ||
| } | ||
|
|
||
| bool MqttClient::Unsubscribe(const char *topic) | ||
| { | ||
| enum MQTTErrors err = mqtt_unsubscribe(client, topic); | ||
| return err == MQTT_OK; | ||
| } | ||
|
|
||
| bool MqttClient::Publish(const char *topic, const void *message, size_t messageSize, QOS qos, bool retain, bool dup) | ||
| { | ||
| uint8_t flags = 0; | ||
|
|
||
| switch (qos) | ||
| { | ||
| case QOS::QOS0: | ||
| flags |= MQTT_PUBLISH_QOS_0; | ||
| break; | ||
|
|
||
| case QOS::QOS1: | ||
| flags |= MQTT_PUBLISH_QOS_1; | ||
| break; | ||
|
|
||
| case QOS::QOS2: | ||
| flags |= MQTT_PUBLISH_QOS_2; | ||
| break; | ||
|
|
||
| default: | ||
| break; | ||
| } | ||
|
|
||
| flags |= (retain) ? MQTT_PUBLISH_RETAIN : 0; | ||
| flags |= (dup) ? MQTT_PUBLISH_DUP : 0; | ||
|
|
||
| enum MQTTErrors err = mqtt_publish(client, topic, message, messageSize, flags); | ||
| return err == MQTT_OK; | ||
| } | ||
|
|
||
| bool MqttClient::Spin() noexcept | ||
| { | ||
| bool res = false; | ||
|
|
||
| if (skt && skt->CanSend()) | ||
| { | ||
| switch (responderState) | ||
| { | ||
| case ResponderState::connecting: | ||
| { | ||
| mqtt_sync(client); | ||
|
|
||
| if (client->typical_response_time >= 0.0f) // ack was received | ||
| { | ||
| if (client->error == MQTT_OK) | ||
| { | ||
| responderState = ResponderState::connected; | ||
| } | ||
| else | ||
| { | ||
| // Re-attempt MQTT connection | ||
| Connect(); | ||
| } | ||
|
|
||
| // Since at this stage we only expect ack (or some error), only work is done | ||
| // when it is actually received and processed. | ||
| res = true; | ||
| } | ||
| } | ||
| break; | ||
|
|
||
| case ResponderState::connected: | ||
| { | ||
| // Save the current receive buffer, so it can be checked if | ||
| // it moved forward (received something). | ||
| uint8_t *curr = client->recv_buffer.curr; | ||
|
|
||
| // Process the socket receive buffer | ||
| mqtt_sync(client); | ||
|
|
||
| bool ok = true, sent = false; | ||
| { | ||
| MutexLocker lock(publishMutex); | ||
|
|
||
| while (outBuf && ok) | ||
| { | ||
| ok = Publish(publishTopic ? publishTopic : reprap.GetNetwork().GetHostname(), | ||
| outBuf->Data(),outBuf->DataLength(), qos, retain, dup); | ||
| if (ok) | ||
| { | ||
| outBuf = OutputBuffer::Release(outBuf); | ||
| sent = true; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (sent) | ||
| { | ||
| // Buffer the messages in the socket for the next loop | ||
| mqtt_sync(client); | ||
| } | ||
|
|
||
| // Did work if processed or sent something | ||
| res = (client->recv_buffer.curr > curr || sent); | ||
| } | ||
| break; | ||
|
|
||
| default: | ||
| break; | ||
| } | ||
| } | ||
| else | ||
| { | ||
| if (responderState == ResponderState::connecting || responderState == ResponderState::connected) | ||
| { | ||
| // Lost the connection | ||
| debugPrintf("connection lost\n"); | ||
| ConnectionLost(); | ||
| } | ||
| } | ||
|
|
||
| return res; | ||
| } | ||
|
|
||
| void MqttClient::Connect() | ||
| { | ||
| mqtt_reinit(client, skt, sendBuf, SendBufferSize, recvBuf, ReceiveBufferSize); | ||
| mqtt_connect(client, clientId, willTopic, willMessage, willMessageSize, username, password, MQTT_CONNECT_CLEAN_SESSION, keepAlive); | ||
| // Buffer the CONNECT message, so the next mqtt_sync can process CONNACK if available. | ||
| mqtt_sync(client); | ||
| responderState = ResponderState::connecting; | ||
| } | ||
|
|
||
| bool MqttClient::Accept(Socket *s, NetworkProtocol protocol) noexcept | ||
| { | ||
| if (responderState == ResponderState::free && HandlesProtocol(protocol) && s->GetInterface() == interface) | ||
| { | ||
| skt = s; | ||
| Connect(); | ||
| return true; | ||
| } | ||
|
|
||
| return false; | ||
| } | ||
|
|
||
| void MqttClient::Disable() noexcept | ||
| { | ||
| MutexLocker lock(MqttClient::Instance()->publishMutex); | ||
| OutputBuffer::ReleaseAll(MqttClient::Instance()->outBuf); | ||
| } | ||
|
|
||
| bool MqttClient::HandlesProtocol(NetworkProtocol protocol) noexcept | ||
| { | ||
| return protocol == MqttProtocol; | ||
| } | ||
|
|
||
| void MqttClient::Terminate(NetworkProtocol protocol, NetworkInterface *interface) noexcept | ||
| { | ||
| if (responderState != ResponderState::free && (HandlesProtocol(protocol) || protocol == AnyProtocol) && skt != nullptr && GetInterface() == interface) | ||
| { | ||
| ConnectionLost(); | ||
| return; | ||
| } | ||
| } | ||
|
|
||
| void MqttClient::Diagnostics(MessageType mt) const noexcept | ||
| { | ||
| GetPlatform().MessageF(mt, " MQTT(%d)", (int)responderState); | ||
| } | ||
|
|
||
| /*static*/ void MqttClient::QueueMessage(const char *msg) noexcept | ||
| { | ||
| if (strlen(msg) < OUTPUT_BUFFER_SIZE - 1) | ||
| { | ||
| MutexLocker lock(MqttClient::Instance()->publishMutex); | ||
| OutputBuffer *buf = nullptr; | ||
| if (OutputBuffer::Allocate(buf)) | ||
| { | ||
| buf->copy(msg); | ||
|
|
||
| if (MqttClient::Instance()->outBuf) | ||
| { | ||
| MqttClient::Instance()->outBuf->Append(buf); | ||
| } | ||
| else | ||
| { | ||
| MqttClient::Instance()->outBuf = buf; | ||
| } | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| /*static*/ void MqttClient::PublishCallback(void** state, struct mqtt_response_publish *published) | ||
| { | ||
| debugPrintf("Received message from topic '%s':\n %s \n", | ||
| (const char*) published->topic_name, (const char*) published->application_message); | ||
| } | ||
|
|
||
| /*static*/ MqttClient *MqttClient::instance; | ||
|
|
||
| /* | ||
| * MQTT-C PAL layer implementation | ||
| */ | ||
|
|
||
| ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags) | ||
| { | ||
| Socket* skt = static_cast<Socket*>(fd); | ||
|
|
||
| size_t res = 0; | ||
|
|
||
| if (len) | ||
| { | ||
| res = skt->Send((const uint8_t*)buf, len); | ||
| } | ||
|
|
||
| return res; | ||
| } | ||
|
|
||
| ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags) | ||
| { | ||
| Socket* skt = static_cast<Socket*>(fd); | ||
|
|
||
| size_t read = 0; | ||
| char c = 0; | ||
|
|
||
| while (read < bufsz && skt->ReadChar(c)) | ||
| { | ||
| ((char*) buf)[read] = c; | ||
| read++; | ||
| } | ||
|
|
||
| return read; | ||
| } | ||
|
|
||
| #endif |
Oops, something went wrong.