Vetinari's $HOME

Writing your own plugins

This is a walk through a new queue plugin, which queues the mail to a (remote) QMQP-Server.

First step is to pull in the necessary modules

 use IO::Socket;
 use Text::Netstring qw( netstring_encode 
                         netstring_decode 
                         netstring_verify 
                         netstring_read );

We know, we need a server to send the mails to. This will be the same for every mail, so we can use arguments to the plugin to configure this server (and port).

Inserting this static config is done in register():

  sub register {
    my ($self, $qp, @args) = @_;
  
    die "No QMQP server specified in qmqp-forward config"
      unless @args;
  
    $self->{_qmqp_timeout} = 120;
  
    if ($args[0] =~ /^([\.\w_-]+)$/) {
      $self->{_qmqp_server} = $1;
    }
    else {
      die "Bad data in qmqp server: $args[0]";
    }
  
    $self->{_qmqp_port} = 628;
    if (@args > 1 and $args[1] =~ /^(\d+)$/) {
      $self->{_qmqp_port} = $1;
    }
  
    $self->log(LOGWARN, "WARNING: Ignoring additional arguments.") 
      if (@args > 2);
  }

We're going to write a queue plugin, so we need to hook to the queue hook.

  sub hook_queue {
    my ($self, $transaction) = @_;
  
    $self->log(LOGINFO, "forwarding to $self->{_qmqp_server}:"
                       ."$self->{_qmqp_port}");

The first step is to open a connection to the remote server.

   my $sock = IO::Socket::INET->new(
                 PeerAddr => $self->{_qmqp_server},
                 PeerPort => $self->{_qmqp_port},
                 Timeout  => $self->{_qmqp_timeout},
                 Proto    => 'tcp')
      or $self->log(LOGERROR, "Failed to connect to "
                      ."$self->{_qmqp_server}:"
                      ."$self->{_qmqp_port}: $!"),
        return(DECLINED);
    $sock->autoflush(1);

The first idea is to build the package we send, in the order described in the paragraph above:

  my $message = $transaction->header->as_string;
  $transaction->body_resetpos;
  while (my $line = $transaction->body_getline) {
    $message .= $line;
  }
  $message  = netstring_encode($message);
  $message .= netstring_encode($transaction->sender->address);
  for ($transaction->recipients) {
    push @rcpt, $_->address;
  }
  $message .= join "", netstring_encode(@rcpt);
  print $sock netstring_encode($message)
    or do {
      my $err = $!;
      $self->_disconnect($sock);
      return(DECLINED, "Failed to print to socket: $err");
    };

This would mean, we have to hold the full message in memory... Not good for large messages, and probably even slower (for large messages).

Luckily it's easy to build a netstring without the help of the Text::Netstring module if you know the size of the string (for more info about netstrings see http://cr.yp.to/proto/netstrings.txt).

We start with the sender and recipient addresses:

  my ($addrs, $headers, @rcpt);
  $addrs = netstring_encode($transaction->sender->address);
  for ($transaction->recipients) {
    push @rcpt, $_->address;
  }
  $addrs .= join "", netstring_encode(@rcpt);

Ok, we got the sender and the recipients, now let's see what size the message is.

  $headers   = $transaction->header->as_string;
  my $msglen = length($headers) + $transaction->body_length;

We've got everything we need. Now build the netstrings for the full package and the message.

First the beginning of the netstring of the full package

  # (+ 2: the ":" and "," of the message's netstring)
  print $sock ($msglen + length($msglen) + 2 + length($addrs))
               .":"
               ."$msglen:$headers" ### beginning of messages netstring
    or do { 
      my $err = $!;
      $self->_disconnect($sock);
      return(DECLINED, 
             "Failed to print to socket: $err");
    };

Go to beginning of the body

  $transaction->body_resetpos;

If the message is spooled to disk, read the message in blocks and write them to the server

  if ($transaction->body_fh) {
    my $buff;
    my $size = read $transaction->body_fh, $buff, 4096;
    unless (defined $size) {
      my $err = $!;
      $self->_disconnect($sock);
      return(DECLINED, "Failed to read from body_fh: $err");
    }
    while ($size) {
      print $sock $buff
        or do { 
          my $err = $!;
          $self->_disconnect($sock);
          return(DECLINED, "Failed to print to socket: $err");
        };
  
      $size = read $transaction->body_fh, $buff, 4096;
      unless (defined $size) {
        my $err = $!;
        $self->_disconnect($sock);
        return(DECLINED, 
               "Failed to read from body_fh: $err");
      }
    }
  }

Else we have to read it line by line ...

  else { 
    while (my $line = $transaction->body_getline) {
      print $sock $line
        or do { 
          my $err = $!;
          $self->_disconnect($sock);
          return(DECLINED, "Failed to print to socket: $err");
        };
    }
  }

Message is at the server, now finish the package.

  print $sock ","    # end of messages netstring
             .$addrs # sender + recpients
             .","    # end of netstring of 
                     #   the full package
    or do { 
      my $err = $!;
      $self->_disconnect($sock);
      return(DECLINED, 
             "Failed to print to socket: $err");
    };

We're done. Now let's see what the remote qmqpd says...

    my $answer = netstring_read($sock);
    $self->_disconnect($sock);
    
    if (defined $answer and netstring_verify($answer)) {
      $answer = netstring_decode($answer);
  
      $answer =~ s/^K// and return(OK,
                                 "Queued! $answer");
      $answer =~ s/^Z// and return(DENYSOFT, 
                                 "Deferred: $answer");
      $answer =~ s/^D// and return(DENY,
                                 "Denied: $answer");
    }

If this is the only queue/* plugin, the client will get a 451 temp error:

    return(DECLINED, "Protocol error");
  }

  sub _disconnect {
    my ($self,$sock) = @_;
    if (defined $sock) {
      eval { close $sock; };
      undef $sock;
    }
  }