File Coverage

blib/lib/AnyEvent/InfluxDB.pm

Criterion	Covered	Total	%
statement	30	658	4.5
branch	0	318	0.0
condition	0	69	0.0
subroutine	10	91	10.9
pod	37	37	100.0
total	77	1173	6.5

#ABSTRACT: An asynchronous library for InfluxDB time-series database
use strict;
use warnings;
package AnyEvent::InfluxDB;
our $AUTHORITY = 'cpan:AJGB';
$AnyEvent::InfluxDB::VERSION = '0.13';
use AnyEvent;
use AnyEvent::HTTP;
use URI;
use URI::QueryParam;
use JSON qw(decode_json);
use List::MoreUtils qw(zip);
use URI::Encode::XS qw( uri_encode );
use Moo;

has [qw( ssl_options username password on_request )] => (
    is => 'ro',
    predicate => 1,
);

has 'server' => (
    is => 'rw',
    default => 'http://localhost:8086',
);

has '_is_ssl' => (
    is => 'lazy',
);

has '_tls_ctx' => (
    is => 'lazy',
);

has '_server_uri' => (
    is => 'lazy',
);


sub _build__tls_ctx {
    my ($self) = @_;

    # no ca/hostname checks
    return 'low' unless $self->has_ssl_options;

    # create ctx
    require AnyEvent::TLS;
    return AnyEvent::TLS->new( %{ $self->ssl_options } );
}

sub _build__is_ssl {
    my ($self) = @_;

    return $self->server =~ /^https/;
}

sub _build__server_uri {
    my ($self) = @_;

    my $url = URI->new( $self->server, 'http' );

    if ( $self->has_username && $self->has_password ) {
        $url->query_param( 'u' => $self->username );
        $url->query_param( 'p' => $self->password );
    }

    return $url;
}

sub _make_url {
    my ($self, $path, $params) = @_;

    my $url = $self->_server_uri->clone;
    $url->path($path);

    while ( my ($k, $v) = each %$params ) {
        $url->query_param( $k => $v );
    }

    return $url;
}

sub _http_request {
    my $cb = pop;
    my ($self, $method, $url, $post_data) = @_;

    if ($self->has_on_request) {
        $self->on_request->($method, $url, $post_data);
    }

    my %args = (
        headers => {
            referer => undef,
            'user-agent' => "AnyEvent-InfluxDB/0.13",
        }
    );
    if ( $method eq 'POST' ) {
        if ( defined $post_data ) {
            $args{'body'} = $post_data;
        } else {
            if ( my $q = $url->query_param_delete('q') ) {
                $args{headers}{'content-type'} = 'application/x-www-form-urlencoded';
                $args{body} = 'q='. uri_encode($q);
            }
        }
    }
    if ( $self->_is_ssl ) {
        $args{tls_ctx} = $self->_tls_ctx;
    }

    my $guard;
    $guard = http_request
        $method => $url->as_string,
        %args,
        sub {
            $cb->(@_);
            undef $guard;
        };
};


sub ping {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/ping', {
        (
            exists $args{wait_for_leader} ?
                ( wait_for_leader => $args{wait_for_leader} )
                :
                ()
        )
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;
            if ( $headers->{Status} eq '204' ) {
                $args{on_success}->( $headers->{'x-influxdb-version'} );
            } else {
                $args{on_error}->( $headers->{Reason} || $body );
            }
        }
    );
}



sub create_database {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'CREATE DATABASE '.
            (
                $args{if_not_exists} ? 'IF NOT EXISTS ' : ''
            )
            .$args{database};
    }

    my $url = $self->_make_url('/query', {
        q => $q,
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                $args{on_success}->( $body );
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub drop_database {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP DATABASE '
            .
            (
                $args{if_exists} ? 'IF EXISTS ' : ''
            ) . $args{database};
    }

    my $url = $self->_make_url('/query', {
        q => $q,
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub drop_series {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP SERIES';

        if ( my $measurement = $args{measurement} ) {
            $q .= ' FROM '. $measurement;
        }
        elsif ( my $measurements = $args{measurements} ) {
            $q .= ' FROM '. join(',', @{ $measurements || [] });
        }

        if ( my $cond = $args{where} ) {
            $q .= ' WHERE '. $cond;
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub delete_series {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DELETE';

        if ( my $measurement = $args{measurement} ) {
            $q .= ' FROM '. $measurement;
        }

        if ( my $cond = $args{where} ) {
            $q .= ' WHERE '. $cond;
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub drop_measurement {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP MEASUREMENT '. $args{measurement};
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_shards {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/query', {
        q => 'SHOW SHARDS'
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $shards = {};
                for my $res ( @{ $data->{results}->[0]->{series} || [] } ) {
                    my $cols = $res->{columns};
                    my $values = $res->{values};
                    $shards->{ $res->{name } } = [
                        map {
                            +{
                                zip(@$cols, @$_)
                            }
                        } @{ $values || [] }
                    ];
                }
                $args{on_success}->($shards);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_shard_groups {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/query', {
        q => 'SHOW SHARD GROUPS'
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $res = $data->{results}->[0]->{series}->[0];
                my $cols = $res->{columns};
                my $values = $res->{values};
                my @shard_groups = (
                    map {
                        +{
                            zip(@$cols, @$_)
                        }
                    } @{ $values || [] }
                );
                $args{on_success}->(@shard_groups);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub drop_shard {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP SHARD '. $args{id};
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_queries {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/query', {
        q => 'SHOW QUERIES'
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $res = $data->{results}->[0]->{series}->[0];
                my $cols = $res->{columns};
                my $values = $res->{values};
                my @queries = (
                    map {
                        +{
                            zip(@$cols, @$_)
                        }
                    } @{ $values || [] }
                );
                $args{on_success}->(@queries);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub kill_query {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'KILL QUERY '. $args{id};
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}



sub create_retention_policy {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'CREATE RETENTION POLICY '. $args{name}
            .' ON '. $args{database}
            .' DURATION '. $args{duration}
            .' REPLICATION '. $args{replication}
            .' SHARD DURATION '. $args{shard_duration};

        $q .= ' DEFAULT' if $args{default};
    }

    my $url = $self->_make_url('/query', {
        q => $q,
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub alter_retention_policy {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'ALTER RETENTION POLICY '. $args{name}
            .' ON '. $args{database};

        $q .= ' DURATION '. $args{duration} if exists $args{duration};
        $q .= ' SHARD DURATION '. $args{shard_duration} if exists $args{shard_duration};
        $q .= ' REPLICATION '. $args{replication} if exists $args{replication};;
        $q .= ' DEFAULT' if $args{default};
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub drop_retention_policy {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP RETENTION POLICY '. $args{name} .' ON '. $args{database};
    }

    my $url = $self->_make_url('/query', {
        q => $q,
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_databases {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/query', {
        q => 'SHOW DATABASES'
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my @names;
                eval {
                    @names = map { $_->[0] } @{ $data->{results}->[0]->{series}->[0]->{values} || [] };
                };
                $args{on_success}->(@names);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_retention_policies {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SHOW RETENTION POLICIES ON '. $args{database};
    }

    my $url = $self->_make_url('/query', {
        q => $q,
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $res = $data->{results}->[0]->{series}->[0];
                my $cols = $res->{columns};
                my $values = $res->{values};
                my @policies = (
                    map {
                        +{
                            zip(@$cols, @$_)
                        }
                    } @{ $values || [] }
                );
                $args{on_success}->(@policies);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_series {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SHOW SERIES';

        if ( my $measurement = $args{measurement} ) {
            $q .= ' FROM '. $measurement;
        }

        if ( my $cond = $args{where} ) {
            $q .= ' WHERE '. $cond;
        }

        if ( my $order_by = $args{order_by} ) {
            $q .= ' ORDER BY '. $order_by;
        }

        if ( my $limit = $args{limit} ) {
            $q .= ' LIMIT '. $limit;

            if ( my $offset = $args{offset} ) {
                $q .= ' OFFSET '. $offset;
            }
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $res = $data->{results}->[0]->{series}->[0];
                my $values = $res->{values};
                my @series = (
                    map { @$_ } @{ $values || [] }
                );
                $args{on_success}->(@series);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_measurements {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SHOW MEASUREMENTS';

        if ( my $measurement = $args{measurement} ) {
            $q .= ' WITH MEASUREMENT '
                . ( $measurement =~ /^\/.*\/$/ ? '=~' : '=' )
                . $measurement;
        }

        if ( my $cond = $args{where} ) {
            $q .= ' WHERE '. $cond;
        }

        if ( my $order_by = $args{order_by} ) {
            $q .= ' ORDER BY '. $order_by;
        }

        if ( my $limit = $args{limit} ) {
            $q .= ' LIMIT '. $limit;

            if ( my $offset = $args{offset} ) {
                $q .= ' OFFSET '. $offset;
            }
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $res = $data->{results}->[0]->{series}->[0];
                my $values = $res->{values};
                my @measurements = (
                    map { @$_ } @{ $values || [] }
                );
                $args{on_success}->(@measurements);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_tag_keys {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SHOW TAG KEYS';

        if ( my $measurement = $args{measurement} ) {
            $q .= ' FROM '. $measurement;
        }

        if ( my $cond = $args{where} ) {
            $q .= ' WHERE '. $cond;
        }

        if ( my $limit = $args{limit} ) {
            $q .= ' LIMIT '. $limit;
        }

        if ( my $offset = $args{offset} ) {
            $q .= ' OFFSET '. $offset;
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $tag_keys = {};
                for my $res ( @{ $data->{results}->[0]->{series} || [] } ) {
                    my $values = $res->{values};
                    $tag_keys->{ $res->{name } } = [
                        map {
                            @$_
                        } @{ $values || [] }
                    ];
                }
                $args{on_success}->($tag_keys);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_tag_values {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SHOW TAG VALUES';

        if ( my $measurement = $args{measurement} ) {
            $q .= ' FROM '. $measurement;
        }

        if ( my $keys = $args{keys} ) {
            $q .= ' WITH KEY IN ('. join(", ", @$keys) .')';
        }
        elsif ( my $key = $args{key} ) {
            $q .= ' WITH KEY = '. $key;
        }

        if ( my $cond = $args{where} ) {
            $q .= ' WHERE '. $cond;
        }

        if ( my $limit = $args{limit} ) {
            $q .= ' LIMIT '. $limit;
        }

        if ( my $offset = $args{offset} ) {
            $q .= ' OFFSET '. $offset;
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $tag_values = {};
                for my $res ( @{ $data->{results}->[0]->{series} || [] } ) {
                    my $cols = $res->{columns};
                    my %col_idx = ( key => 0, value => 1 );
                    for ( my $i = 0; $i < @{ $cols || [] }; $i++ ) {
                        $col_idx{ $cols->[$i] } = $i;
                    }
                    my $values = $res->{values};
                    for my $v ( @{ $values || [] } ) {
                        push @{ $tag_values->{ $res->{name } }->{ $v->[ $col_idx{key} ] } },
                            $v->[ $col_idx{value} ];
                    }
                }
                $args{on_success}->($tag_values);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_field_keys {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SHOW FIELD KEYS';

        if ( my $measurement = $args{measurement} ) {
            $q .= ' FROM '. $measurement;
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $field_keys = {};
                for my $res ( @{ $data->{results}->[0]->{series} || [] } ) {
                    my $values = $res->{values};
                    $field_keys->{ $res->{name } } = [
                        map {
                            @$_
                        } @{ $values || [] }
                    ];
                }
                $args{on_success}->($field_keys);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub create_user {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'CREATE USER '. $args{username}
            .' WITH PASSWORD \''. $args{password} .'\'';

        $q .= ' WITH ALL PRIVILEGES' if $args{all_privileges};
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub set_user_password {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SET PASSWORD FOR '. $args{username}
            .' = \''. $args{password} .'\'';
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_users {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/query', {
        q => 'SHOW USERS'
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $res = $data->{results}->[0]->{series}->[0];
                my $cols = $res->{columns};
                my $values = $res->{values};
                my @users = (
                    map {
                        +{
                            zip(@$cols, @$_)
                        }
                    } @{ $values || [] }
                );
                $args{on_success}->(@users);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub grant_privileges {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'GRANT ';

        if ( $args{all_privileges} ) {
            $q .= 'ALL PRIVILEGES';
        } else {
            $q .= $args{access} .' ON '. $args{database};
        }
        $q .= ' TO '. $args{username};
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_grants {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'SHOW GRANTS FOR '. $args{username};
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $res = $data->{results}->[0]->{series}->[0];
                my $cols = $res->{columns};
                my $values = $res->{values};
                my @grants = (
                    map {
                        +{
                            zip(@$cols, @$_)
                        }
                    } @{ $values || [] }
                );
                $args{on_success}->(@grants);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}



sub revoke_privileges {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'REVOKE ';

        if ( $args{all_privileges} ) {
            $q .= 'ALL PRIVILEGES';
        } else {
            $q .= $args{access} .' ON '. $args{database};
        }
        $q .= ' FROM '. $args{username};
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}



sub drop_user {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP USER '. $args{username};
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub create_continuous_query {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        my $resample = '';
        if ( $args{every} || $args{for} ) {
            $resample = ' RESAMPLE';
            if ( $args{every} ) {
                $resample .= ' EVERY '. $args{every};
            }
            if ( $args{for} ) {
                $resample .= ' FOR '. $args{for};
            }
        }
        $q = 'CREATE CONTINUOUS QUERY '. $args{name}
            .' ON '. $args{database}
            . $resample
            .' BEGIN '. $args{query}
            .' END';
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub drop_continuous_query {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP CONTINUOUS QUERY '. $args{name} . ' ON '. $args{database};
    }

    my $url = $self->_make_url('/query', {
        q => $q
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_continuous_queries {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => 'SHOW CONTINUOUS QUERIES'
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $cqs = {};
                for my $res ( @{ $data->{results}->[0]->{series} || [] } ) {
                    my $cols = $res->{columns};
                    my $values = $res->{values};
                    $cqs->{ $res->{name } } = [
                        map {
                            +{
                                zip(@$cols, @$_)
                            }
                        } @{ $values || [] }
                    ];
                }
                $args{on_success}->($cqs);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}



sub _to_line {
    my $data = shift;

    my $t = $data->{tags} || {};
    my $f = $data->{fields} || {};

    return $data->{measurement}
        .(
            $t ?
                    ','.
                    join(',',
                        map {
                            join('=', $_, $t->{$_})
                        } sort { $a cmp $b } keys %$t
                    )
                :
                ''
        )
        . ' '
        .(
            join(',',
                map {
                    join('=', $_, $f->{$_})
                } keys %$f
            )
        )
        .(
            $data->{time} ?
                ' '. $data->{time}
                :
                ''
        );
}

sub write {
    my ($self, %args) = @_;

    my $data = ref $args{data} eq 'ARRAY' ?
        join("\n", map { ref $_ eq 'HASH' ? _to_line($_) : $_ } @{ $args{data} })
        :
        ref $args{data} eq 'HASH' ? _to_line($args{data}) : $args{data};

    my $url = $self->_make_url('/write', {
        db => $args{database},
        (
            $args{consistency} ?
                ( consistency => $args{consistency} )
                :
                ()
        ),
        (
            $args{rp} ?
                ( rp => $args{rp} )
                :
                ()
        ),
        (
            $args{precision} ?
                ( precision => $args{precision} )
                :
                ()
        ),
        (
            $args{one} ?
                ( one => $args{one} )
                :
                ()
        )
    });

    $self->_http_request( POST => $url, $data,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '204' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub select {
    my ($self, %args) = @_;

    my $method = 'GET';
    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
        if ( $q =~ /\s+INTO\s+/i ) {
            $method = 'POST';
        }
    } else {
        $q = 'SELECT '. $args{fields};

        if ( my $into = $args{into} ) {
            $q .= ' INTO '. $into;
            $method = 'POST';
        }

        $q .= ' FROM '. $args{measurement};

        if ( my $cond = $args{where} ) {
            $q .= ' WHERE '. $cond;
        }

        if ( my $group = $args{group_by} ) {
            $q .= ' GROUP BY '. $group;

            if ( my $fill = $args{fill} ) {
                $q .= ' fill('. $fill .')';
            }
        }

        if ( my $order_by = $args{order_by} ) {
            $q .= ' ORDER BY '. $order_by;
        }

        if ( my $limit = $args{limit} ) {
            $q .= ' LIMIT '. $limit;

            if ( my $offset = $args{offset} ) {
                $q .= ' OFFSET '. $offset;
            }
        }

        if ( my $slimit = $args{slimit} ) {
            $q .= ' SLIMIT '. $slimit;

            if ( my $soffset = $args{soffset} ) {
                $q .= ' SOFFSET '. $soffset;
            }
        }
    }

    my $url = $self->_make_url('/query', {
        db => $args{database},
        q => $q,
        (
            $args{rp} ?
                ( rp => $args{rp} )
                :
                ()
        ),
        (
            $args{epoch} ?
                ( epoch => $args{epoch} )
                :
                ()
        ),
        (
            $args{chunk_size} ?
                ( chunk_size => $args{chunk_size} )
                :
                ()
        ),
    });

    $self->_http_request( $method => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $series = [
                    map {
                        my $res = $_;

                        my $cols = $res->{columns};
                        my $values = $res->{values};

                        +{
                            name => $res->{name},
                            values => [
                                map {
                                    +{
                                        zip(@$cols, @$_)
                                    }
                                } @{ $values || [] }
                            ]
                        }
                    } @{ $data->{results}->[0]->{series} || [] }
                ];
                $args{on_success}->($series);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub query {
    my ($self, %args) = @_;

    my $url = $self->_server_uri->clone;
    $url->path('/query');
    $url->query_form_hash( $args{query} );

    my $method = $args{method} || 'GET';

    $self->_http_request( $method => $url,
        sub {
            $args{on_response}->(@_);
        }
    );
}


sub create_subscription {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'CREATE SUBSCRIPTION '. $args{name} .' ON '
            . $args{database} .'.'. $args{rp}
            . ' DESTINATIONS '. $args{mode} .' '
            . (
                ref $args{destinations} eq 'ARRAY' ?
                    join(", ", @{ $args{destinations} || [] } )
                    :
                    $args{destinations}
            );
    }

    my $url = $self->_make_url('/query', {
        q => $q,
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub show_subscriptions {
    my ($self, %args) = @_;

    my $url = $self->_make_url('/query', {
        q => 'SHOW SUBSCRIPTIONS'
    });

    $self->_http_request( GET => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' ) {
                my $data = decode_json($body);
                my $subscriptions = {};
                for my $res ( @{ $data->{results}->[0]->{series} || [] } ) {
                    my $cols = $res->{columns};
                    my $values = $res->{values};
                    $subscriptions->{ $res->{name } } = [
                        map {
                            +{
                                zip(@$cols, @$_)
                            }
                        } @{ $values || [] }
                    ];
                }
                $args{on_success}->($subscriptions);
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}


sub drop_subscription {
    my ($self, %args) = @_;

    my $q;
    if ( exists $args{q} ) {
        $q = $args{q};
    } else {
        $q = 'DROP SUBSCRIPTION '. $args{name} .' ON '
            . $args{database} .'.'. $args{rp};
    }

    my $url = $self->_make_url('/query', {
        q => $q,
    });

    $self->_http_request( POST => $url,
        sub {
            my ($body, $headers) = @_;

            if ( $headers->{Status} eq '200' && $body eq '{"results":[{}]}' ) {
                $args{on_success}->();
            } else {
                $args{on_error}->( $body );
            }
        }
    );
}



1;

__END__