client.d

module influxdb.client;

import vibe.core.net;
import vibe.core.log;

string DEFAULT_ADDRESS = "127.0.0.1";
ushort DEFAULT_UDP_PORT = 4455;

private final class InfluxConnection
{
    private
    {
        string m_host;
        ushort m_port;
        UDPConnection m_conn;
    }

    this(string host, ushort port)
    {
        m_host = host;
        m_port = port;
        m_conn = listenUDP(0);

    }

    InfluxConnection connect()
    {
        m_conn.connect(m_host, m_port);
        return this;
    }

    void write(string payload)
    {
        m_conn.send(cast(ubyte[]) payload);
    }

}

import vibe.core.connectionpool;
import influxdb.measurement;

final class InfluxClient
{
    private
    {
        ConnectionPool!InfluxConnection m_connections;
    }

    this(string host = DEFAULT_ADDRESS, ushort port = DEFAULT_UDP_PORT)
    {
        m_connections = new ConnectionPool!InfluxConnection({
            return new InfluxConnection(host, port).connect();
        });
    }

    LockedConnection!InfluxConnection getConnection()
    {
        return m_connections.lockConnection();
    }

    import influxdb.measurement;

    void sendMeasurement(Measurement m)
    {
        string dta = m.toInfluxdbLine();
        import std.stdio;

        writeln(dta);
        getConnection().write(dta);
    }
}

unittest
{

    //needs an influxdb running
    InfluxClient c = new InfluxClient();

    auto m = Measurement("xxx2");
    m["val1"] = 44L;
    //m.addTag("tagKey", "tagVal");
    auto m2 = Measurement("xxx2");
    m2["val1"] = 55L;
    bool b = true;
    m2["bools"] = b;
    m2["bools2"] = false;
    m2.addTag("tagKey", "tagVal2");
    c.sendMeasurement(m);
    c.sendMeasurement(m2);
}

measurement.d

module influxdb.measurement;

import std.conv, std.string;
import std.algorithm : map, joiner, sort;
import std.traits;

struct InfluxData
{
    string[string] data;
    alias data this;

    @property string toInflux()
    {
        if (data.length == 0)
        {
            return "";
        }
        return data.keys.sort!().map!(key => key.replace(" ",
                "\\ ") ~ "=" ~ data[key]).joiner(",").to!string;
    }

    void opIndexAssign(T)(T value, string name)
    {
        data[name] = value.to!string;
    }

}

unittest
{

    InfluxData d = InfluxData();
    assert(d.toInflux == "");
    d["a"] = "1";
    d["b"] = "2";
    d["c c"] = "3";
    assert(d.toInflux == "a=1,b=2,c\\ c=3");

}

struct InfluxValues
{
    InfluxData id;
    alias id this;

    void opIndexAssign(bool value, string name)
    {
        id[name] = (value == true ? "True" : "False");
    }

    //Integer types need an additional i suffix
    void opIndexAssign(T)(T value, string name) if (isIntegral!T)
    {
        id[name] = value.to!string ~ "i";
    }

    void opIndexAssign(T)(T value, string name) if (!isIntegral!T)
    {
        data[name] = value.to!string;
    }

}

unittest
{
    auto v = InfluxValues();
    v["myInt"] = 4;
    v["myFloat"] = 0.2;
    v["myBool"] = true;
    string expected = "myBool=True,myFloat=0.2,myInt=4i";
    assert(v.toInflux() == expected, format("Error %s should be %s", v.toInflux(), expected));
}

//creates a measurement
// Measurement m = Measurement("mySeries");
// m["aBool"] = true;
// m.addTag("tagKey", "tagVal");
struct Measurement
{

    private
    {
        string series;
        ulong timestamp;
        InfluxData tags;

    }
    InfluxValues values;
    //for now directly export the values
    alias values this;

    this(string name)
    {
        series = name;
    }

    //adds a tag and its value;
    void addTag(string key, string val)
    {
        tags[key] = val;
    }

    string toInfluxdbLine()
    {
        //do not send timestamp if its not set
        string ts = (timestamp > 0) ? timestamp.to!string : "";
        //do not add comma if no tags are being sent
        string sep = (tags.length > 0) ? "," : "";
        return format("%s%s%s %s %s", series, sep, tags.toInflux(), values.toInflux(), ts);
    }
}