File Coverage

blib/lib/AnyEvent/SlackRTM.pm

Criterion	Covered	Total	%
statement	30	117	25.6
branch	2	28	7.1
condition	2	14	14.2
subroutine	10	33	30.3
pod	11	11	100.0
total	55	203	27.0

package AnyEvent::SlackRTM;
$AnyEvent::SlackRTM::VERSION = '1.3';
use v5.14;

# ABSTRACT: AnyEvent module for interacting with the Slack RTM API

use AnyEvent;
use AnyEvent::WebSocket::Client 0.12;
use Carp;
use Furl;
use JSON;
use Try::Tiny;

our $START_URL = 'https://slack.com/api/rtm.connect';


sub new {
    my ($class, $token, $client_opts) = @_;

    $client_opts //= {};
    croak "Client options must be passed as a HashRef" unless ref $client_opts eq 'HASH';

    my $client;
    try {
        $client = AnyEvent::WebSocket::Client->new(%$client_opts);
    } catch {
        croak "Can't create client object: $_";
    };

    return bless {
        token    => $token,
        client   => $client,
        registry => {},
    }, $class;
}


sub start {
    my $self = shift;

    use vars qw( $VERSION );
    $VERSION //= '*-devel';

    my $furl = Furl->new(
        agent   => "AnyEvent::SlackRTM/$VERSION",
        timeout => $self->{client}->timeout,
    );

    my $res = $furl->get($START_URL . '?token=' . $self->{token});
    my $start = try {
        decode_json($res->content);
    }
    catch {
        my $status = $res->status;
        my $message = $res->content;
        croak "unable to start, Slack call failed: $status $message";
    };

    my $ok  = $start->{ok};
    croak "unable to start, Slack returned an error: $start->{error}"
    unless $ok;

    # Store this stuff in case we want it
    $self->{metadata} = $start;

    # We've now asked to re-open the connection,
    # so don't close again on timeout.
    delete $self->{closed};

    $self->{client}->connect( $start->{url} )->cb( sub {
        my $client = shift;

        delete $self->{finished};
        $self->{started}++;
        $self->{id} = 1;

        my $conn = $self->{conn} = $client->recv;
        $conn->on( each_message => sub { $self->_handle_incoming(@_) } );
        $conn->on( finish       => sub { $self->_handle_finish(@_) } );

        my $started = localtime;
        $self->{_last_keep_alive} = time;
        $self->{keep_alive}       = AnyEvent->timer(
            after    => 15,
            interval => 15,
            cb       => sub {
                my $id    = $self->{id};
                my $now   = time;
                my $since = $now - $self->{_last_keep_alive};
                if ( $since > 30 ) {
                    # will trigger a finish, which will reconnect
                    # if $self->{closed} is not set.
                    $conn->close;
                }
                elsif ( $since > 10 ) {
                    $self->ping( { keep_alive => $now } );
                }
            },
        );
    } );
}


sub metadata { shift->{metadata} // {} }
sub quiet {
    my $self = shift;

    if (@_) {
        $self->{quiet} = shift;
    }

    return $self->{quiet} // '';
}


sub on {
    my ($self, %registrations) = @_;

    for my $type (keys %registrations) {
        my $cb = $registrations{ $type };
        $self->{registry}{$type} = $cb;
    }
}


sub off {
    my ($self, @types) = @_;
    delete $self->{registry}{$_} for @types;
}

sub _do {
    my ($self, $type, @args) = @_;

    if (defined $self->{registry}{$type}) {
        $self->{registry}{$type}->($self, @args);
    }
}


sub send {
    my ($self, $msg) = @_;

    croak "Cannot send because the Slack connection is not started"
    unless $self->{started};
    croak "Cannot send because Slack has not yet said hello"
    unless $self->{said_hello};
    croak "Cannot send because the connection is finished"
    if $self->{finished};

    $msg->{id} = $self->{id}++;

    $self->{conn}->send(encode_json($msg));
}


sub ping {
    my ($self, $msg) = @_;

    $self->send({
        %{ $msg // {} },
        type => 'ping'
    });
}

sub _handle_incoming {
    my ($self, $conn, $raw) = @_;

    my $msg = try {
        decode_json($raw->body);
    }
    catch {
        my $message = $raw->body;
        croak "unable to decode incoming message: $message";
    };

    $self->{_last_keep_alive} = time;

    # Handle errors when they occur
    if ($msg->{error}) {
        $self->_handle_error($conn, $msg);
    }

    # Handle the initial hello
    elsif ($msg->{type} eq 'hello') {
        $self->_handle_hello($conn, $msg);
    }

    # Periodic response to our pings
    elsif ($msg->{type} eq 'pong') {
        $self->_handle_pong($conn, $msg);
    }

    # And anything else...
    else {
        $self->_handle_other($conn, $msg);
    }
}


sub said_hello { shift->{said_hello} // '' }
sub finished { shift->{finished} // '' }

sub _handle_hello {
    my ($self, $conn, $msg) = @_;

    $self->{said_hello}++;

    $self->_do(hello => $msg);
}

sub _handle_error {
    my ($self, $conn, $msg) = @_;

    carp "Error #$msg->{error}{code}: $msg->{error}{msg}"
        unless $self->{quiet};

    $self->_do(error => $msg);
}

sub _handle_pong {
    my ($self, $conn, $msg) = @_;

    $self->_do($msg->{type}, $msg);
}

sub _handle_other {
    my ($self, $conn, $msg) = @_;

    $self->_do($msg->{type}, $msg);
}

sub _handle_finish {
    my ($self, $conn) = @_;

    # Cancel the keep_alive watchdog
    undef $self->{keep_alive};

    $self->{finished}++;

    $self->_do('finish');

    $self->start unless $self->{closed};
}


sub close {
    my ($self) = @_;
    $self->{closed}++;
    $self->{conn}->close;
}

__END__