couchdb-changes.pl

   1 #!/usr/bin/perl
   2
   3 # http://wiki.apache.org/couchdb/HTTP_database_API#Changes
   4
   5 use warnings;
   6 use strict;
   7
   8 use lib 'common/mojo/lib';
   9
  10 use Mojo::Client;
  11 use Mojo::JSON;
  12 use Data::Dump qw(dump);
  13 use JSON::XS;
  14
  15 my $url = 'http://localhost:5984/monitor/_changes?feed=continuous;include_docs=true;since=';
  16 my $seq = 0;
  17 our $last_id_rev = '';
  18
  19 my $client = Mojo::Client->new;
  20 my $json   = Mojo::JSON->new;
  21 my $error;
  22
  23 $client->keep_alive_timeout(90); # couchdb timeout is 60s
  24
  25 while( ! $error ) {
  26
  27         warn "GET $url$seq\n";
  28         my $tx = $client->build_tx( GET => $url . $seq );
  29         $tx->res->body(sub{
  30                 my ( $content, $body ) = @_;
  31
  32                 warn "## BODY $body\n";
  33
  34                 if ( length($body) == 0 ) {
  35                         warn "# empty chunk, heartbeat?\n";
  36                         return;
  37                 }
  38
  39                 foreach my $change ( split(/\r?\n/, $body) ) { # we can get multiple documents in one chunk
  40
  41                         my $data = $json->decode($change);
  42
  43                         if ( exists $data->{error} ) {
  44                                 $error = $data;
  45                         } elsif ( exists $data->{last_seq} ) {
  46                                 $seq = $data->{last_seq};
  47                         } elsif ( $data->{seq} <= $seq ) {
  48                                 warn "# stale $body";
  49                         } elsif ( exists $data->{changes} ) {
  50
  51                                 my $id  = $data->{id} || warn "no id?";
  52                                 my $rev = $data->{changes}->[0]->{rev} || warn "no rev?";
  53                                    $seq = $data->{seq} || warn "no seq?";
  54
  55                                 if ( $last_id_rev eq "$id $rev" ) {
  56                                         warn "# duplicate $last_id_rev\n";
  57                                 } else {
  58                                         $last_id_rev = "$id $rev";
  59                                         warn "# ",dump( $data );
  60                                 }
  61
  62                         } else {
  63                                 warn "UNKNOWN", dump($data);
  64                         }
  65
  66                 }
  67
  68         });
  69         $client->start($tx);
  70
  71 }
  72
  73 die dump($error) if $error;