6 $VERSION=do { my @r=(q$Revision$=~/\d+/g); sprintf "%d.".("%03d"x$#r),@r; };
14 use Carp qw(cluck confess);
15 require IO::Socket::INET;
21 open DBI_PWD,$DBI_pwd or die "open \"$DBI_pwd\": $!";
23 close DBI_PWD or warn "close DBI_pwd: $!";
26 my $DBI=DBI->connect_cached("DBI:mysql:database=$DBI_database;host=","$DBI_user",$DBI_pwd,{
27 "PrintError"=>0, # handled by "RaiseError" below
29 "ShowErrorStatement"=>1,
31 }) or confess "Failed DBI->connect(): $!";
38 eval { $DBI->do("drop table $name"); };
39 $DBI->do("create table $name (".join(",",@$cols).")");
44 create_table($DB_table,[
45 "id int not null auto_increment primary key",
46 "time timestamp not null", # assume ." default now()"
47 "message longtext not null",
48 "retries int null default 0", # null=>done, 0=not yet tried to submit
51 $DBI->do("alter table $DB_table add index (retries,id)");
63 close STDIN or cluck "close STDIN: $!";
66 # assume "retries"=>0,
68 my $prep=$DBI->prepare_cached("insert into $DB_table (".join(",",keys(%row)).")"
69 ." values (".join(",",map("?",keys(%row))).")");
70 $prep->execute(values(%row));
71 print $prep->{"mysql_insertid"}."\n";
77 confess if !defined $pid;
78 exit 0 if $pid; # parent
87 open LOCK,">>$Lock_pathname" or die "open-append \"$Lock_pathname\": $!";
88 if (!flock LOCK,LOCK_EX|LOCK_NB) {
89 # NEVER unlink here, we are not the lock owning process!
93 my $sth=$DBI->prepare("select id,message from $DB_table where retries is not null"
94 # process only non-problematic mails during rerun
95 .($submitonce_run==1 ? "" : " and retries=0")
96 ." order by retries asc,id asc");
101 while (my $row=$sth->fetchrow_hashref()) {
102 $DBI->do("update $DB_table set retries=retries+1 where id=".$row->{"id"});
104 $sock=IO::Socket::INET->new(
105 "PeerAddr"=>$PeerAddr,
107 ) or confess "IO::Socket::INET->new(\"$PeerAddr\"): $!";
108 $sock->connected() or confess "socket not connected";
110 $sock->printflush(length($row->{"message"})."\n".$row->{"message"});
111 alarm $Socket_timeout and $sock->timeout($Socket_timeout) if $Socket_timeout;
113 my $gotlen=$sock->sysread($got,1);
114 confess $row->{"id"}.": sysread(1)=".(!defined $gotlen ? "undef" : $gotlen).": $!"
115 if !defined($gotlen) || $gotlen!=1;
118 # Prevent mailing errors from cron invoking us etc.
119 #print STDERR "FAIL:".$row->{"id"}."\n";
123 $DBI->do("update $DB_table set retries=null where id=".$row->{"id"});
125 print $progresschar.$row->{"id"}.($got eq "1" ? "" : "=FAIL");
129 $sock->shutdown(0); # stopped reading
130 $sock->printflush("BYE\n");
131 $sock->shutdown(2); # stopped using
134 print "\n" if $progresschar;
135 unlink $Lock_pathname;
137 return $progresschar;
142 1 while submitonce();
147 my $sth=$DBI->prepare("select message from $DB_table where retries is not null order by id");
149 while (my $row=$sth->fetchrow_hashref()) {
150 print $row->{"message"},"\n";
156 my($keyword,$interval)=@_;
158 # FIXME: SQL "now()" is raced against the block above
159 my $sth=$DBI->prepare("select id,time,retries from $DB_table where time>now()");
161 while (my $row=$sth->fetchrow_hashref()) {
162 warn "Message time in future: ".join(",",map(
163 "$_=".(!defined $row->{$_} ? "NULL" : $row->{$_})
167 return if $interval eq "";
169 my $print=s/^print://;
170 s/(\d+)y/($1*12)."m"/ge;
171 s/(\d+)m/($1*30)."d"/ge;
172 s/(\d+)d/($1*24)."h"/ge;
173 s/(\d+)h/($1*60)."M"/ge;
174 s/(\d+)M/($1*60)."s"/ge;
176 $sec+=$1 while s/(\d+)s//g;
177 die "Interval parse error; left \"$_\", parsed: $interval" if $_ ne "";
178 $sth=$DBI->prepare(($print ? "select id" : "delete")
179 ." from $DB_table where retries is null and time<from_unixtime(unix_timestamp()-$sec)");
182 print $sth->rows()."\n";
185 while (my $row=$sth->fetchrow_hashref()) {
186 print $row->{"id"},"\n";
192 $Getopt::Long::ignorecase=0;
200 "V|version",sub { print "perlmail-submit: $VERSION\n"; exit 0; },