File Coverage

blib/lib/Net/SMTP/Pipelining.pm

Criterion	Covered	Total	%
statement	21	197	10.6
branch	0	36	0.0
condition	0	3	0.0
subroutine	7	18	38.8
pod	9	9	100.0
total	37	263	14.0

package Net::SMTP::Pipelining;

use version; $VERSION = qv('0.0.4');

use strict;
use warnings;
use Net::Cmd;
use IO::Socket;

use base("Net::SMTP");

sub pipeline {
    my ( $self, $mail ) = @_;

    if ( !defined( $self->supports("PIPELINING") ) ) {
        my $message = qq(Server does not support PIPELINING, banner was ").$self->banner().qq(");
        push @{ ${*$self}{'smtp_pipeline_errors'} },
            {
             command => "EHLO",
             code    => "",
             message => $message,
         };
        warn $message;
        return;
    }

    my @rcpts =
        ref( $mail->{to} ) eq ref( [] )
        ? @{ $mail->{to} }
        : $mail->{to};

    my @send = ( "MAIL FROM: " . $self->_addr( $mail->{mail} ) );
    push @send, map { "RCPT TO: " . $self->_addr($_) } @rcpts;
    push @send, "DATA";

    ${*$self}{'smtp_pipeline_pending_answers'} += scalar(@send);
    delete ${*$self}{'net_cmd_last_ch'};
    ${*$self}{'smtp_pipeline_errors'} = [];
    ${*$self}{'smtp_pipeline_sent'}   = [];

    # RFC 2920:
    # "If nonblocking operation is not supported, however, client SMTP
    # implementations MUST also check the TCP window size and make sure
    # that each group of commands fits entirely within the window. The
    # window size is usually, but not always, 4K octets.  Failure to
    # perform this check can lead to deadlock conditions."
    #
    # We do use non-blocking IO, but there doesn't seem to be a good
    # way of obtaining the TCP window size from a socket. All the MTAs
    # I've examined seem to ignore this issue, except for Postfix, which
    # simply sets the send buffer to the attempted message length and
    # dies on error. We'll report the error and abort.
    #
    # TODO: Add a test for this
    my $length;
    my $total_msg = join( "\015\012", @send ) . "\015\012";
    do { use bytes; $length = length($total_msg); };
    if ( $length >= $self->sockopt(SO_SNDBUF) ) {
        if ( !$self->sockopt( SO_SNDBUF, $length ) ) {
            $self->reset();
            push @{ ${*$self}{'smtp_pipeline_errors'} },
                {
                command => $total_msg,
                code    => 599,
                message =>
                    "Message too large for TCP window and could not set SO_SNDBUF to length $length: $!"
                };
            push @{ ${*$self}{'smtp_pipeline_rcpts'}{'failed'} }, @rcpts;
            return;
        }
    }
    for (@send) {
        $self->command($_);
        push @{ ${*$self}{'smtp_pipeline_sent'} }, $_;
    }
    my $success = $self->_pipe_flush();
    my @codes   = @{ $self->pipe_codes() };

    my $prev_send = scalar(@codes) > scalar(@send) ? 1 : 0;

    my %failed;

    for my $i ( $prev_send .. $#codes ) {
        my $exp = $i == $#codes ? 3 : 2;
        my ($command) = ( $send[ $i - $prev_send ] =~ m/^(\w+)/ );
        if ( $codes[$i] =~ m/^$exp/ ) {
            if ( $command eq "RCPT" ) {
                push @{ ${*$self}{'smtp_pipeline_rcpts'}{'accepted'} },
                    $rcpts[ $i - $prev_send - 1 ];
            }
        } else {

            push @{ ${*$self}{'smtp_pipeline_errors'} },
                {
                command => $send[ $i - $prev_send ],
                code    => $codes[$i],
                message => ${*$self}{'smtp_pipeline_messages'}[$i],
                };
            if ( $command eq "RCPT" ) {
                push @{ ${*$self}{'smtp_pipeline_rcpts'}{'failed'} },
                    $rcpts[ $i - $prev_send - 1 ];
            } else {
                $failed{$command} = $codes[$i];
            }
        }
    }
    if ( exists $failed{"MAIL"} || exists $failed{"DATA"} ) {
        $self->reset();
        return;
    }

    if ( scalar @{ ${*$self}{'smtp_pipeline_rcpts'}{'failed'} } > 0 ) {
        if ( scalar @{ ${*$self}{'smtp_pipeline_rcpts'}{'accepted'} } > 0 ) {
            $success = undef;
        } else {

  #    From RFC 2920:
  #    "Client SMTP implementations that employ pipelining MUST check ALL
  #    statuses associated with each command in a group. For example, if
  #    none of the RCPT TO recipient addresses were accepted the client must
  #    then check the response to the DATA command -- the client cannot
  #    assume that the DATA command will be rejected just because none of
  #    the RCPT TO commands worked.  If the DATA command was properly
  #    rejected the client SMTP can just issue RSET, but if the DATA command
  #    was accepted the client SMTP should send a single dot."
  #
  #    Untested (because Net::Server::Mail doesn't apparently allow you
  #    to return a 354 after all recipients have failed), but this should work
            $self->_pipe_dataend();
            return;
        }
    }

    $success = $self->datasend( $mail->{data} ) ? $success : undef;

    push @{ ${*$self}{'smtp_pipeline_sent'} }, $mail->{data};

    ${*$self}{'smtp_pipeline_pending_answers'}++;

    $self->_pipe_dataend();

    return $success;
}

sub pipe_recipients {
    return ${ *{ $_[0] } }{'smtp_pipeline_rcpts'};
}

sub pipe_rcpts_failed {
    return ${ *{ $_[0] } }{'smtp_pipeline_rcpts'}{'failed'};
}

sub pipe_rcpts_succeeded {
    return ${ *{ $_[0] } }{'smtp_pipeline_rcpts'}{'succeeded'};
}

sub pipe_sent {
    return ${ *{ $_[0] } }{'smtp_pipeline_sent'};
}

sub pipe_errors {
    return ${ *{ $_[0] } }{'smtp_pipeline_errors'};
}

sub _pipe_dataend {
    my $self = shift;
    my $end  = "\015\012.\015\012";

    # TODO: add test for failed write
    syswrite( $self, $end, 5 ) or warn "Last character not sent: $!";
    $self->debug_print( 1, ".\n" )
        if ( $self->debug );
}

sub pipe_flush {
    my $self = shift;
    ${*$self}{'smtp_pipeline_sent'} = [];
    $self->_pipe_flush();
}

sub _pipe_flush {
    my $self = shift;

    ${*$self}{'smtp_pipeline_messages'} = [];
    ${*$self}{'smtp_pipeline_codes'}    = [];
    ${*$self}{'net_cmd_resp'}           = [];

    while (
        scalar @{ ${*$self}{'smtp_pipeline_messages'} }
        < ${*$self}{'smtp_pipeline_pending_answers'} )
    {
        $self->response();
        push @{ ${*$self}{'smtp_pipeline_messages'} }, [ $self->message() ];
        push @{ ${*$self}{'smtp_pipeline_codes'} }, $self->code();
        ${*$self}{'net_cmd_resp'} = [];
    }
    push @{ ${*$self}{'net_cmd_resp'} },
        ${*$self}{'smtp_pipeline_messages'}[-1];
    ${*$self}{'smtp_pipeline_pending_answers'} = 0;
    delete ${*$self}{'net_cmd_last_ch'};

    if (scalar( @{ $self->pipe_codes() } )
        > scalar( @{ $self->pipe_sent() } ) )
    {
        my $prev_code = ${*$self}{'smtp_pipeline_codes'}[0];
        my $prev_resp = ${*$self}{'smtp_pipeline_messages'}[0];
        if ( $prev_code =~ m/^2/ ) {
            @{ ${*$self}{'smtp_pipeline_rcpts'}{'succeeded'} } =
                @{ ${*$self}{'smtp_pipeline_rcpts'}{'accepted'} };
            ${*$self}{'smtp_pipeline_rcpts'}{'failed'} = [];
        } else {
            ${*$self}{'smtp_pipeline_rcpts'}{'succeeded'} = [];
            @{ ${*$self}{'smtp_pipeline_rcpts'}{'failed'} } =
                @{ ${*$self}{'smtp_pipeline_rcpts'}{'accepted'} };
            push @{ ${*$self}{'smtp_pipeline_errors'} },
                {
                command => "DATA",
                code    => $prev_code,
                message => $prev_resp,
                };
        }
    } else {
        ${*$self}{'smtp_pipeline_rcpts'}{'failed'}    = [];
        ${*$self}{'smtp_pipeline_rcpts'}{'succeeded'} = [];
    }
    ${*$self}{'smtp_pipeline_rcpts'}{'accepted'} = [];

    if ( grep { $_ !~ /^[23]/ } @{ $self->pipe_codes() } ) {
        return;
    } else {
        return 1;
    }
}

sub pipe_codes {
    my $self = shift;
    ${*$self}{'smtp_pipeline_codes'};
}

sub pipe_messages {
    my $self = shift;
    ${*$self}{'smtp_pipeline_messages'};
}

1;    # Magic true value required at end of module
__END__