couchdb-trigger.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
  13 my $url = 'http://localhost:5984/monitor';
  14 my $seq = 0;
  15
  16 my $client = Mojo::Client->new;
  17 our $json   = Mojo::JSON->new;
  18 sub info { warn $_[0], " ",$json->encode($_[1]),$/ }
  19 sub debug { info "# $_[0]", $_[1] }
  20 my $error;
  21
  22 $client->keep_alive_timeout(90); # couchdb timeout is 60s
  23
  24 while( ! $error ) {
  25
  26         my $changes_feed = "$url/_changes?feed=continuous;include_docs=true;since=$seq";
  27         info 'GET' => $changes_feed;
  28         my $tx = $client->build_tx( GET => $changes_feed );
  29         $tx->res->body(sub{
  30                 my ( $content, $body ) = @_;
  31
  32                 debug 'BODY' => $body;
  33
  34                 if ( length($body) == 0 ) {
  35                         warn "# empty chunk, heartbeat?\n";
  36                         return;
  37                 }
  38
  39                 foreach ( split(/\r?\n/, $body) ) { # we can get multiple documents in one chunk
  40
  41                         my $change = $json->decode($_);
  42
  43                         if ( exists $change->{error} ) {
  44                                 $error = $change;
  45                         } elsif ( exists $change->{last_seq} ) {
  46                                 $seq = $change->{last_seq};
  47                         } elsif ( $change->{seq} <= $seq ) {
  48                                 info "ERROR: stale" => $change;
  49                         } elsif ( exists $change->{changes} ) {
  50
  51                                 my $id  = $change->{id} || warn "no id?";
  52                                 my $rev = $change->{changes}->[0]->{rev} || warn "no rev?";
  53                                    $seq = $change->{seq} || warn "no seq?";
  54
  55                                 debug 'change' => $change;
  56
  57                                 if ( my $trigger = $change->{doc}->{trigger} ) {
  58                                         if ( exists $trigger->{active} ) {
  59                                                 debug 'trigger.active',  $change->{doc}->{trigger}->{active};
  60                                         } else {
  61                                                 $trigger->{active} = [ time() ];
  62
  63                                                 debug 'TRIGGER start PUT ', $change->{doc};
  64                                                 $client->put( "$url/$id" => $json->encode( $change->{doc} ) => sub {
  65                                                         my ($client,$tx) = @_;
  66                                                         if ($tx->error) {
  67                                                                 if ( $tx->res->code == 409 ) {
  68                                                                         info "TRIGGER ABORTED started on another worker? ", $tx->error;
  69                                                                 } else {
  70                                                                         info "ERROR ", $tx->error;
  71                                                                 }
  72                                                         } else {
  73                                                                 my $res = $tx->res->json;
  74                                                                 $change->{doc}->{_rev} = $res->{rev};
  75                                                                 debug "TRIGGER execute ", $change->{doc};
  76
  77                                                                 # FIXME trigger logic?
  78
  79                                                                 push @{ $trigger->{active} }, time(); # timestamp step
  80
  81                                                                 $client->put( "$url/$id" => $json->encode( $change->{doc} ) => sub {
  82                                                                         my ($client,$tx) = @_;
  83                                                                         if ($tx->error) {
  84                                                                                 info "ERROR", $tx->error;
  85                                                                         } else {
  86                                                                                 my $res = $tx->res->json;
  87                                                                                 $change->{doc}->{_rev} = $res->{rev};
  88                                                                                 info "TRIGGER finish ", $change->{doc};
  89                                                                         }
  90                                                                 })->process;
  91                                                         }
  92                                                 })->process;
  93                                         }
  94                                 }
  95                         } else {
  96                                 warn "UNKNOWN", $json->encode($change);
  97                         }
  98
  99                 }
 100
 101         });
 102         $client->start($tx);
 103
 104 }
 105
 106 die "ERROR ", $json->encode($error) if $error;