cc76cca9453db6d407c9e7ed3a5398a961c68d90
[PerlMail.git] / 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 my $DBI_CACHE=0;        # Cache DBI requests - may cause: MySQL server has gone away
42
43
44 open DBI_PWD,$DBI_pwd or die "open \"$DBI_pwd\": $!";
45 $DBI_pwd=<DBI_PWD>;
46 close DBI_PWD or warn "close DBI_pwd: $!";
47 chomp $DBI_pwd;
48
49 my $DBI;
50 sub DBI
51 {
52         return $DBI if $DBI_CACHE && $DBI;
53         $DBI=DBI->connect("DBI:mysql:database=$DBI_database;host=","$DBI_user",$DBI_pwd,{
54                 "PrintError"=>0,        # handled by "RaiseError" below
55                 "RaiseError"=>1,
56                 "ShowErrorStatement"=>1,
57                 "AutoCommit"=>1,
58                 }) or confess "Failed DBI->connect(): $!";
59         return $DBI;
60 }
61
62 # $name,@$cols
63 sub create_table
64 {
65 my($name,$cols)=@_;
66
67         eval { DBI()->do("drop table $name"); };
68         DBI()->do("create table $name (".join(",",@$cols).")");
69 }
70
71 sub initdb
72 {
73         create_table($DB_table,[
74                                         "id int not null auto_increment primary key",
75                                         "time timestamp not null",      # assume ." default now()"
76                                         "message longtext not null",
77                                         "retries int null default 0",   # null=>done, 0=not yet tried to submit
78                                         ],
79                         );
80         DBI()->do("alter table $DB_table add index (retries,id)");
81         print "done.\n";
82         exit 0;
83 }
84
85 sub store
86 {
87         my $message;
88         {
89                 local $/;
90                 $message=<STDIN>;
91                 }
92         close STDIN or cluck "close STDIN: $!";
93         my %row=(
94                         "message"=>$message,
95                         # assume "retries"=>0,
96                         );
97         my $prep=DBI()->prepare("insert into $DB_table (".join(",",keys(%row)).")"
98                         ." values (".join(",",map("?",keys(%row))).")");
99         $prep->execute(values(%row));
100         print $prep->{"mysql_insertid"}."\n";
101 }
102
103 sub forkoff
104 {
105         my $pid=fork();
106         confess if !defined $pid;
107         $DBI=undef();   # Prevent: Server has gone away
108         exit 0 if $pid; # parent
109         # child
110 }
111
112 my $submitonce_run=0;
113 sub submitonce
114 {
115         $submitonce_run++;
116         local *LOCK;
117         open LOCK,">>$Lock_pathname" or die "open-append \"$Lock_pathname\": $!";
118         if (!flock LOCK,LOCK_EX|LOCK_NB) {
119                 # NEVER unlink here, we are not the lock owning process!
120                 print "LOCKED\n";
121                 exit 0;
122                 }
123         my $sth=DBI()->prepare("select id,message from $DB_table where retries is not null"
124                         # process only non-problematic mails during rerun
125                         .($submitonce_run==1 ? "" : " and retries=0")
126                         ." order by retries asc,id asc");
127         $sth->execute();
128         my $progresschar="";
129         autoflush STDOUT 1;
130         my $sock;
131         while (my $row=$sth->fetchrow_hashref()) {
132                 DBI()->do("update $DB_table set retries=retries+1 where id=".$row->{"id"});
133                 if (!$sock) {
134                         $sock=IO::Socket::INET->new(
135                                         "PeerAddr"=>$PeerAddr,
136                                         "Proto"   =>"tcp",
137                                         ) or confess "IO::Socket::INET->new(\"$PeerAddr\"): $!";
138                         $sock->connected() or confess "socket not connected";
139                         }
140                 $sock->printflush(length($row->{"message"})."\n".$row->{"message"});
141                 alarm $Socket_timeout and $sock->timeout($Socket_timeout) if $Socket_timeout;
142                 my $got;
143                 my $gotlen=$sock->sysread($got,1);
144                 confess $row->{"id"}.": sysread(1)=".(!defined $gotlen ? "undef" : $gotlen).": $!"
145                                 if !defined($gotlen) || $gotlen!=1;
146                 alarm 0;
147                 if ($got ne "1") {
148                         # Prevent mailing errors from cron invoking us etc.
149                         #print STDERR "FAIL:".$row->{"id"}."\n";
150                         undef $sock;
151                         }
152                 else {
153                         DBI()->do("update $DB_table set retries=null where id=".$row->{"id"});
154                         }
155                 print $progresschar.$row->{"id"}.($got eq "1" ? "" : "=FAIL");
156                 $progresschar=",";
157                 }
158         if ($sock) {
159                 $sock->shutdown(0);     # stopped reading
160                 $sock->printflush("BYE\n");
161                 $sock->shutdown(2);     # stopped using
162                 undef $sock;
163                 }
164         print "\n" if $progresschar;
165         unlink $Lock_pathname;
166         close LOCK;
167         return $progresschar;
168 }
169
170 sub submit
171 {
172         1 while submitonce();
173 }
174
175 sub print_messages
176 {
177 my($cond)=@_;
178
179         my $sth=DBI()->prepare("select message from $DB_table $cond order by id");
180         $sth->execute();
181         while (my $row=$sth->fetchrow_hashref()) {
182                 print $row->{"message"},"\n";
183                 }
184 }
185
186 sub pending
187 {
188         print_messages("where retries is not null");
189 }
190
191 sub dump
192 {
193         print_messages("");
194 }
195
196 sub clean
197 {
198 my($keyword,$interval)=@_;
199
200         # FIXME: SQL "now()" is raced against the block above
201         my $sth=DBI()->prepare("select id,time,retries from $DB_table where time>now()");
202         $sth->execute();
203         while (my $row=$sth->fetchrow_hashref()) {
204                 warn "Message time in future: ".join(",",map(
205                                 "$_=".(!defined $row->{$_} ? "NULL" : $row->{$_})
206                                 ,keys(%$row)));
207                 }
208
209         return if $interval eq "";
210         local $_=$interval;
211         my $print=s/^print://;
212         s/(\d+)y/($1*12)."m"/ge;
213         s/(\d+)m/($1*30)."d"/ge;
214         s/(\d+)d/($1*24)."h"/ge;
215         s/(\d+)h/($1*60)."M"/ge;
216         s/(\d+)M/($1*60)."s"/ge;
217         my $sec=0;
218         $sec+=$1 while s/(\d+)s//g;
219         die "Interval parse error; left \"$_\", parsed: $interval" if $_ ne "";
220         $sth=DBI()->prepare(($print ? "select id" : "delete")
221                         ." from $DB_table where retries is null and time<from_unixtime(unix_timestamp()-$sec)");
222         $sth->execute();
223         if (!$print) {
224                 print $sth->rows()."\n";
225                 }
226         else {
227                 while (my $row=$sth->fetchrow_hashref()) {
228                         print $row->{"id"},"\n";
229                         }
230                 }
231 }
232
233 my $optwrap_err;
234 sub optwrap
235 {
236 my($func,@args)=@_;
237
238         # Prevent successful return due to --forkoff in the case of failed --store when using:
239         # perlmail-submit --store --forkoff --submit
240         if (!eval { &{$func}(@args); 1; }) {
241                 $optwrap_err||=$@||$!;
242                 die "!FINISH";
243                 die "NOTREACHED";
244                 }
245 }
246
247 $Getopt::Long::ignorecase=0;
248 # &GetOptions will return success due to: die "!FINISH"
249 # but our error detection is done by $optwrap_err.
250 GetOptions(
251                   "initdb" ,sub { optwrap \&initdb,@_; },
252                   "store"  ,sub { optwrap \&store,@_; },
253                   "forkoff",sub { optwrap \&forkoff,@_; },
254                   "submit" ,sub { optwrap \&submit,@_; },
255                   "pending",sub { optwrap \&pending,@_; },
256                   "dump"   ,sub { optwrap \&dump,@_; },
257                   "clean:s",sub { optwrap \&clean,@_; },
258                 "V|version",sub { print "perlmail-submit: $VERSION\n"; exit 0; },
259                 );
260 die $optwrap_err if defined $optwrap_err;
261 exit 0;