north_http.cpp

/*
 * Fledge HTTP north plugin.
 *
 * Copyright (c) 2020 Dianomic Systems
 *
 * Released under the Apache 2.0 Licence
 *
 * Author: Mark Riddoch
 */

#include <http_north.h>
#include <iostream>
#include <string_utils.h>
#include <rapidjson/document.h>
#include <crypto.hpp>

using namespace std;
using namespace rapidjson;

/**
 * Constructor for the north HTTP plugin class.
 * Creates the two streams and sets the failed over state
 *
 * @param config	The plugin configuration
 */
HttpNorth::HttpNorth(ConfigCategory *config) : m_failedOver(false)
{
	string url = config->getValue("URL");
	m_primary = new HttpStream(config, url);
	if (!m_username.empty())
		m_primary->setAuth(m_username, m_password);
	url = config->getValue("URL2");
	if (url.length() > 0)
	{
		m_secondary = new HttpStream(config, url);
		if (!m_username.empty())
			m_primary->setAuth(m_username, m_password);
	}
	else
	{
		m_secondary = NULL;
	}
	if (config->itemExists("proxy"))
	{
		string proxy = config->getValue("proxy");
		if (proxy.compare(0, 5, "http:") == 0 || proxy.compare(0, 5, "HTTP:") == 0 
				|| proxy.compare(0, 6, "https:") == 0  || proxy.compare(0, 6, "HTTPS:") == 0)
		{
			Logger::getLogger()->warn("Expected proxy address without protocol prefix");
			size_t found = proxy.find_first_of("//");
			if (found != string::npos)
			{
				found += 2;
				string s = proxy.substr(found);
				found = s.find_first_of("/");
				if (found != string::npos)
				{
					proxy = s.substr(0, found);
				}
				else
				{
					proxy = s;
				}
				Logger::getLogger()->warn("Stripped of URL components to use '%s' as proxy", proxy.c_str());
			}
		}
		Logger::getLogger()->info("Using proxy server %s", proxy.c_str());
		m_primary->setProxy(proxy);
		if (m_secondary)
			m_secondary->setProxy(proxy);
	}
	string headers = config->getValue("headers");
	Document doc;
	doc.Parse(headers.c_str());
	if (doc.HasParseError() == false && doc.IsObject())
	{
		for (auto& m : doc.GetObject())
		{
			if (m.value.IsString())
			{
				string name = m.name.GetString();
				string value = m.value.GetString();
				m_primary->addHeader(name, value);
				if (m_secondary)
					m_secondary->addHeader(name, value);
			}
		}
	}

	// Process any script
	m_script = config->getItemAttribute("script", ConfigCategory::FILE_ATTR);
	m_content = config->getValue("script");
        if (m_script.empty() == false && m_content.empty() == false)
	{
		if ((m_python = new PythonScript(m_script)) != NULL)
		{
			Logger::getLogger()->info("Initialise script '%s': %s", m_script.c_str(), m_content.c_str());
			m_python->setScript(m_content);
		}
		else
		{
			Logger::getLogger()->error("Failed to create a Python runtime");
		}
	}
	else
	{
		m_python = NULL;
	}
	m_username = config->getValue("username");
	m_password = config->getValue("password");
}

/**
 * Destructor for the HTTP north class
 */
HttpNorth::~HttpNorth()
{
	if (m_primary)
		delete m_primary;
	if (m_secondary)
		delete m_secondary;
	if (m_python)
		delete m_python;
}

/**
 * Send the set of readings to the north system.
 *
 * Uses the priamry or secondary connection, if defined to send
 * the readings as JSON to the north FogLAMP.
 *
 * The member varaible m_failedOver is used to determine if the
 * primary or secoindary connection should be used.
 *
 * @param readings	The readings to send
 * @return uint32_t 	The numebr of readings sent
 */
uint32_t HttpNorth::send(const vector<Reading *> readings)
{
	string data;

	if (m_python)
	{
		uint32_t rval = 0;

		for (vector<Reading *>::const_iterator elem = readings.begin(); elem != readings.end(); ++elem)
		{
			if (m_python->execute(*elem, data) && sendData(data))
		{
				rval++;
			}
			else
			{
				Logger::getLogger()->error("Failed to convert and send payload");
				return rval;
			}
		}

		return rval;
	}
	else
	{
		ostringstream jsonData;
		jsonData << "[";

		// Fetch Reading* data
		for (vector<Reading *>::const_iterator elem = readings.begin(); elem != readings.end(); ++elem)
		{
			string value;
			getReadingString(value, **elem);
			jsonData << value << (elem < (readings.end() -1 ) ? ", " : "");
		}

		jsonData << "]";
		data = jsonData.str();
		if (sendData(data))
			return readings.size();
		else
			return 0;
	}
}

/**
 * Send the data to the HTTP destination
 *
 * @param data	The data to send
 * @return bool	True if the data was sent
 */
bool HttpNorth::sendData(const string& data)
{
	Logger::getLogger()->debug("Send data %s", data.c_str());
	lock_guard<mutex> guard(m_mutex);
	if (m_failedOver)
	{
		if (m_secondary && m_secondary->send(data))
		{
			return true;
		}
		if (m_primary && m_primary->send(data))
		{
			m_failedOver = false;
			return true;
		}
	}
	else
	{
		if (m_primary && m_primary->send(data))
		{
			return true;
		}
		if (m_secondary && m_secondary->send(data))
		{
			m_failedOver = true;
			return true;
		}
	}
	return false;
}

/**
 * Format a reading to send
 *
 * @param value	The JSON formatted reading
 * @param reading	The reading to format
 */
void HttpNorth::getReadingString(string& value, const Reading& reading)
{
	// Convert reading data into JSON string
	value.append("{\"timestamp\" : \"" + reading.getAssetDateUserTime(Reading::FMT_STANDARD) + "Z" + "\"");
	value.append(",\"asset\" : \"" + reading.getAssetName() + "\"");
	value.append(",\"readings\" : {");

	// Get reading data
	const vector<Datapoint*> data = reading.getReadingData();

	bool first = true;
	for (vector<Datapoint*>::const_iterator it = data.begin(); it != data.end(); ++it)
	{
		if (first)
			first = false;
		else
			value.append(",");
		if ((*it)->getData().getType() == DatapointValue::T_STRING)
		{
			string str = (*it)->getData().toStringValue();
			StringEscapeQuotes(str);
			value.append("\"" + (*it)->getName() + "\": \"" + str + "\"");
		}
		else
		{
			value.append("\"" + (*it)->getName() + "\": " + (*it)->getData().toString());
		}
	}

	value.append("}}");
}

/**
 * Constructor for the HttpStream class
 *
 * @param config	The plugin configuration
 * @param url		The url to connect to
 */

HttpNorth::HttpStream::HttpStream(ConfigCategory *config, string& url) : m_sender(NULL)
{
	/**
	 * Handle the HTTP(S) parameters here
	 */
	unsigned int retrySleepTime = atoi(config->getValue("retrySleepTime").c_str());
	unsigned int maxRetry = atoi(config->getValue("maxRetry").c_str());
	unsigned int timeout = atoi(config->getValue("HttpTimeout").c_str());
	/**
	 * Extract host, port, path from URL
	 */
	size_t findProtocol = url.find_first_of(":");
	string protocol = url.substr(0,findProtocol);

	string tmpUrl = url.substr(findProtocol + 3);
	size_t findPort = tmpUrl.find_first_of(":");
	string hostName = tmpUrl.substr(0, findPort);

	size_t findPath = tmpUrl.find_first_of("/");
	string port;
	if (findPath == string::npos)
	{
		port = tmpUrl.substr(findPort + 1);
		m_path = "/";
	}
	else
	{
		port = tmpUrl.substr(findPort + 1 , findPath - findPort -1);
		m_path = tmpUrl.substr(findPath);
	}

	/**
	 * Allocate the HTTP(S) handler for "Hostname : port",
	 * connect_timeout and request_timeout.
	 */
	string hostAndPort(hostName + ":" + port);	

	if (protocol == string("http"))
		m_sender = new SimpleHttp(hostAndPort,
							timeout,
							timeout,
							retrySleepTime,
							maxRetry);
	else if (protocol == string("https"))
		m_sender = new SimpleHttps(hostAndPort,
							 timeout,
							 timeout,
							 retrySleepTime,
							 maxRetry);

	if (!m_sender)
		Logger::getLogger()->error("http-north C plugin: Failed to create HTTP(S) sender for %s", url.c_str());

	m_header.push_back(pair<string, string>("Content-Type", "application/json"));
}

/**
 * Desstructor fo the HttpStream
 */
HttpNorth::HttpStream::~HttpStream()
{
	if (m_sender)
		delete m_sender;
}

/**
 * Set a proxy server for the HTTP connection
 *
 * @param proxy	The host and port of the proxy server
 */
void HttpNorth::HttpStream::setProxy(const string& proxy)
{
	if (m_sender)
		m_sender->setProxy(proxy);
}

/**
 * Add header fields
 *
 * @param name	The header field name
 * @param value	the header field value
 */
void HttpNorth::HttpStream::addHeader(const string& name, const string& value)
{
	m_header.push_back(pair<string, string>(string(name), string(value)));
}

/**
 * Create the credentials
 *
 * @param username	The username
 * @param password	The password
 */
void HttpNorth::HttpStream::setAuth(const string& username, const string& password)
{
	string credentials = SimpleWeb::Crypto::Base64::encode(username + ":" + password);

	m_header.push_back(pair<string, string>("Authorization", "Basic " + credentials));
}

/**
 * Send the given payload on this HttpStream
 *
 * @param data	The JSON payload
 * @return bool	The status of the send operation.
 */
bool HttpNorth::HttpStream::send(const string& data)
{
	if (!m_sender)
		return false;
	
	try
	{
		int res = m_sender->sendRequest("POST", m_path, m_header, data);
		if (res != 200 && res != 204 && res != 201)
		{
			Logger::getLogger()->error("http-north C plugin: Sending JSON readings HTTP(S) error: %d", res);
			return false;
		}

		Logger::getLogger()->info("http-north C plugin: Successfully sent readings");

		// Return number of sent readings to the caller
		return true;
	}
	catch (const std::exception& e)
	{
		Logger::getLogger()->error("http-north C plugin: Sending JSON data exception: %s", e.what());
		return false;
	}
}