File Coverage

blib/lib/Protocol/Redis.pm

Criterion	Covered	Total	%
statement	76	79	96.2
branch	29	34	85.2
condition	4	4	100.0
subroutine	10	10	100.0
pod	6	6	100.0
total	125	133	93.9

package Protocol::Redis;

use strict;
use warnings;
use 5.008_001;

our $VERSION = '1.0011';

require Carp;

sub new {
    my $class = shift;
    $class = ref $class if ref $class;

    my $self = {@_};

    return unless $self->{api} == '1';

    bless $self, $class;

    $self->on_message(delete $self->{on_message});
    $self->{_messages} = [];

    $self;
}

sub api {
    my $self = shift;

    $self->{api};
}

my %simple_types = ('+' => 1, '-' => 1, ':' => 1);
my $rn = "\r\n";

sub encode {
    my $self = shift;

    my $encoded_message = '';
    while (@_) {
        my $message = shift;

        # Bulk string
        if ($message->{type} eq '$') {
            if (defined $message->{data}) {
                $encoded_message .= '$' . length($message->{data}) . "\r\n" . $message->{data} . "\r\n";
            }
            else {
                $encoded_message .= "\$-1\r\n";
            }
        }
        # Array (multi bulk)
        elsif ($message->{type} eq '*') {
            if (defined $message->{data}) {
                $encoded_message .= '*' . scalar(@{$message->{data}}) . "\r\n";
                unshift @_, @{$message->{data}};
            }
            else {
                $encoded_message .= "*-1\r\n";
            }
        }
        # String, error, integer
        elsif (exists $simple_types{$message->{type}}) {
            $encoded_message .= $message->{type} . $message->{data} . "\r\n";
        }
        else {
            Carp::croak(qq/Unknown message type $message->{type}/);
        }
    }

    return $encoded_message;
}

sub get_message {
    shift @{$_[0]->{_messages}};
}

sub on_message {
    my ($self, $cb) = @_;
    $self->{_on_message_cb} = $cb || \&_gather_messages;
}

sub parse {
    my $self        = shift;
    $self->{_buffer}.= shift;

    my $message = $self->{_message} ||= {};
    my $buffer  = \$self->{_buffer};

    CHUNK:
    while ((my $pos = index($$buffer, "\r\n")) != -1) {
        # Check our state: are we parsing new message or completing existing
        if (!$message->{type}) {
            if ($pos < 1) {
                Carp::croak(qq/Unexpected input "$$buffer"/);
            }

            $message->{type} = substr $$buffer, 0, 1;
            $message->{_argument}  = substr $$buffer, 1, $pos - 1;
            substr $$buffer, 0, $pos + 2, ''; # Remove type + argument + \r\n
        }

        # Simple Strings, Errors, Integers
        if (exists $simple_types{$message->{type}}) {
            $message->{data} = delete $message->{_argument};
        }
        # Bulk Strings
        elsif ($message->{type} eq '$') {
            if ($message->{_argument} eq '-1') {
                $message->{data} = undef;
            }
            elsif (length($$buffer) >= $message->{_argument} + 2) {
                $message->{data} = substr $$buffer, 0, $message->{_argument}, '';
                substr $$buffer, 0, 2, ''; # Remove \r\n
            }
            else {
                return # Wait more data
            }
        }
        # Arrays
        elsif ($message->{type} eq '*') {
            if ($message->{_argument} eq '-1') {
                $message->{data} = undef;
            } else {
                $message->{data} = [];
                if ($message->{_argument} > 0) {
                    $message = $self->{_message} = {_parent => $message};
                    next;
                }
            }
        }
        # Invalid input
        else {
            Carp::croak(qq/Unexpected input "$self->{_message}{type}"/);
        }

        delete $message->{_argument};
        delete $self->{_message};

        # Fill parents with data
        while (my $parent = delete $message->{_parent}) {
            push @{$parent->{data}}, $message;

            if (@{$parent->{data}} < $parent->{_argument}) {
                $message = $self->{_message} = {_parent => $parent};
                next CHUNK;
            }
            else {
                $message = $parent;
                delete $parent->{_argument};
            }
        }

        $self->{_on_message_cb}->($self, $message);
        $message = $self->{_message} = {};
    }
}

sub _gather_messages {
    push @{$_[0]->{_messages}}, $_[1];
}

1;
__END__