File Coverage

blib/lib/Protocol/Redis.pm

Criterion	Covered	Total	%
statement	118	120	98.3
branch	23	28	82.1
condition			n/a
subroutine	22	22	100.0
pod	6	6	100.0
total	169	176	96.0

package Protocol::Redis;

use strict;
use warnings;
use 5.008_001;

our $VERSION = 1.0006;

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->{_state} = \&_state_new_message;

    $self;
}

sub api {
    my $self = shift;

    $self->{api};
}

my %message_type_encoders = (
    '+' => \&_encode_string,
    '-' => \&_encode_string,
    ':' => \&_encode_string,
    '$' => \&_encode_bulk,
    '*' => \&_encode_multi_bulk,
);

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

    if (my $encoder = $message_type_encoders{$message->{type}}) {
        $encoder->($self, $message);
    }
    else {
        Carp::croak(qq/Unknown message type $message->{type}/);
    }
}

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

    $message->{type} . $message->{data} . "\r\n";
}

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

    my $data = $message->{data};

    return '$-1' . "\r\n"
      unless defined $data;

    '$' . length($data) . "\r\n" . $data . "\r\n";
}

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

    my $data = $message->{data};

    return '*-1' . "\r\n"
      unless defined $data;

    my $e_message = '*' . scalar(@$data) . "\r\n";
    foreach my $element (@$data) {
        $e_message .= $self->encode($element);
    }

    $e_message;
}


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

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

sub parse {
    my ($self, $chunk) = @_;

    # Pass chunk to current vertex.
    # Some vertices can return unparsed chunk. In this case
    # cycle will pass chunk to next vertex.
    1 while $chunk = $self->{_state}->($self, $chunk);
}

sub _message_parsed {
    my ($self, $chunk) = @_;

    my $message = delete $self->{_cmd};

    if (my $cb = $self->{_on_message_cb}) {
        $cb->($self, $message);
    }
    else {
        push @{$self->{_messages}}, $message;
    }

    $self->{_state} = \&_state_new_message;
    $chunk;
}

my %message_type_parsers = (
    '+' => \&_state_string_message,
    '-' => \&_state_string_message,
    ':' => \&_state_string_message,
    '$' => \&_state_bulk_message,
    '*' => \&_state_multibulk_message,
);

sub _state_parse_message_type {
    my ($self, $chunk) = @_;

    my $cmd = substr $chunk, 0, 1, '';

    if ($cmd) {
        if (my $parser = $message_type_parsers{$cmd}) {
            $self->{_cmd}{type} = $cmd;
            $self->{_state} = $parser;
            return $chunk;
        }

        Carp::croak(qq/Unexpected input "$cmd"/);
    }
}

sub _state_new_message {
    my ($self, $chunk) = @_;

    $self->{_cmd} = {type => undef, data => undef};

    $self->{_state_cb} = \&_message_parsed;

    $self->{_state} = \&_state_parse_message_type;
    $chunk;
}

sub _state_string_message {
    my ($self, $chunk) = @_;

    my $str = $self->{_state_string} .= $chunk;
    my $i = index $str, "\r\n";

    # string isn't full
    return if $i < 0;

    # We got full string
    $self->{_cmd}{data} = substr $str, 0, $i, '';

    # Delete newline
    substr $str, 0, 2, '';

    delete $self->{_state_string};

    $self->{_state_cb}->($self, $str);
}

sub _state_bulk_message {
    my ($self, $chunk) = @_;

    my $bulk_state_cb = $self->{_state_cb};

    # Read bulk message size
    $self->{_state_cb} = sub {
        my ($self, $chunk) = @_;

        $self->{_bulk_size} = delete $self->{_cmd}{data};

        if ($self->{_bulk_size} == -1) {

            # Nil
            $self->{_cmd}{data} = undef;
            $bulk_state_cb->($self, $chunk);
        }
        else {
            $self->{_state_cb} = $bulk_state_cb;
            $self->{_state}    = \&_state_bulk_message_data;
            $chunk;
        }
    };
    $self->{_state} = \&_state_string_message;
    $chunk;
}

sub _state_bulk_message_data {
    my ($self, $chunk) = @_;

    my $str = $self->{_state_string} .= $chunk;

    # String + newline parsed
    return unless length $str >= $self->{_bulk_size} + 2;

    $self->{_cmd}{data} = substr $str, 0, $self->{_bulk_size}, '';

    # Delete ending newline
    substr $str, 0, 2, '';

    delete $self->{_state_string};
    delete $self->{_bulk_size};

    $self->{_state_cb}->($self, $str);
}

sub _state_multibulk_message {
    my ($self, $chunk) = @_;

    my $mbulk_state_cb = delete $self->{_state_cb};
    my $data           = [];
    my $mbulk_process;

    my $arguments_num;

    $mbulk_process = sub {
        my ($self, $chunk) = @_;

        push @$data,
          { type => delete $self->{_cmd}{type},
            data => delete $self->{_cmd}{data}
          };

        if (scalar @$data == $arguments_num) {

            # Cleanup
            $mbulk_process = undef;
            delete $self->{_state_cb};

            # Return message
            $self->{_cmd}{type} = '*';
            $self->{_cmd}{data} = $data;
            $mbulk_state_cb->($self, $chunk);
        }
        else {

            # read next string
            $self->{_state_cb} = $mbulk_process;
            $self->{_state}    = \&_state_parse_message_type;
            $chunk;
        }
    };

    $self->{_state_cb} = sub {
        my ($self, $chunk) = @_;

        # Number of Multi-Bulk message
        $arguments_num = delete $self->{_cmd}{data};
        if ($arguments_num < 1) {
            $mbulk_process = undef;
            $self->{_cmd}{data} = $arguments_num == 0 ? [] : undef;
            $mbulk_state_cb->($self, $chunk);
        }
        else {

            # We got messages
            $self->{_state_cb} = $mbulk_process;
            $self->{_state}    = \&_state_parse_message_type;
            $chunk;
        }
    };

    # Get number of messages
    $self->{_state} = \&_state_string_message;
    $chunk;
}

1;
__END__