Peter M. Groen / FlashMQ

Commit 3022c275e8f778c89ee97c58b2e0df8d04fdb904

Authored by Wiebe Cazemier
1 parent 9dab7a48

Initiate topic alias to subscribers 

I think it's correct, but mosquitto_sub doesn't seem to support it, so I
can't test.

Also some other stuff I happen to see.
Inline Side-by-side
Showing 10 changed files with 68 additions and 20 deletions
client.cpp
  View file @3022c27
... ... @@ -208,7 +208,23 @@ int Client::writeMqttPacket(const MqttPacket &packet)
208 208  
209 209 int Client::writeMqttPacketAndBlameThisClient(PublishCopyFactory &copyFactory, char max_qos, uint16_t packet_id)
210 210 {
211   - MqttPacket *p = copyFactory.getOptimumPacket(max_qos, this->protocolVersion);
  211 + const Settings *settings = ThreadGlobals::getSettings();
  212 + uint16_t topic_alias = 0;
  213 + bool skip_topic = false;
  214 +
  215 + if (protocolVersion >= ProtocolVersion::Mqtt5 && settings->maxOutgoingTopicAliases > this->curOutgoingTopicAlias)
  216 + {
  217 + uint16_t &id = this->outgoingTopicAliases[copyFactory.getTopic()];
  218 +
  219 + if (id > 0)
  220 + skip_topic = true;
  221 + else
  222 + id = ++this->curOutgoingTopicAlias;
  223 +
  224 + topic_alias = id;
  225 + }
  226 +
  227 + MqttPacket *p = copyFactory.getOptimumPacket(max_qos, this->protocolVersion, topic_alias, skip_topic);
212 228  
213 229 assert(p->getQos() <= max_qos);
214 230  
... ...
client.h
  View file @3022c27
... ... @@ -84,6 +84,9 @@ class Client
84 84  
85 85 std::unordered_map<uint16_t, std::string> topicAliases;
86 86  
  87 + uint16_t curOutgoingTopicAlias = 0;
  88 + std::unordered_map<std::string, uint16_t> outgoingTopicAliases;
  89 +
87 90 Logger *logger = Logger::getInstance();
88 91  
89 92 void setReadyForWriting(bool val);
... ...
mqtt5properties.cpp
  View file @3022c27
... ... @@ -125,6 +125,11 @@ void Mqtt5PropertyBuilder::writeCorrelationData(const std::string &amp;correlationDa
125 125 writeStr(Mqtt5Properties::CorrelationData, correlationData);
126 126 }
127 127  
  128 +void Mqtt5PropertyBuilder::writeTopicAlias(const uint16_t id)
  129 +{
  130 + writeUint16(Mqtt5Properties::TopicAlias, id, clientSpecificBytes);
  131 +}
  132 +
128 133 void Mqtt5PropertyBuilder::setNewUserProperties(const std::shared_ptr<std::vector<std::pair<std::string, std::string>>> &userProperties)
129 134 {
130 135 assert(!this->userProperties);
... ... @@ -134,7 +139,7 @@ void Mqtt5PropertyBuilder::setNewUserProperties(const std::shared_ptr&lt;std::vecto
134 139 this->userProperties = userProperties;
135 140 }
136 141  
137   -void Mqtt5PropertyBuilder::writeUint32(Mqtt5Properties prop, const uint32_t x, std::vector<char> &target)
  142 +void Mqtt5PropertyBuilder::writeUint32(Mqtt5Properties prop, const uint32_t x, std::vector<char> &target) const
138 143 {
139 144 size_t pos = target.size();
140 145 const size_t newSize = pos + 5;
... ... @@ -154,16 +159,21 @@ void Mqtt5PropertyBuilder::writeUint32(Mqtt5Properties prop, const uint32_t x, s
154 159  
155 160 void Mqtt5PropertyBuilder::writeUint16(Mqtt5Properties prop, const uint16_t x)
156 161 {
157   - size_t pos = genericBytes.size();
  162 + writeUint16(prop, x, this->genericBytes);
  163 +}
  164 +
  165 +void Mqtt5PropertyBuilder::writeUint16(Mqtt5Properties prop, const uint16_t x, std::vector<char> &target) const
  166 +{
  167 + size_t pos = target.size();
158 168 const size_t newSize = pos + 3;
159   - genericBytes.resize(newSize);
  169 + target.resize(newSize);
160 170  
161 171 const uint8_t a = static_cast<uint8_t>(x >> 8);
162 172 const uint8_t b = static_cast<uint8_t>(x);
163 173  
164   - genericBytes[pos++] = static_cast<uint8_t>(prop);
165   - genericBytes[pos++] = a;
166   - genericBytes[pos] = b;
  174 + target[pos++] = static_cast<uint8_t>(prop);
  175 + target[pos++] = a;
  176 + target[pos] = b;
167 177 }
168 178  
169 179 void Mqtt5PropertyBuilder::writeUint8(Mqtt5Properties prop, const uint8_t x)
... ...
mqtt5properties.h
  View file @3022c27
... ... @@ -13,8 +13,9 @@ class Mqtt5PropertyBuilder
13 13 std::shared_ptr<std::vector<std::pair<std::string, std::string>>> userProperties;
14 14 VariableByteInt length;
15 15  
16   - void writeUint32(Mqtt5Properties prop, const uint32_t x, std::vector<char> &target);
  16 + void writeUint32(Mqtt5Properties prop, const uint32_t x, std::vector<char> &target) const;
17 17 void writeUint16(Mqtt5Properties prop, const uint16_t x);
  18 + void writeUint16(Mqtt5Properties prop, const uint16_t x, std::vector<char> &target) const;
18 19 void writeUint8(Mqtt5Properties prop, const uint8_t x);
19 20 void writeStr(Mqtt5Properties prop, const std::string &str);
20 21 void write2Str(Mqtt5Properties prop, const std::string &one, const std::string &two);
... ... @@ -43,6 +44,7 @@ public:
43 44 void writeResponseTopic(const std::string &str);
44 45 void writeUserProperty(std::string &&key, std::string &&value);
45 46 void writeCorrelationData(const std::string &correlationData);
  47 + void writeTopicAlias(const uint16_t id);
46 48 void setNewUserProperties(const std::shared_ptr<std::vector<std::pair<std::string, std::string>>> &userProperties);
47 49 };
48 50  
... ...
mqttpacket.cpp
  View file @3022c27
... ... @@ -108,7 +108,8 @@ MqttPacket::MqttPacket(const ProtocolVersion protocolVersion, const Publish &amp;_pu
108 108  
109 109 this->protocolVersion = protocolVersion;
110 110  
111   - this->publishData.topic = _publish.topic;
  111 + if (!_publish.skipTopic)
  112 + this->publishData.topic = _publish.topic;
112 113  
113 114 if (_publish.splitTopic)
114 115 splitTopic(this->publishData.topic, this->publishData.subtopics);
... ... @@ -133,7 +134,7 @@ MqttPacket::MqttPacket(const ProtocolVersion protocolVersion, const Publish &amp;_pu
133 134 if (protocolVersion >= ProtocolVersion::Mqtt5)
134 135 {
135 136 // Step 1: make certain properties available as objects, because FlashMQ needs access to them for internal logic.
136   - if (_publish.propertyBuilder)
  137 + if (_publish.propertyBuilder) // TODO: only do this when there are user properties. Otherwise we don't need it.
137 138 {
138 139 this->publishData.constructPropertyBuilder();
139 140 this->publishData.propertyBuilder->setNewUserProperties(_publish.propertyBuilder->getUserProperties());
... ... @@ -343,7 +344,7 @@ void MqttPacket::handleConnect()
343 344 uint16_t max_qos_packets = settings.maxQosMsgPendingPerClient;
344 345 uint32_t session_expire = settings.expireSessionsAfterSeconds > 0 ? settings.expireSessionsAfterSeconds : std::numeric_limits<uint32_t>::max();
345 346 uint32_t max_packet_size = settings.maxPacketSize;
346   - uint16_t max_topic_aliases = settings.maxTopicAliases;
  347 + uint16_t max_topic_aliases = settings.maxOutgoingTopicAliases;
347 348 bool request_response_information = false;
348 349 bool request_problem_information = false;
349 350  
... ... @@ -820,6 +821,7 @@ void MqttPacket::handlePublish()
820 821 case Mqtt5Properties::MessageExpiryInterval:
821 822 publishData.createdAt = std::chrono::steady_clock::now();
822 823 publishData.expiresAfter = std::chrono::seconds(readFourBytesToUint32());
  824 + break;
823 825 case Mqtt5Properties::TopicAlias:
824 826 {
825 827 const uint16_t alias_id = readTwoBytesToUInt16();
... ...
publishcopyfactory.cpp
  View file @3022c27
... ... @@ -17,15 +17,17 @@ PublishCopyFactory::PublishCopyFactory(Publish *publish) :
17 17  
18 18 }
19 19  
20   -MqttPacket *PublishCopyFactory::getOptimumPacket(const char max_qos, const ProtocolVersion protocolVersion)
  20 +MqttPacket *PublishCopyFactory::getOptimumPacket(const char max_qos, const ProtocolVersion protocolVersion, uint16_t topic_alias, bool skip_topic)
21 21 {
22 22 if (packet)
23 23 {
24   - if (packet->containsClientSpecificProperties())
  24 + if (protocolVersion >= ProtocolVersion::Mqtt5 && (packet->containsClientSpecificProperties() || topic_alias > 0))
25 25 {
26 26 Publish newPublish(packet->getPublishData());
27 27 newPublish.splitTopic = false;
28 28 newPublish.qos = max_qos;
  29 + newPublish.topicAlias = topic_alias;
  30 + newPublish.skipTopic = skip_topic;
29 31 newPublish.setClientSpecificProperties();
30 32 this->oneShotPacket = std::make_unique<MqttPacket>(protocolVersion, newPublish);
31 33 return this->oneShotPacket.get();
... ...
publishcopyfactory.h
  View file @3022c27
... ... @@ -29,7 +29,7 @@ public:
29 29 PublishCopyFactory(const PublishCopyFactory &other) = delete;
30 30 PublishCopyFactory(PublishCopyFactory &&other) = delete;
31 31  
32   - MqttPacket *getOptimumPacket(const char max_qos, const ProtocolVersion protocolVersion);
  32 + MqttPacket *getOptimumPacket(const char max_qos, const ProtocolVersion protocolVersion, uint16_t topic_alias, bool skip_topic);
33 33 char getEffectiveQos(char max_qos) const;
34 34 const std::string &getTopic() const;
35 35 const std::vector<std::string> &getSubtopics();
... ...
settings.h
  View file @3022c27
... ... @@ -42,7 +42,8 @@ public:
42 42 bool authPluginSerializeAuthChecks = false;
43 43 int clientInitialBufferSize = 1024; // Must be power of 2
44 44 int maxPacketSize = 268435461; // 256 MB + 5
45   - uint16_t maxTopicAliases = 65535;
  45 + uint16_t maxIncomingTopicAliases = 65535;
  46 + uint16_t maxOutgoingTopicAliases = 0; // TODO: setting, when I can confirm with clients that support it that it works.
46 47 #ifdef TESTING
47 48 bool logDebug = true;
48 49 #else
... ...
types.cpp
  View file @3022c27
... ... @@ -117,7 +117,8 @@ PublishBase::PublishBase(const std::string &amp;topic, const std::string &amp;payload, c
117 117  
118 118 size_t PublishBase::getLengthWithoutFixedHeader() const
119 119 {
120   - int result = topic.length() + payload.length() + 2;
  120 + const int topicLength = this->skipTopic ? 0 : topic.length();
  121 + int result = topicLength + payload.length() + 2;
121 122  
122 123 if (qos)
123 124 result += 2;
... ... @@ -131,14 +132,23 @@ size_t PublishBase::getLengthWithoutFixedHeader() const
131 132 */
132 133 void PublishBase::setClientSpecificProperties()
133 134 {
  135 + if (this->createdAt.time_since_epoch().count() && this->topicAlias == 0)
  136 + return;
  137 +
134 138 if (propertyBuilder)
135 139 propertyBuilder->clearClientSpecificBytes();
  140 + else
  141 + propertyBuilder = std::make_shared<Mqtt5PropertyBuilder>();
136 142  
137   - auto now = std::chrono::steady_clock::now();
138   - std::chrono::seconds newExpiresAfter = std::chrono::duration_cast<std::chrono::seconds>(now - createdAt);
139   -
140   - if (newExpiresAfter.count() > 0)
  143 + if (createdAt.time_since_epoch().count() > 0)
  144 + {
  145 + auto now = std::chrono::steady_clock::now();
  146 + std::chrono::seconds newExpiresAfter = std::chrono::duration_cast<std::chrono::seconds>(now - createdAt);
141 147 propertyBuilder->writeMessageExpiryInterval(newExpiresAfter.count());
  148 + }
  149 +
  150 + if (topicAlias > 0)
  151 + propertyBuilder->writeTopicAlias(this->topicAlias);
142 152 }
143 153  
144 154 void PublishBase::constructPropertyBuilder()
... ...
... ... @@ -206,6 +206,8 @@ public:
206 206 bool splitTopic = true;
207 207 std::chrono::time_point<std::chrono::steady_clock> createdAt;
208 208 std::chrono::seconds expiresAfter;
  209 + uint16_t topicAlias = 0;
  210 + bool skipTopic = false;
209 211 std::shared_ptr<Mqtt5PropertyBuilder> propertyBuilder; // Only contains data for sending, not receiving
210 212  
211 213 PublishBase() = default;
... ...