File Coverage

blib/lib/EV/ClickHouse.pm

Criterion	Covered	Total	%
statement	14	89	15.7
branch	0	70	0.0
condition	0	74	0.0
subroutine	5	8	62.5
pod	1	1	100.0
total	20	242	8.2

package EV::ClickHouse;
use strict;
use warnings;

use EV;

BEGIN {
    our $VERSION = '0.02';
    use XSLoader;
    XSLoader::load __PACKAGE__, $VERSION;
}

*q          = \&query;
*reconnect  = \&reset;
*disconnect = \&finish;

sub _uri_unescape { my $s = $_[0]; $s =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/ge; $s }

sub new {
    my ($class, %args) = @_;

    # Connection URI: clickhouse://user:pass@host:port/database
    # host accepts a bracketed IPv6 literal (e.g. clickhouse://[::1]:9000/db)
    if (my $uri = delete $args{uri}) {
        if ($uri =~ m{^clickhouse(?:\+(\w+))?://(?:([^:@]*?)(?::([^@]*))?\@)?(\[[^\]]+\]|[^/:?]+)(?::(\d+))?(?:/([^?]*))?(?:\?(.*))?$}) {
            my ($proto, $u, $pw, $h, $p, $db, $qs) = ($1, $2, $3, $4, $5, $6, $7);
            $h =~ s/^\[(.*)\]$/$1/;
            $args{protocol} //= $proto if $proto;
            $args{user}     //= _uri_unescape($u)  if defined $u && $u ne '';
            $args{password} //= _uri_unescape($pw) if defined $pw;
            $args{host}     //= $h;
            $args{port}     //= $p     if defined $p;
            $args{database} //= _uri_unescape($db) if defined $db && $db ne '';
            if (defined $qs) {
                for my $pair (split /&/, $qs) {
                    my ($k, $v) = split /=/, $pair, 2;
                    $args{$k} //= _uri_unescape($v) if defined $k && defined $v;
                }
            }
        } else {
            die "EV::ClickHouse: invalid URI '$uri'\n";
        }
    }

    my $loop = delete $args{loop} || EV::default_loop;
    my $self = $class->_new($loop);

    $self->on_error(exists $args{on_error} ? delete $args{on_error} : sub { die @_ });
    $self->on_connect(delete $args{on_connect})      if exists $args{on_connect};
    $self->on_progress(delete $args{on_progress})    if exists $args{on_progress};
    $self->on_disconnect(delete $args{on_disconnect}) if exists $args{on_disconnect};
    $self->on_trace(delete $args{on_trace})          if exists $args{on_trace};

    my $host     = delete $args{host}     // '127.0.0.1';
    my $port     = delete $args{port};
    my $protocol = delete $args{protocol} // 'http';
    my $user     = delete $args{user}     // 'default';
    my $password = delete $args{password} // '';
    my $db_alias = delete $args{db};
    my $database = delete $args{database} // $db_alias // 'default';
    my $tls      = delete $args{tls}      // 0;
    my $tls_ca_file    = delete $args{tls_ca_file};
    my $tls_skip_verify = delete $args{tls_skip_verify} // 0;

    # options
    my $compress        = delete $args{compress}        // 0;
    my $session_id      = delete $args{session_id};
    my $connect_timeout = delete $args{connect_timeout};
    my $query_timeout   = delete $args{query_timeout};
    my $auto_reconnect    = delete $args{auto_reconnect}    // 0;
    my $keepalive         = delete $args{keepalive}          // 0;
    my $reconnect_delay   = delete $args{reconnect_delay}    // 0;
    my $reconnect_max_delay = delete $args{reconnect_max_delay} // 0;

    # decode options (native protocol)
    my $decode_datetime = delete $args{decode_datetime}  // 0;
    my $decode_decimal  = delete $args{decode_decimal}   // 0;
    my $decode_enum     = delete $args{decode_enum}      // 0;
    my $named_rows      = delete $args{named_rows}       // 0;

    die "EV::ClickHouse: unknown protocol '$protocol' (expected 'http' or 'native')\n"
        unless $protocol eq 'http' || $protocol eq 'native';

    $port //= ($protocol eq 'native') ? 9000 : 8123;

    $self->_set_protocol($protocol eq 'native' ? 1 : 0);
    $self->_set_compress($compress)            if $compress;
    $self->_set_session_id($session_id)        if defined $session_id;
    $self->_set_connect_timeout($connect_timeout) if $connect_timeout;
    $self->_set_query_timeout($query_timeout)  if $query_timeout;
    $self->_set_tls($tls)                      if $tls;
    $self->_set_tls_ca_file($tls_ca_file)      if defined $tls_ca_file;
    $self->_set_tls_skip_verify($tls_skip_verify) if $tls_skip_verify;
    $self->_set_auto_reconnect($auto_reconnect) if $auto_reconnect;
    $self->_set_keepalive($keepalive)          if $keepalive;
    $self->_set_reconnect_delay($reconnect_delay) if $reconnect_delay;
    $self->_set_reconnect_max_delay($reconnect_max_delay) if $reconnect_max_delay;

    # compute decode_flags bitmask
    my $decode_flags = 0;
    $decode_flags |= 1 if $decode_datetime;  # DECODE_DT_STR
    $decode_flags |= 2 if $decode_decimal;   # DECODE_DEC_SCALE
    $decode_flags |= 4 if $decode_enum;      # DECODE_ENUM_STR
    $decode_flags |= 8 if $named_rows;       # DECODE_NAMED_ROWS
    $self->_set_decode_flags($decode_flags)   if $decode_flags;

    my $settings = delete $args{settings};
    $self->_set_settings($settings)            if $settings;

    warn "EV::ClickHouse->new: unknown parameter(s): " . join(', ', sort keys %args) . "\n"
        if %args;

    $self->connect($host, $port, $user, $password, $database);

    $self;
}

1;

__END__