File Coverage

blib/lib/BusyBird/Timeline.pm

Criterion	Covered	Total	%
statement	169	173	97.6
branch	36	42	85.7
condition	20	27	74.0
subroutine	42	42	100.0
pod	15	15	100.0
total	282	299	94.3

package BusyBird::Timeline;
use v5.8.0;
use strict;
use warnings;
use BusyBird::Util qw(set_param);
use BusyBird::Log qw(bblog);
use BusyBird::Flow;
use BusyBird::Watcher::Aggregator;
use BusyBird::DateTime::Format 0.04;
use BusyBird::Config;
use Async::Selector 1.0;
use Data::UUID;
use Carp;
use Storable qw(dclone);
use Scalar::Util qw(weaken looks_like_number);
use DateTime;

our @CARP_NOT = qw(BusyBird::Config);

sub new {
    my ($class, %args) = @_;
    my $self = bless {
        filter_flow => BusyBird::Flow->new,
        selector => Async::Selector->new,
        unacked_counts => {total => 0},
        config => BusyBird::Config->new(type => "timeline", with_default => 0),
        id_generator => Data::UUID->new,
    }, $class;
    $self->set_param(\%args, 'name', undef, 1);
    $self->set_param(\%args, 'storage', undef, 1);
    $self->set_param(\%args, 'watcher_max', 512);
    croak 'name must not be empty' if $self->{name} eq '';
    $self->_init_selector();
    $self->_update_unacked_counts();
    return $self;
}

sub _log {
    my ($self, $level, $msg) = @_;
    bblog($level, $self->name . ": $msg");
}

sub _update_unacked_counts {
    my ($self) = @_;
    $self->get_unacked_counts(callback => sub {
        my ($error, $unacked_counts) = @_;
        if(defined($error)) {
            $self->_log('error', "error while updating unacked count: $error");
            return;
        }
        $self->{unacked_counts} = $unacked_counts;
        $self->{selector}->trigger('unacked_counts');
    });
}

sub _init_selector {
    my ($self) = @_;
    weaken $self;
    $self->{selector}->register(unacked_counts => sub {
        my ($exp_unacked_counts) = @_;
        if(!defined($exp_unacked_counts) || ref($exp_unacked_counts) ne 'HASH') {
            croak "unacked_counts watcher: condition input must be a hash-ref";
        }
        return { %{$self->{unacked_counts}} } if !%$exp_unacked_counts;
        foreach my $key (keys %$exp_unacked_counts) {
            my $exp_val = $exp_unacked_counts->{$key} || 0;
            my $got_val = $self->{unacked_counts}{$key} || 0;
            return { %{$self->{unacked_counts}} } if $exp_val != $got_val;
        }
        return undef;
    });
    $self->{selector}->register(watcher_quota => sub {
        my ($in) = @_;
        my @watchers = $self->{selector}->watchers('watcher_quota');
        if(int(@watchers) <= $self->{watcher_max}) {
            return undef;
        }
        my $watcher_age = $in->{age} || 0;
        return $watcher_age > $self->{watcher_max} ? 1 : undef;
    });
}

sub name {
    return shift->{name};
}

sub _get_from_storage {
    my ($self, $method, $args_ref) = @_;
    $args_ref->{timeline} = $self->name;
    local @CARP_NOT = (ref($self->{storage}));
    $self->{storage}->$method(%$args_ref);
}

sub get_statuses {
    my ($self, %args) = @_;
    $self->_get_from_storage("get_statuses", \%args);
}

sub get_unacked_counts {
    my ($self, %args) = @_;
    $self->_get_from_storage("get_unacked_counts", \%args);
}

sub _write_statuses {
    my ($self, $method, $args_ref) = @_;
    $args_ref->{timeline} = $self->name;
    local @CARP_NOT = (ref($self->{storage}));
    my $orig_callback = $args_ref->{callback};
    $self->{storage}->$method(%$args_ref, callback => sub {
        $self->_update_unacked_counts();
        goto $orig_callback if defined($orig_callback);
    });
}

sub put_statuses {
    my ($self, %args) = @_;
    $self->_write_statuses('put_statuses', \%args);
}

sub delete_statuses {
    my ($self, %args) = @_;
    $self->_write_statuses('delete_statuses', \%args);
}

sub ack_statuses {
    my ($self, %args) = @_;
    $self->_write_statuses('ack_statuses', \%args);
}

sub add_statuses {
    my ($self, %args) = @_;
    if(!defined($args{statuses})) {
        croak "statuses argument is mandatory";
    }
    my $ref = ref($args{statuses});
    if($ref eq "HASH") {
        $args{statuses} = [ $args{statuses} ];
    }elsif($ref ne "ARRAY") {
        croak "statuses argument must be a status or an array-ref of statuses";
    }
    my $statuses = dclone($args{statuses});
    my $final_callback = $args{callback};
    $self->{filter_flow}->execute($statuses, sub {
        my $filter_result = shift;
        my $cur_time;
        foreach my $status (@$filter_result) {
            next if !defined($status) || ref($status) ne 'HASH';
            if(!defined($status->{id})) {
                $status->{id} = sprintf('busybird://%s/%s', $self->name, $self->{id_generator}->create_str);
            }
            if(!defined($status->{created_at})) {
                $cur_time ||= DateTime->now;
                $status->{created_at} = BusyBird::DateTime::Format->format_datetime($cur_time);
            }
        }
        $self->_write_statuses('put_statuses', {
            mode => 'insert', statuses => $filter_result,
            callback => $final_callback
        });
    });
}

sub add {
    my ($self, $statuses, $callback) = @_;
    $self->add_statuses(statuses => $statuses, callback => $callback);
}

sub contains {
    my ($self, %args) = @_;
    $self->_get_from_storage("contains", \%args);
}

sub add_filter {
    my ($self, $filter, $is_async) = @_;
    if(!$is_async) {
        my $sync_filter = $filter;
        $filter = sub {
            my ($statuses, $done) = @_;
            @_ = $sync_filter->($statuses);
            goto $done;
        };
    }
    $self->{filter_flow}->add($filter);
}

sub add_filter_async {
    my ($self, $filter) = @_;
    $self->add_filter($filter, 1);
}

sub set_config {
    shift()->{config}->set_config(@_);
}

sub get_config {
    shift()->{config}->get_config(@_);
}

sub watch_unacked_counts {
    my ($self, %watch_args) = @_;
    my $callback = $watch_args{callback};
    my $assumed = $watch_args{assumed};
    if(!defined($callback) || ref($callback) ne 'CODE') {
        croak "watch_unacked_counts: callback must be a code-ref";
    }
    if(!defined($assumed) || ref($assumed) ne 'HASH') {
        croak "watch_unacked_counts: assumed must be a hash-ref";
    }
    $assumed = +{ %$assumed };
    foreach my $key (keys %$assumed) {
        next if $key eq 'total' || (looks_like_number($key) && int($key) == $key);
        delete $assumed->{$key};
    }
    my $watcher = BusyBird::Watcher::Aggregator->new();
    my $orig_watcher = $self->{selector}->watch(
        unacked_counts => $assumed, watcher_quota => { age => 0 }, sub {
            my ($orig_w, %res) = @_;
            if($res{watcher_quota}) {
                $watcher->cancel();
                $callback->("watcher cancelled because it is too old", $watcher);
                return;
            }
            if($res{unacked_counts}) {
                $callback->(undef, $watcher, $res{unacked_counts});
                return;
            }
            confess("Something terrible happened.");
        }
    );
    $watcher->add($orig_watcher);
    if($watcher->active) {
        my @quota_watchers = $self->{selector}->watchers('watcher_quota');
        foreach my $w (@quota_watchers) {
            my %cond = $w->conditions;
            $cond{watcher_quota}{age}++;
        }
        $self->{selector}->trigger('watcher_quota');
    }
    return $watcher;
}


1;

__END__