Net::LDAP and POE

[Jonathan Steinert]

So, for the last couple days I've been thinking about how to make 
Net::LDAP happy to work with poe.

After discussing things with people on IRC, whining about Convert::ASN1 
and several other things I finally got this working. The current 
implementation consists of a Subclass to Net::LDAP called POE_LDAP (no, 
this will not be the final name) and a POE Filter class for receiving 
messages called Filter_ASN1 (same comment). POE_LDAP is an object that 
(so far) is POE safe, and non-blocking. It uses postbacks and callbacks 
to get back to your POE sessions. I chose this route because it allows 
me to get this done in under 200 lines of code total, including the example.

I'll attach the files here, and would love to hear from people on just 
about anything regarding it. The code is /NOT/ intended for production 
use by any means, if you can't tell by just reading it. There are print 
statements littered about and several hacks in place to get the job 
done. I just wanted to proto up something that worked to make sure it 
could be done.

Please let me know what you think, on or off list. I hope to turn this 
in to a production-quality implementation soon.

Thanks for reading,

Jonathan Steinert
#!/usr/bin/perl

$| = 1;
use strict;
use warnings;

use POE;
use POE_LDAP;

POE::Session->create(
  inline_states => {
    _start => sub {
      print "Start\n";
      my $ldap = $_[HEAP]->{foo} = POE_LDAP->new( 'localhost', async => 1 );
      print "LDAP: $ldap\n";
      $ldap->bind( callback => $_[SESSION]->postback( 'bound' ) );
    },
    _stop => sub {
      print "Stop\n";
    },
    bound => sub {
      print "Bound! $_[ARG1]->[0]\n";
      $_[HEAP]->{foo}->search( base => "ou=People,dc=kuiki,dc=net", filter => "(objectClass=person)", callback => $_[SESSION]->postback( 'search' ) );
    },
    search => sub {
      print "Got Search result: @{$_[ARG1]}\n";
      if (exists( $_[ARG1]->[1] ) && $_[ARG1]->[1]->can( 'dump' )) {
        print $_[ARG1]->[1]->dump();
      }
      elsif (@{$_[ARG1]} == 1) {
        delete $_[HEAP]->{foo};
      }
    },
  },
);

print "Before run()\n";

POE::Kernel->run();

print "After run()\n";
package Filter_ASN1;

use strict;
use warnings;
use Convert::ASN1 qw(asn_decode_tag asn_decode_length);

sub new {
  my $class = shift;

  my $self = bless {
    buffer => '',
  }, (ref $class || $class);

  return $self;
}

sub get {
  my $self = shift;
  my $blocks = shift;

  $self->{buffer} .= join( '', @$blocks );

  my $return_blocks = [];
  
  while (1) {
    my ($tb, $tag) = asn_decode_tag( $self->{buffer} ) or last;
    my ($lb, $len) = asn_decode_length( substr( $self->{buffer},$tb,8 ) ) or last;
    my $length = $tb + $lb + $len;
    
    if ($length <= length $self->{buffer}) {
      push @$return_blocks, substr( $self->{buffer},0,$length );
      substr( $self->{buffer}, 0, $length ) = '';
    }
    else {
      last;
    }
  }
  return $return_blocks;
}


sub put {
  die( "Unimplemented call to put()\n" );
}

1;
package POE_LDAP;

use base 'Net::LDAP';

use Convert::ASN1;
use Net::LDAP::ASN qw(LDAPResponse);
use Filter_ASN1;
use POE qw(Filter::Stream Wheel::ReadWrite Driver::SysRW);
use WeakRef;

use strict;
use warnings;

my $poe_session;
my $poe_states = {
  _start => sub {
    print "New POE_LDAP Session coming online\n";
  },
  _stop => sub {
    print "POE_LDAP Session going offline\n";
    $poe_session = undef;
  },
  new_ldap => sub {
    print "Registering New Net::LDAP($_[ARG0]) for comm.\n";
    my $wheel = $_[ARG0]->{$_[ARG0] . 'poe_wheel'} = POE::Wheel::ReadWrite->new(
      Handle => $_[ARG0]->socket(),
      Driver => POE::Driver::SysRW->new(),
      InputFilter => Filter_ASN1->new(),
      OutputFilter => POE::Filter::Stream->new(),
      InputEvent => 'got_input',
      FlushedEvent => 'flushed_output',
      ErrorEvent => 'wheel_error',
    );
    weaken($_[HEAP]->{wheels}->{$wheel->ID()} = $_[ARG0]);
  },
  remove_ldap => sub {
    if ( exists( $_[ARG0]->{$_[ARG0] . 'poe_wheel'} ) ) {
      print "Valid destruction($_[ARG0])\n";
      my $wheel = delete $_[ARG0]->{$_[ARG0] . 'poe_wheel'};
      delete $_[HEAP]->{wheels}->{$wheel->ID()};
      $_[KERNEL]->refcount_decrement( $poe_session->ID(), 'POE-LDAP' );
    }
    else {
      print "Invalid destruction($_[ARG0])\n";
    }
  },
  got_input => sub {
    my $result = $LDAPResponse->decode($_[ARG0]);

    my $mid = $result->{messageID};
    my $mesg = $_[HEAP]->{wheels}->{$_[ARG1]}->{net_ldap_mesg}->{$mid};

    unless ($mesg) {
      if (my $ext = $result->{protocolOp}{extendedResp}) {
   	if (($ext->{responseName} || '') eq '1.3.6.1.4.1.1466.20036') {
	  # notice of disconnection
 	  die("Notice of Disconnection");
  	}
      }

      print "Unexpected PDU, ignored\n";
      return;
    }
    
    $mesg->decode($result);
  },
};

sub new {
  print "new POE_LDAP object being blessed into existance\n";
  my $class = shift;

  my $net_ldap = $class->SUPER::new(@_);

  $poe_session ||= POE::Session->create(
    inline_states => $poe_states,
  );

  POE::Kernel->refcount_increment( $poe_session->ID(), 'POE-LDAP' );
  POE::Kernel->call( $poe_session, 'new_ldap', $net_ldap );
  
  return $net_ldap;
}

sub DESTROY {
  my $self = shift;
  
  POE::Kernel->call( $poe_session, 'remove_ldap', $self );
  
  $self->SUPER::DESTROY(@_);
}

sub _sendmesg {
  my $self = shift;
  my $mesg = shift;

  $self->{$self . 'poe_wheel'}->put( $mesg->pdu );

  my $mid = $mesg->mesg_id;

  $self->{net_ldap_mesg}->{$mid} = $mesg;
}