Skip to content

Commit

Permalink
influxdb: fix http response parsing, refactor module scope (#2153)
Browse files Browse the repository at this point in the history
  • Loading branch information
@mcspr
mcspr committed Feb 18, 2020
1 parent 3747d2e commit be7ca13
Show file tree
Hide file tree
Showing 2 changed files with 77 additions and 66 deletions.
138 changes: 73 additions & 65 deletions code/espurna/influxdb.ino
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,79 +8,92 @@ Copyright (C) 2017-2019 by Xose Pérez <xose dot perez at gmail dot com>

#if INFLUXDB_SUPPORT

#include <ESPAsyncTCP.h>
#include <map>
#include <memory>

#include "broker.h"
#include "libs/AsyncClientHelpers.h"

const char INFLUXDB_REQUEST_TEMPLATE[] PROGMEM = "POST /write?db=%s&u=%s&p=%s HTTP/1.1\r\nHost: %s:%u\r\nContent-Length: %d\r\n\r\n";
const char InfluxDb_http_success[] = "HTTP/1.1 204";
const char InfluxDb_http_template[] PROGMEM = "POST /write?db=%s&u=%s&p=%s HTTP/1.1\r\nHost: %s:%u\r\nContent-Length: %d\r\n\r\n";

constexpr const unsigned long INFLUXDB_CLIENT_TIMEOUT = 5000;
constexpr const size_t INFLUXDB_DATA_BUFFER_SIZE = 256;
class AsyncInfluxDB : public AsyncClient {
public:

bool _idb_enabled = false;
String _idb_host;
uint16_t _idb_port = 0;
constexpr static const unsigned long ClientTimeout = 5000;
constexpr static const size_t DataBufferSize = 256;

AsyncClientState state = AsyncClientState::Disconnected;
String host;
uint16_t port = 0;

std::map<String, String> _idb_values;
String _idb_data;
bool _idb_flush = false;
std::map<String, String> values;
String payload;

std::unique_ptr<AsyncClient> _idb_client = nullptr;
bool _idb_connecting = false;
bool _idb_connected = false;
bool flush = false;
uint32_t timestamp = 0;
};

uint32_t _idb_client_ts = 0;
bool _idb_enabled = false;
std::unique_ptr<AsyncInfluxDB> _idb_client = nullptr;

// -----------------------------------------------------------------------------

void _idbInitClient() {

_idb_client = std::make_unique<AsyncClient>();
_idb_client = std::make_unique<AsyncInfluxDB>();

_idb_client->onDisconnect([](void * s, AsyncClient * client) {
_idb_client->payload.reserve(AsyncInfluxDB::DataBufferSize);

_idb_client->onDisconnect([](void * s, AsyncClient * ptr) {
auto *client = reinterpret_cast<AsyncInfluxDB*>(ptr);
DEBUG_MSG_P(PSTR("[INFLUXDB] Disconnected\n"));
_idb_flush = false;
_idb_data = "";
_idb_client_ts = 0;
_idb_connected = false;
_idb_connecting = false;
client->flush = false;
client->payload = "";
client->timestamp = 0;
client->state = AsyncClientState::Disconnected;
}, nullptr);

_idb_client->onTimeout([](void * s, AsyncClient * client, uint32_t time) {
DEBUG_MSG_P(PSTR("[INFLUXDB] Network timeout after %ums\n"), time);
client->close(true);
}, nullptr);

_idb_client->onData([](void * arg, AsyncClient * client, void * response, size_t len) {
_idb_client->onData([](void * arg, AsyncClient * ptr, void * response, size_t len) {
// ref: https://docs.influxdata.com/influxdb/v1.7/tools/api/#summary-table-1
const char idb_success[] = "HTTP/1.1 204";
const bool result = (len > sizeof(idb_success) && (0 == strncmp((char*) response, idb_success, sizeof(idb_success))));
DEBUG_MSG_P(PSTR("[INFLUXDB] %s response after %ums\n"), result ? "Success" : "Failure", millis() - _idb_client_ts);
_idb_client_ts = millis();
client->close();
auto *client = reinterpret_cast<AsyncInfluxDB*>(ptr);
if (client->state == AsyncClientState::Connected) {
client->state = AsyncClientState::Disconnecting;

const bool result = (len > sizeof(InfluxDb_http_success) && (0 == strncmp((char*) response, InfluxDb_http_success, strlen(InfluxDb_http_success))));
DEBUG_MSG_P(PSTR("[INFLUXDB] %s response after %ums\n"), result ? "Success" : "Failure", millis() - client->timestamp);

client->timestamp = millis();
client->close();
}
}, nullptr);

_idb_client->onPoll([](void * arg, AsyncClient * client) {
unsigned long ts = millis() - _idb_client_ts;
if (ts > INFLUXDB_CLIENT_TIMEOUT) {
_idb_client->onPoll([](void * arg, AsyncClient * ptr) {
auto *client = reinterpret_cast<AsyncInfluxDB*>(ptr);
unsigned long ts = millis() - client->timestamp;
if (ts > AsyncInfluxDB::ClientTimeout) {
DEBUG_MSG_P(PSTR("[INFLUXDB] No response after %ums\n"), ts);
client->close(true);
return;
}

if (_idb_data.length()) {
client->write(_idb_data.c_str(), _idb_data.length());
_idb_data = "";
if (client->payload.length()) {
client->write(client->payload.c_str(), client->payload.length());
client->payload = "";
}
});

_idb_client->onConnect([](void * arg, AsyncClient * client) {
_idb_client->onConnect([](void * arg, AsyncClient * ptr) {

auto *client = reinterpret_cast<AsyncInfluxDB*>(ptr);

_idb_client_ts = millis();
_idb_connected = true;
_idb_connecting = false;
client->timestamp = millis();
client->state = AsyncClientState::Connected;

DEBUG_MSG_P(PSTR("[INFLUXDB] Connected to %s:%u\n"),
IPAddress(client->getRemoteAddress()).toString().c_str(),
Expand All @@ -89,11 +102,11 @@ void _idbInitClient() {

constexpr const int BUFFER_SIZE = 256;
char headers[BUFFER_SIZE];
int len = snprintf_P(headers, sizeof(headers), INFLUXDB_REQUEST_TEMPLATE,
int len = snprintf_P(headers, sizeof(headers), InfluxDb_http_template,
getSetting("idbDatabase", INFLUXDB_DATABASE).c_str(),
getSetting("idbUsername", INFLUXDB_USERNAME).c_str(),
getSetting("idbPassword", INFLUXDB_PASSWORD).c_str(),
_idb_host.c_str(), _idb_port, _idb_data.length()
client->host.c_str(), client->port, client->payload.length()
);
if ((len < 0) || (len > BUFFER_SIZE - 1)) {
client->close(true);
Expand Down Expand Up @@ -151,59 +164,56 @@ void _idbBrokerStatus(const String& topic, unsigned char id, unsigned int value)

bool idbSend(const char * topic, const char * payload) {
if (!_idb_enabled) return false;
if (_idb_connecting || _idb_connected) return false;
if (_idb_client->state != AsyncClientState::Disconnected) return false;

_idb_values[topic] = payload;
_idb_flush = true;
_idb_client->values[topic] = payload;
_idb_client->flush = true;

return true;
}

void _idbSend(const String& host, const uint16_t port) {
if (_idb_connected || _idb_connecting) return;
if (_idb_client->state != AsyncClientState::Disconnected) return;

DEBUG_MSG_P(PSTR("[INFLUXDB] Sending to %s:%u\n"), host.c_str(), port);

// TODO: cache `Host: <host>:<port>` instead of storing things separately?
_idb_host = host;
_idb_port = port;

_idb_client_ts = millis();
_idb_connecting = _idb_client->connect(host.c_str(), port);
// TODO: cache `Host: <host>:<port>` header instead of storing things separately?
_idb_client->host = host;
_idb_client->port = port;
_idb_client->timestamp = millis();
_idb_client->state = _idb_client->connect(host.c_str(), port)
? AsyncClientState::Connecting
: AsyncClientState::Disconnected;

if (!_idb_connecting) {
if (_idb_client->state == AsyncClientState::Disconnected) {
DEBUG_MSG_P(PSTR("[INFLUXDB] Connection to %s:%u failed\n"), host.c_str(), port);
_idb_client->close(true);
}
}

void _idbFlush() {
// Clean-up client object when not in use
if (_idb_client && !_idb_enabled && !_idb_connected && !_idb_connecting) {
if (_idb_client && !_idb_enabled && (_idb_client->state == AsyncClientState::Disconnected)) {
_idb_client = nullptr;
}

// Wait until current connection is finished
if (!_idb_flush) return;
if (_idb_connected || _idb_connecting) return;
if (!_idb_client) return;
if (!_idb_client->flush) return;
if (_idb_client->state != AsyncClientState::Disconnected) return;

// Wait until connected
if (!wifiConnected()) return;

// TODO: MDNS_CLIENT_SUPPORT is deprecated
String host = getSetting("idbHost", INFLUXDB_HOST);
#if MDNS_CLIENT_SUPPORT
host = mdnsResolve(host);
#endif

const auto host = getSetting("idbHost", INFLUXDB_HOST);
const auto port = getSetting<uint16_t>("idbPort", INFLUXDB_PORT);

// TODO: should we always store specific pairs like tspk keeps relay / sensor readings?
// note that we also send heartbeat data, persistent values should be flagged
const String device = getSetting("hostname");

_idb_data = "";
for (auto& pair : _idb_values) {
_idb_client->payload = "";
for (auto& pair : _idb_client->values) {
if (!isNumber(pair.second.c_str())) {
String quoted;
quoted.reserve(pair.second.length() + 2);
Expand All @@ -218,9 +228,9 @@ void _idbFlush() {
PSTR("%s,device=%s value=%s\n"),
pair.first.c_str(), device.c_str(), pair.second.c_str()
);
_idb_data += buffer;
_idb_client->payload += buffer;
}
_idb_values.clear();
_idb_client->values.clear();

_idbSend(host, port);
}
Expand Down Expand Up @@ -254,8 +264,6 @@ void idbSetup() {
espurnaRegisterReload(_idbConfigure);
espurnaRegisterLoop(_idbFlush);

_idb_data.reserve(INFLUXDB_DATA_BUFFER_SIZE);

#if TERMINAL_SUPPORT
terminalRegisterCommand(F("IDB.SEND"), [](Embedis* e) {
if (e->argc != 4) {
Expand Down
5 changes: 4 additions & 1 deletion code/espurna/libs/AsyncClientHelpers.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,13 @@

#pragma once

#include <ESPAsyncTCP.h>

enum class AsyncClientState {
Disconnected,
Connecting,
Connected
Connected,
Disconnecting
};


0 comments on commit be7ca13

Please sign in to comment.