File Coverage

blib/lib/MojoX/Redis.pm

Criterion	Covered	Total	%
statement	107	130	82.3
branch	17	32	53.1
condition	11	20	55.0
subroutine	20	22	90.9
pod	3	5	60.0
total	158	209	75.6

package MojoX::Redis;

use strict;
use warnings;

our $VERSION = 0.87;
use base 'Mojo::Base';

use Mojo::IOLoop;
use List::Util   ();
use Scalar::Util ();
use Encode       ();
require Carp;

__PACKAGE__->attr(server   => '127.0.0.1:6379');
__PACKAGE__->attr(ioloop   => sub { Mojo::IOLoop->singleton });
__PACKAGE__->attr(error    => undef);
__PACKAGE__->attr(timeout  => 300);
__PACKAGE__->attr(encoding => 'UTF-8');
__PACKAGE__->attr(
    on_error => sub {
        sub {
            my $redis = shift;
            warn "Redis error: ", $redis->error, "\n";
          }
    }
);

__PACKAGE__->attr(
    protocol_redis => sub {
        require Protocol::Redis;
        "Protocol::Redis";
    }
);

our @COMMANDS = qw/
  append auth bgrewriteaof bgsave blpop brpop brpoplpush config_get config_set
  config_resetstat dbsize debug_object debug_segfault decr decrby del discard
  echo exec exists expire expireat flushall flushdb get getbit getrange getset
  hdel hexists hget hgetall hincrby hkeys hlen hmget hmset hset hsetnx hvals
  incr incrby info keys lastsave lindex linsert llen lpop lpush lpushx lrange
  lrem lset ltrim mget monitor move mset msetnx multi persist ping psubscribe
  publish punsubscribe quit randomkey rename renamenx rpop rpoplpush rpush
  rpushx sadd save scard sdiff sdiffstore select set setbit setex setnx
  setrange shutdown sinter sinterstore sismember slaveof smembers smove sort
  spop srandmember srem strlen subscribe sunion sunionstore sync ttl type
  unsubscribe unwatch watch zadd zcard zcount zincrby zinterstore zrange
  zrangebyscore zrank zrem zremrangebyrank zremrangebyscore zrevrange
  zrevrangebyscore zrevrank zscore zunionstore
  /;

sub AUTOLOAD {
    my ($package, $cmd) = our $AUTOLOAD =~ /^([\w\:]+)\:\:(\w+)$/;

    Carp::croak(qq|Can't locate object method "$cmd" via "$package"|)
      unless List::Util::first { $_ eq $cmd } @COMMANDS;

    my $self = shift;

    my $args = [@_];
    my $cb   = $args->[-1];
    if (ref $cb ne 'CODE') {
        $cb = undef;
    }
    else {
        pop @$args;
    }

    $self->execute($cmd, $args, $cb);
}

sub DESTROY {
    my $self = shift;

    # Loop
    return unless my $loop = $self->ioloop;

    # Cleanup connection
    $loop->remove($self->{_connection})
      if $self->{_connection};
}

sub connect {
    my $self = shift;

    # drop old connection
    if ($self->connected) {
        $self->ioloop->remove($self->{_connection});
    }

    $self->server =~ m{^([^:]+)(:(\d+))?};
    my $address = $1;
    my $port = $3 || 6379;

    Scalar::Util::weaken $self;

    $self->{_protocol} = $self->_create_protocol;

    # connect
    $self->{_connecting} = 1;
    $self->{_connection} = $self->ioloop->client(
        {   address => $address,
            port    => $port
        },
        sub {
            my ($loop, $err, $stream) = @_;


            delete $self->{_connecting};
            $stream->timeout($self->timeout);
            $self->_send_next_message;

            $stream->on(
                read => sub {
                    my ($stream, $chunk) = @_;
                    $self->{_protocol}->parse($chunk);
                }
            );
            $stream->on(
                close => sub {
                    my $str = shift;
                    $self->{error} ||= 'disconnected';
                    $self->_inform_queue;

                    delete $self->{_message_queue};

                    delete $self->{_connecting};
                    delete $self->{_connection};
                }
            );
            $stream->on(
                error => sub {
                    my ($str, $error) = @_;
                    $self->error($error);
                    $self->_inform_queue;

                    $self->on_error->($self);
                    $self->ioloop->remove($self->{_connection});
                }
            );

        }
    );

    return $self;
}

sub connected {
    my $self = shift;

    return $self->{_connection};
}

sub execute {
    my ($self, $command, $args, $cb) = @_;

    if (!$cb && ref $args eq 'CODE') {
        $cb   = $args;
        $args = [];
    }
    elsif (!ref $args) {
        $args = [$args];
    }

    unshift @$args, uc $command;

    my $mqueue = $self->{_message_queue} ||= [];
    my $cqueue = $self->{_cb_queue}      ||= [];


    push @$mqueue, $args;
    push @$cqueue, $cb;

    $self->connect unless $self->{_connection};
    $self->_send_next_message;

    return $self;
}

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

    $self->ioloop->start;
    return $self;
}

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

    $self->ioloop->stop;
    return $self;
}

sub _create_protocol {
    my $self = shift;

    my $protocol = $self->protocol_redis->new(api => 1);
    $protocol->on_message(
        sub {
            my ($parser, $command) = @_;
            $self->_return_command_data($command);
        }
    );

    Carp::croak(q/Protocol::Redis implementation doesn't support APIv1/)
      unless $protocol;

    $protocol;
}

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

    if ((my $id = $self->{_connection}) && !$self->{_connecting}) {
        while (my $args = shift @{$self->{_message_queue}}) {
            my $cmd_arg = [];
            my $cmd = {type => '*', data => $cmd_arg};
            foreach my $token (@$args) {
                $token = Encode::encode($self->encoding, $token)
                  if $self->encoding;
                push @$cmd_arg, {type => '$', data => $token};
            }
            my $message = $self->{_protocol}->encode($cmd);

            $self->ioloop->stream($id)->write($message);
        }
    }
}


sub _reencode_message {
    my ($self, $message) = @_;

    my ($type, $data) = @{$message}{'type', 'data'};

    # Decode data
    if ($type ne '*' && $self->encoding && $data) {
        $data = Encode::decode($self->encoding, $data);
    }

    if ($type eq '-') {
        $self->error($data);
        $self->on_error->($self);
        return;
    }
    elsif ($type ne '*') {
        return [$data];
    }
    else {
        my $reencoded_data = [];
        foreach my $item (@$data) {
            my $message = $self->_reencode_message($item);
            push @$reencoded_data, $message;
        }
        return $reencoded_data;
    }
}

sub _return_command_data {
    my ($self, $message) = @_;

    my $data = $self->_reencode_message($message);

    my $cb = shift @{$self->{_cb_queue}};
    $cb->($self, $data) if $cb;

    # Reset error after callback dispatching
    $self->error(undef);
}


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

    for my $cb (@{$self->{_cb_queue}}) {
        $cb->($self) if $cb;
    }
    $self->{_queue} = [];
}

1;
__END__