AnyEvent::CouchDB::Stream - Watch changes from a CouchDB database.

  use AnyEvent::CouchDB::Stream;
  my $listener = AnyEvent::CouchDB::Stream->new(
      url       => 'http://localhost:5984',
      database  => 'test',
      on_change => sub {
          my $change = shift;
          warn "document $change->{_id} updated";
      },
      on_keepalive => sub {
          warn "ping\n";
      },
      timeout => 1,
  );

package AnyEvent::CouchDB::Stream;

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

our $VERSION = '0.01';

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 $on_change    = delete $args{on_change};
    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 $uri = URI->new($server);
    $uri->path( $db. '/_changes' );
    $uri->query_form( filter => $filter, feed => "continuous" );

    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}");
                }
                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) {
                            $on_change->(JSON::decode_json($json));
                        }
                        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;
                    };
                }
           }
        );
    }
    $self;
}

1;
__END__