perlmail-submit

   1 #! /usr/bin/perl
   2 #
   3 #       $Id$
   4 # Copyright (C) 2002-2003 Jan Kratochvil <project-PerlMail@jankratochvil.net>
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  19
  20
  21 use vars qw($VERSION);
  22 $VERSION=do { my @r=(q$Revision$=~/\d+/g); sprintf "%d.".("%03d"x$#r),@r; };
  23 use strict;
  24 use warnings;
  25
  26 use File::Basename;
  27 BEGIN {
  28         use lib $ENV{"PERLMAIL_BASEDIR"} || File::Basename::dirname($0);
  29         use PerlMail::Config;
  30         }
  31
  32 use Getopt::Long;
  33 use DBI;
  34 use Carp qw(cluck confess);
  35 require IO::Socket::INET;
  36 use IO::Handle;
  37 use POSIX qw(mktime);
  38 use Fcntl qw(:flock);
  39
  40
  41 open DBI_PWD,$DBI_pwd or die "open \"$DBI_pwd\": $!";
  42 $DBI_pwd=<DBI_PWD>;
  43 close DBI_PWD or warn "close DBI_pwd: $!";
  44 chomp $DBI_pwd;
  45
  46 my $DBI=DBI->connect_cached("DBI:mysql:database=$DBI_database;host=","$DBI_user",$DBI_pwd,{
  47                 "PrintError"=>0,        # handled by "RaiseError" below
  48                 "RaiseError"=>1,
  49                 "ShowErrorStatement"=>1,
  50                 "AutoCommit"=>1,
  51                 }) or confess "Failed DBI->connect(): $!";
  52
  53 # $name,@$cols
  54 sub create_table
  55 {
  56 my($name,$cols)=@_;
  57
  58         eval { $DBI->do("drop table $name"); };
  59         $DBI->do("create table $name (".join(",",@$cols).")");
  60 }
  61
  62 sub initdb
  63 {
  64         create_table($DB_table,[
  65                                         "id int not null auto_increment primary key",
  66                                         "time timestamp not null",      # assume ." default now()"
  67                                         "message longtext not null",
  68                                         "retries int null default 0",   # null=>done, 0=not yet tried to submit
  69                                         ],
  70                         );
  71         $DBI->do("alter table $DB_table add index (retries,id)");
  72         print "done.\n";
  73         exit 0;
  74 }
  75
  76 sub store
  77 {
  78         my $message;
  79         {
  80                 local $/;
  81                 $message=<STDIN>;
  82                 }
  83         close STDIN or cluck "close STDIN: $!";
  84         my %row=(
  85                         "message"=>$message,
  86                         # assume "retries"=>0,
  87                         );
  88         my $prep=$DBI->prepare_cached("insert into $DB_table (".join(",",keys(%row)).")"
  89                         ." values (".join(",",map("?",keys(%row))).")");
  90         $prep->execute(values(%row));
  91         print $prep->{"mysql_insertid"}."\n";
  92 }
  93
  94 sub forkoff
  95 {
  96         my $pid=fork();
  97         confess if !defined $pid;
  98         exit 0 if $pid; # parent
  99         # child
 100 }
 101
 102 my $submitonce_run=0;
 103 sub submitonce
 104 {
 105         $submitonce_run++;
 106         local *LOCK;
 107         open LOCK,">>$Lock_pathname" or die "open-append \"$Lock_pathname\": $!";
 108         if (!flock LOCK,LOCK_EX|LOCK_NB) {
 109                 # NEVER unlink here, we are not the lock owning process!
 110                 print "LOCKED\n";
 111                 exit 0;
 112                 }
 113         my $sth=$DBI->prepare("select id,message from $DB_table where retries is not null"
 114                         # process only non-problematic mails during rerun
 115                         .($submitonce_run==1 ? "" : " and retries=0")
 116                         ." order by retries asc,id asc");
 117         $sth->execute();
 118         my $progresschar="";
 119         autoflush STDOUT 1;
 120         my $sock;
 121         while (my $row=$sth->fetchrow_hashref()) {
 122                 $DBI->do("update $DB_table set retries=retries+1 where id=".$row->{"id"});
 123                 if (!$sock) {
 124                         $sock=IO::Socket::INET->new(
 125                                         "PeerAddr"=>$PeerAddr,
 126                                         "Proto"   =>"tcp",
 127                                         ) or confess "IO::Socket::INET->new(\"$PeerAddr\"): $!";
 128                         $sock->connected() or confess "socket not connected";
 129                         }
 130                 $sock->printflush(length($row->{"message"})."\n".$row->{"message"});
 131                 alarm $Socket_timeout and $sock->timeout($Socket_timeout) if $Socket_timeout;
 132                 my $got;
 133                 my $gotlen=$sock->sysread($got,1);
 134                 confess $row->{"id"}.": sysread(1)=".(!defined $gotlen ? "undef" : $gotlen).": $!"
 135                                 if !defined($gotlen) || $gotlen!=1;
 136                 alarm 0;
 137                 if ($got ne "1") {
 138                         # Prevent mailing errors from cron invoking us etc.
 139                         #print STDERR "FAIL:".$row->{"id"}."\n";
 140                         undef $sock;
 141                         }
 142                 else {
 143                         $DBI->do("update $DB_table set retries=null where id=".$row->{"id"});
 144                         }
 145                 print $progresschar.$row->{"id"}.($got eq "1" ? "" : "=FAIL");
 146                 $progresschar=",";
 147                 }
 148         if ($sock) {
 149                 $sock->shutdown(0);     # stopped reading
 150                 $sock->printflush("BYE\n");
 151                 $sock->shutdown(2);     # stopped using
 152                 undef $sock;
 153                 }
 154         print "\n" if $progresschar;
 155         unlink $Lock_pathname;
 156         close LOCK;
 157         return $progresschar;
 158 }
 159
 160 sub submit
 161 {
 162         1 while submitonce();
 163 }
 164
 165 sub pending
 166 {
 167         my $sth=$DBI->prepare("select message from $DB_table where retries is not null order by id");
 168         $sth->execute();
 169         while (my $row=$sth->fetchrow_hashref()) {
 170                 print $row->{"message"},"\n";
 171                 }
 172 }
 173
 174 sub clean
 175 {
 176 my($keyword,$interval)=@_;
 177
 178         # FIXME: SQL "now()" is raced against the block above
 179         my $sth=$DBI->prepare("select id,time,retries from $DB_table where time>now()");
 180         $sth->execute();
 181         while (my $row=$sth->fetchrow_hashref()) {
 182                 warn "Message time in future: ".join(",",map(
 183                                 "$_=".(!defined $row->{$_} ? "NULL" : $row->{$_})
 184                                 ,keys(%$row)));
 185                 }
 186
 187         return if $interval eq "";
 188         local $_=$interval;
 189         my $print=s/^print://;
 190         s/(\d+)y/($1*12)."m"/ge;
 191         s/(\d+)m/($1*30)."d"/ge;
 192         s/(\d+)d/($1*24)."h"/ge;
 193         s/(\d+)h/($1*60)."M"/ge;
 194         s/(\d+)M/($1*60)."s"/ge;
 195         my $sec=0;
 196         $sec+=$1 while s/(\d+)s//g;
 197         die "Interval parse error; left \"$_\", parsed: $interval" if $_ ne "";
 198         $sth=$DBI->prepare(($print ? "select id" : "delete")
 199                         ." from $DB_table where retries is null and time<from_unixtime(unix_timestamp()-$sec)");
 200         $sth->execute();
 201         if (!$print) {
 202                 print $sth->rows()."\n";
 203                 }
 204         else {
 205                 while (my $row=$sth->fetchrow_hashref()) {
 206                         print $row->{"id"},"\n";
 207                         }
 208                 }
 209 }
 210
 211
 212 $Getopt::Long::ignorecase=0;
 213 die if !GetOptions(
 214                   "initdb" ,\&initdb,
 215                   "store"  ,\&store,
 216                   "forkoff",\&forkoff,
 217                   "submit" ,\&submit,
 218                   "pending",\&pending,
 219                   "clean:s",\&clean,
 220                 "V|version",sub { print "perlmail-submit: $VERSION\n"; exit 0; },
 221                 );
 222 exit 0;