view release on metacpan or  search on metacpan

lib/AnyEvent/CouchDB/Stream.pm  view on Meta::CPAN

package AnyEvent::CouchDB::Stream;

use strict;
use warnings;
use URI;
use AnyEvent::HTTP;
use Scalar::Util;
use JSON;
use Try::Tiny;
use MIME::Base64;

our $VERSION = '0.02';

sub new {
    my $class        = shift;
    my %args         = @_;
    my $server       = delete $args{url};
    my $db           = delete $args{database};
    my $timeout      = delete $args{timeout};
    my $filter       = delete $args{filter};
    my $since        = delete $args{since} || 0;
    my $on_change    = delete $args{on_change};
    my $heartbeat    = delete $args{heartbeat} || 5000;
    my $on_error     = delete $args{on_error} || sub { die @_ };
    my $on_eof       = delete $args{on_eof} || sub { };
    my $on_keepalive = delete $args{on_keepalive} || sub { };
    my $headers      = delete $args{headers}
        || { 'Content-Type' => 'application/json' };
    my $include_docs = delete $args{include_docs};

    my $uri = URI->new($server);
    $uri->path( $db. '/_changes' );
    $uri->query_form( filter => $filter, feed => "continuous", since => $since, heartbeat => $heartbeat, include_docs => $include_docs );

    if (my $userinfo = $uri->userinfo) {
        $headers->{Authorization} = 'Basic ' . encode_base64($userinfo, '');
    }

    my $self = bless {}, $class;

    {
        Scalar::Util::weaken( my $self = $self );
        my $set_timeout = $timeout
            ? sub {
            $self->{timeout}
                = AE::timer( $timeout, 0, sub { $on_error->('timeout') } );
            }
            : sub { };

        $set_timeout->();

        $self->{connection_guard} = http_get(
            $uri,
            headers   => $headers,
            on_header => sub {
                my ($headers) = @_;
                if ( $headers->{Status} ne '200' ) {
                    return $on_error->(
                        "$headers->{Status}: $headers->{Reason}: $uri");
                }
                return 1;
            },
           want_body_handle => 1,
            sub {
                my ( $handle, $headers ) = @_;
                if ($handle) {
                    $handle->on_error(
                        sub {
                            undef $handle;
                            $on_error->( $_[2] );
                        }
                    );
                    $handle->on_eof(
                        sub {
                            undef $handle;
                            $on_eof->(@_);
                        }
                    );
                    my $reader;
                    $reader = sub {
                        my ( $handle, $json ) = @_;
                        $set_timeout->();
                        if ($json) {
                            try {
                              $on_change->(JSON::decode_json($json));
                            } 
                            catch {
                              # I'm not sure why, but sometimes
                              # some weird characters show up
                              # in between the JSON objects.
                            };
                        }
                        else {
                            $on_keepalive->();
                        }
                        $handle->push_read( line => $reader );
                    };
                    $handle->push_read( line => $reader );
                    $self->{guard} = AnyEvent::Util::guard {
                        $on_eof->();
                        $handle->destroy if $handle;
                        undef $reader;