* @brief Drains both the frame queue and the raw-bytes queue. */
| 196 | * @brief Drains both the frame queue and the raw-bytes queue. |
| 197 | */ |
| 198 | void MQTT::PublisherWorker::processData() |
| 199 | { |
| 200 | DataModel::FrameConsumerWorker<DataModel::TimestampedFramePtr>::processData(); |
| 201 | |
| 202 | if (!consumerEnabled()) |
| 203 | return; |
| 204 | |
| 205 | if (!isResourceOpen()) |
| 206 | return; |
| 207 | |
| 208 | const int mode = m_mode->load(std::memory_order_relaxed); |
| 209 | |
| 210 | if (m_rawQueue && mode != static_cast<int>(Publisher::Mode::RawRxData)) { |
| 211 | TimestampedRawBytes drain; |
| 212 | while (m_rawQueue->try_dequeue(drain)) |
| 213 | ; |
| 214 | } |
| 215 | if (m_frameQueueBytes && mode != static_cast<int>(Publisher::Mode::ScriptDriven)) { |
| 216 | TimestampedRawBytes drain; |
| 217 | while (m_frameQueueBytes->try_dequeue(drain)) |
| 218 | ; |
| 219 | } |
| 220 | |
| 221 | if (m_cfg.topicBase.isEmpty()) |
| 222 | return; |
| 223 | |
| 224 | if (mode == static_cast<int>(Publisher::Mode::RawRxData) && m_rawQueue) { |
| 225 | QMqttTopicName topic(m_cfg.topicBase); |
| 226 | if (!topic.isValid()) |
| 227 | return; |
| 228 | |
| 229 | m_rawBatchBuffer.resize(0); |
| 230 | TimestampedRawBytes item; |
| 231 | while (m_rawQueue->try_dequeue(item)) |
| 232 | if (item.data && !item.data->data.isEmpty()) |
| 233 | m_rawBatchBuffer += item.data->data; |
| 234 | |
| 235 | if (!m_rawBatchBuffer.isEmpty()) |
| 236 | publishAndCount(topic, m_rawBatchBuffer); |
| 237 | |
| 238 | return; |
| 239 | } |
| 240 | |
| 241 | if (mode == static_cast<int>(Publisher::Mode::ScriptDriven) && m_frameQueueBytes) { |
| 242 | const QString topicStr = m_cfg.scriptTopic.isEmpty() ? m_cfg.topicBase : m_cfg.scriptTopic; |
| 243 | QMqttTopicName topic(topicStr); |
| 244 | if (!topic.isValid()) |
| 245 | return; |
| 246 | |
| 247 | recompileScriptIfNeeded(); |
| 248 | if (!m_script || !m_script->isLoaded()) |
| 249 | return; |
| 250 | |
| 251 | QByteArray aggregate; |
| 252 | aggregate.reserve(4096); |
| 253 | |
| 254 | TimestampedRawBytes item; |
| 255 | while (m_frameQueueBytes->try_dequeue(item)) { |