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; }