File Coverage

blib/lib/POE/Component/EasyDBI.pm

Criterion	Covered	Total	%
statement	232	391	59.3
branch	88	206	42.7
condition	20	78	25.6
subroutine	30	35	85.7
pod	1	19	5.2
total	371	729	50.8

package POE::Component::EasyDBI;

use strict;
use warnings FATAL =>'all';

# Initialize our version
our $VERSION = '1.30';

# Import what we need from the POE namespace
use POE;
use POE::Session;
use POE::Filter::Reference;
use POE::Filter::Line;
use POE::Wheel::Run;
use POE::Component::EasyDBI::SubProcess;

use Params::Util qw( _ARRAY _HASH );
use Scalar::Util qw( reftype );


# Miscellaneous modules
use Carp;
use vars qw($AUTOLOAD);

our %COMMANDS = map { $_ => 1 } qw(
    commit
    rollback
    begin_work
    func
    method
    insert
    do
    single
    quote
    arrayhash
    hashhash
    hasharray
    array
    arrayarray
    hash
    keyvalhash
    shutdown
    combo
);

sub AUTOLOAD {
    my $self = shift;
    my $method = $AUTOLOAD;
    $method =~ s/.*:://;
    my $call = ( $method =~ s/^_// ) ? 1 : 0;

    return unless $method =~ /[^A-Z]/;

    $method = lc( $method );

    croak "EasyDBI command $method does not exist"
        unless ( exists( $COMMANDS{ $method } ) );

    my @args = ( $method eq 'shutdown' ) ? @_ : { @_ };
    if ( $call ) {
        $poe_kernel->call($self->{ID} => $method => @args);
    } else {
        $poe_kernel->post($self->{ID} => $method => @args);
    }
}

# TODO use constants?
sub MAX_RETRIES () { 5 }
sub DEBUG () { 0 }

# Autoflush on STDOUT
# Select returns last selected handle
# So, reselect it after selecting STDOUT and setting Autoflush
#select((select(STDOUT), $| = 1)[0]);

sub spawn {
    my ($self,$session) = &new;
    return $session;
}

sub create {
    &spawn;
}

sub new {
    my $class = shift;

    # The options hash
    my %opt;

    # Support passing in a hash ref or a regular hash
    if ((@_ & 1) && _HASH($_[0])) {
        %opt = %{$_[0]};
    } else {
        # Sanity checking
        if (@_ & 1) {
            croak('POE::Component::EasyDBI requires an even number of options '
            .'passed to new()/spawn() call');
        }
        %opt = @_;
    }

    # lowercase keys
    %opt = map { lc($_) => $opt{$_} } keys %opt;

    my @valid = qw(
        dsn
        username
        password
        options
        alias
        max_retries
        ping_timeout
        use_cancel
        no_connect_failures
        connect_error
        reconnect_wait
        connected
        no_warnings
        sig_ignore_off
        no_cache
        alt_fork
        stopwatch
        query_logger
    );

    # check the DSN
    # username/password/port other options
    # should be part of the DSN
    if (!exists($opt{dsn})) {
        croak('DSN is required to spawn a new POE::Component::EasyDBI '
        .'instance!');
    }

    # check the USERNAME
    if (!exists($opt{username})) {
        croak('username is required to spawn a new POE::Component::EasyDBI '
        .'instance!');
    }

    # check the PASSWORD
    if (!exists($opt{password})) {
        croak('password is required to spawn a new POE::Component::EasyDBI '
            .'instance!');
    }

    # check the reconnect wait time
    if (exists($opt{reconnect_wait}) && $opt{reconnect_wait} < 1) {
        warn('A reconnect_wait of less than 1 second is NOT recommended, '
        .'continuing anyway');
    } elsif (!$opt{reconnect_wait}) {
        $opt{reconnect_wait} = 2;
    }

    # check the session alias
    if (!exists($opt{alias})) {
        # Debugging info...
        DEBUG && warn 'Using default Alias EasyDBI';

        # Set the default
        $opt{alias} = 'EasyDBI';
    }

    # check for connect error event
    if (exists($opt{connect_error})) {
        if (_ARRAY($opt{connect_error})) {
            unless ($#{$opt{connect_error}} > 0) {
                warn('connect_error must be an array reference that contains '
                .'at least a session and event, ignoring');
                delete $opt{connect_error};
            }
        } else {
            warn('connect_error must be an array reference that contains at '
            .'least a session and event, ignoring');
            delete $opt{connect_error};
        }
    }

    # check for connect event
    if (exists($opt{connected})) {
        if (_ARRAY($opt{connected})) {
            unless ($#{$opt{connected}} > 0) {
                warn('connected must be an array reference that contains '
                .'at least a session and event, ignoring');
                delete $opt{connected};
            }
        } else {
            warn('connected must be an array reference that contains at '
            .'least a session and event, ignoring');
            delete $opt{connected};
        }
    }

    if (exists($opt{options})) {
        unless (_HASH($opt{options})) {
            warn('options must be a hash ref, ignoring');
            delete $opt{options};
        }
    }

    if ($opt{stopwatch}) {
        eval "use Time::Stopwatch";
        if ($@) {
            warn('cannot use stopwatch (ignored), Time::Stopwatch not installed? ');
            delete $opt{stopwatch};
        }
    }

    my $keep = { map { $_ => delete $opt{$_} } @valid };
    $keep->{'parent_pid'} = $$;

    # Anything left over is unrecognized
    if (keys %opt) {
        croak('Unrecognized keys/options ('.join(',',(keys %opt))
            .') were present in new() call to POE::Component::EasyDBI!');
    }

    my $self = bless($keep,ref $class || $class);

    # Create a new session for ourself
    my $session = POE::Session->create(
        # Our subroutines
        'object_states' =>  [
            $self => {
                # Maintenance events
                '_start'        =>  'start',
                '_stop'         =>  'stop',
                'setup_wheel'   =>  'setup_wheel',
                'sig_child'     =>  'sig_child',

                # child events
                'child_error'   =>  'child_error',
                'child_closed'  =>  'child_closed',
                'child_STDOUT'  =>  'child_STDOUT',
                'child_STDERR'  =>  'child_STDERR',

                # database events
                (map { $_ => 'db_handler', uc($_) => 'db_handler' } keys %COMMANDS),

                # redefine
                'combo'         =>  'combo_query',
                'COMBO'         =>  'combo_query',
                'shutdown'      =>  'shutdown_poco',
                'SHUTDOWN'      =>  'shutdown_poco',


                # Queue handling
                'send_query'    =>  'send_query',
                'check_queue'   =>  'check_queue',
                'print_queue'   =>  'print_queue',
            },
        ],

        # Set up the heap for ourself
        'heap'  =>  {
            # The queue of DBI calls
            'queue'         =>  [],
            'idcounter'     =>  0,

            # The Wheel::Run object placeholder
            'wheel'         =>  undef,

            # How many times have we re-created the wheel?
            'retries'       =>  0,

            # Are we shutting down?
            'shutdown'      =>  0,

            # Valid options
            'opts'          => $keep,

            # The alia/s we will run under
            'alias'         =>  $keep->{alias},

            # Number of times to retry connection
            # (if connection failures option is off)
            'max_retries'   => $keep->{max_retries} || MAX_RETRIES,

            # Connection failure option
            'no_connect_failures' => $keep->{no_connect_failures} || 0,

            # extra params for actions
            action_params => {
                commit => [],
                rollback => [],
                begin_work => [],
                func => [qw( args )],
                method => [qw( function args )],
                single => [qw( separator )],
                insert => [qw( insert hash table last_insert_id )],
                array => [qw( chunked separator )],
                arrayarray => [qw( chunked )],
                keyvalhash => [qw( primary_key chunked )],
                hashhash => [qw( primary_key chunked )],
                hasharray => [qw( primary_key chunked )],
                arrayhash => [qw( chunked )],
            },
        },
    ) or die 'Unable to create a new session!';

    # save the session id
    $self->{ID} = $session->ID;

    return wantarray ? ($self,$session) : $self;
}

# This subroutine handles shutdown signals
sub shutdown_poco {
    my ($kernel, $heap) = @_[KERNEL,HEAP];

    # Check for duplicate shutdown signals
    if ($heap->{shutdown}) {
        # Okay, let's see what's going on
        if ($heap->{shutdown} == 1 && ! defined $_[ARG0]) {
            # Duplicate shutdown events
            DEBUG && warn 'Duplicate shutdown event fired!';
            return;
        } elsif ($heap->{shutdown} == 2) {
            # Tried to shutdown_NOW again...
            DEBUG && warn 'Duplicate shutdown NOW fired!';
            return;
        }
    }

    # Check if we got "NOW"
    if (defined($_[ARG0]) && uc($_[ARG0]) eq 'NOW') {
        # Actually shut down!
        $heap->{shutdown} = 2;

        if ($heap->{wheel}) {
            # KILL our subprocess
            $heap->{wheel}->kill(9);
        }

        # Delete the wheel, so we have nothing to keep
        # the GC from destructing us...
        delete $heap->{wheel};

        # Go over our queue, and do some stuff
        foreach my $queue (@{$heap->{queue}}) {
            # Skip the special EXIT actions we might have put on the queue
            if ($queue->{action} && $queue->{action} eq 'EXIT') { next }

            # Post a failure event to all the queries on the Queue
            # informing them that we have been shutdown...
            $queue->{error} = 'POE::Component::EasyDBI was '
                        .'shut down forcibly!';
            $kernel->post( $queue->{session}, $queue->{event}, $queue);

            # Argh, decrement the refcount
            $kernel->refcount_decrement( $queue->{session}, 'EasyDBI' );
        }

        # Tell the kernel to kill us!
        $kernel->signal( $_[SESSION], 'KILL' );
    } else {
        # Gracefully shut down...
        $heap->{shutdown} = 1;

        # Put into the queue EXIT for the child
        $kernel->yield( 'send_query', {
                action          =>  'EXIT',
                sql             =>  undef,
                placeholders    =>  undef,
            }
        );
    }
}

sub combo_query {
    my ($kernel, $heap, $args) = @_[KERNEL,HEAP,ARG0];

    # Get the arguments
    unless (_HASH($args)) {
        $args = { @_[ ARG0 .. $#_ ] };
    }

    # Add some stuff to the args
    # defaults to sender, but can be specified
    unless (defined($args->{session})) {
        $args->{session} = $_[SENDER]->ID();
        DEBUG && print "setting session to $args->{session}\n";
    }

    $args->{action} = $_[STATE];

    if (!exists($args->{event})) {
        # Nothing much we can do except drop this quietly...
        warn "Did not receive an event argument from caller ".$_[SENDER]->ID
            ."  State: " . $_[STATE] . " Args: " . join('',%$args);
        return;
    } else {
        my $a = ref($args->{event});
        unless (!ref($a) || $a =~ m/postback/i || $a =~ m/callback/i) {
            warn "Received an malformed event argument from caller"
                ." (only postbacks, callbacks and scalar allowed) "
                .$_[SENDER]->ID." -> State: " . $_[STATE] . " Event: $args->{event}"
                ." Args: " . %$args;
            return;
        }
    }

    my @res;
    my $handle = sub {
        push(@res, shift);
        $poe_kernel->post( $args->{session} => $args->{event} => @res ) if (delete $res[ -1 ]->{__last});
    };

    foreach my $i ( 0 .. $#{ $args->{queries} } ) {
        my ($type, $arg) = %{ $args->{queries}->[ $i ] };

        # Copy pass-through options
        for my $key ( keys %{ $args } ) {
            next if defined $arg->{$key} || $key eq 'queries';
            $arg->{$key} = $args->{$key};
        }

        $arg->{event} = $handle;
        $arg->{__last} = ( $i == $#{ $args->{queries} } );
        $kernel->call( $_[SESSION] => $type => $arg );
    }

    return;
}

# This subroutine handles queries
sub db_handler {
    my ($kernel, $heap) = @_[KERNEL,HEAP];

    # Get the arguments
    my $args;
    if (_HASH($_[ARG0])) {
        $args = $_[ARG0];
    } else {
        warn "first parameter must be a hash ref, trying to adjust. "
            ."(fix this to get rid of this message)";
        $args = { @_[ARG0 .. $#_ ] };
    }

    # fix a stupid spelling mistake
    if ($args->{seperator}) {
        $args->{separator} = $args->{seperator};
    }

    # Add some stuff to the args
    # defaults to sender, but can be specified
    unless (defined($args->{session})) {
        $args->{session} = $_[SENDER]->ID();
        DEBUG && print "setting session to $args->{session}\n";
    }

    $args->{action} = $_[STATE];

    if (!exists($args->{event})) {
        # Nothing much we can do except drop this quietly...
        warn "Did not receive an event argument from caller ".$_[SENDER]->ID
            ."  State: " . $_[STATE] . " Args: " . join(',',%$args);
        return;
    } else {
        my $a = ref($args->{event});
        unless (!ref($a) || $a =~ m/postback/i || $a =~ m/callback/i) {
            warn "Received an malformed event argument from caller"
                ." (only postbacks, callbacks and scalar allowed) "
                .$_[SENDER]->ID." -> State: " . $_[STATE] . " Event: $args->{event}"
                ." Args: " . %$args;
            return;
        }
    }

    if (!defined($args->{sql})) {
        unless ($args->{action} =~ m/^(insert|func|method|commit|rollback|begin_work)$/i) {
            $args->{error} = 'sql is not defined!';
            # Okay, send the error to the Failure Event
            $kernel->post($args->{session}, $args->{event}, $args);
            return;
        }
    } else {
        if (ref($args->{sql})) {
            $args->{error} = 'sql is not a scalar!';
            if ( reftype( $args->{event} ) && reftype( $args->{event} ) eq 'CODE' ) {
                my $callback = delete $args->{event};
                $callback->($args);
            } else {
                # send the error to the Failure Event
                $kernel->post($args->{session}, $args->{event}, $args);
            }
            return;
        }
    }

    # Check for placeholders
    if (!defined($args->{placeholders})) {
        # Create our own empty placeholders
        $args->{placeholders} = [];
    } else {
        unless (_ARRAY($args->{placeholders})) {
            $args->{error} = 'placeholders is not an array!';
            if ( reftype( $args->{event} ) && reftype( $args->{event} ) eq 'CODE' ) {
                my $callback = delete $args->{event};
                $callback->($args);
            } else {
                # Okay, send the error to the Failure Event
                $kernel->post($args->{session}, $args->{event}, $args);
            }
            return;
        }
    }

    # Check for primary_key on HASHHASH or ARRAYHASH queries
    if ($args->{action} eq 'HASHHASH' || $args->{action} eq 'HASHARRAY') {
        if (!defined($args->{primary_key})) {
            $args->{error} =  'primary_key is not defined! It must '
                            .'be a column name or a 1 based index of a column';
            if ( reftype( $args->{event} ) && reftype( $args->{event} ) eq 'CODE' ) {
                my $callback = delete $args->{event};
                $callback->($args);
            } else {
                $kernel->post($args->{session}, $args->{event}, $args);
            }
            return;
        } else {
            $args->{error} = 'primary_key is not a scalar!';
            if (ref($args->{sql})) {
                # send the error to the Failure Event
                if ( reftype( $args->{event} ) && reftype( $args->{event} ) eq 'CODE' ) {
                    my $callback = delete $args->{event};
                    $callback->($args);
                } else {
                    $kernel->post($args->{session}, $args->{event}, $args);
                }
                return;
            }
        }
    }

    # Check if we have shutdown or not
    if ($heap->{shutdown}) {
        $args->{error} = 'POE::Component::EasyDBI is shutting'
                    .' down now, requests are not accepted!';
        # Do not accept this query
        if ( reftype( $args->{event} ) && reftype( $args->{event} ) eq 'CODE' ) {
            my $callback = delete $args->{event};
            $callback->($args);
        } else {
            $kernel->post($args->{session}, $args->{event}, $args);
        }
        return;
    }

    # Increment the refcount for the session that is sending us this query
    $kernel->refcount_increment($args->{session}, 'EasyDBI');

    if ($args->{session} ne $_[SENDER]->ID()) {
        $kernel->refcount_increment($_[SENDER]->ID(), 'EasyDBI');
        $args->{sendersession} = $_[SENDER]->ID();
    }

    # Okay, fire off this query!
    $kernel->call($_[SESSION] => 'send_query' => $args);

    return;
}

# This subroutine starts the process of sending a query
sub send_query {
    my ($kernel, $heap, $args) = @_[KERNEL,HEAP,ARG0];

    # Validate that we have something
    if (!defined($args) || !_HASH($args) ) {
        return;
    }

    # Add the ID to the query
    $args->{id} = $heap->{idcounter}++;

    # Add this query to the queue
    push(@{ $heap->{queue} }, $args);

    # Send the query!
    $kernel->call($_[SESSION], 'check_queue');

    return;
}

# This subroutine is the meat - sends queries to the subprocess
sub check_queue {
    my ($kernel, $heap) = @_[KERNEL,HEAP];

    # Check if the subprocess is currently active
    return unless (!$heap->{active});

    # Check if we have a query in the queue
    return unless (scalar(@{ $heap->{queue} }) > 0);

    # shutting down?
    return unless ($heap->{shutdown} != 2);

    if ($heap->{opts}{stopwatch}) {
        tie $heap->{queue}->[0]->{stopwatch}, 'Time::Stopwatch';
    }

    # Copy what we need from the top of the queue
    my %queue;
    foreach (
        qw(id sql action placeholders no_cache begin_work commit method)
        ,@{$heap->{action_params}->{$heap->{queue}->[0]->{action}}}
    ) {
        next unless (defined($heap->{queue}->[0]->{$_}));
        $queue{$_} = $heap->{queue}->[0]->{$_};
    }

    # Send data only if we are not shutting down...
    # Set the child to 'active'
    $heap->{active} = 1;

    # Put it in the wheel
    $heap->{wheel}->put(\%queue);

    return;
}

sub print_queue {
    my ($kernel, $heap) = @_[KERNEL,HEAP];
    return scalar @{$heap->{queue}};
}

# This starts the EasyDBI
sub start {
    my ($kernel, $heap) = @_[KERNEL,HEAP];

    # Set up the alias for ourself
    $kernel->alias_set($heap->{alias}) if ($heap->{alias} ne '');

    # Create the wheel
    $kernel->call( $_[SESSION] => 'setup_wheel' );

    return;
}

# This sets up the WHEEL
sub setup_wheel {
    my ($kernel, $heap) = @_[KERNEL,HEAP];

    # Are we shutting down?
    if ($heap->{shutdown}) {
#       if ($heap->{wheel}) {
#           $heap->{wheel}->kill();
#       }
        # Do not re-create the wheel...
        delete $heap->{wheel};
        return;
    }

    # Check if we should set up the wheel
    if ($heap->{retries} == $heap->{max_retries}) {
        die 'POE::Component::EasyDBI tried '.$heap->{max_retries}
            .' times to create a Wheel and is giving up...';
    }

    my %prog = (
        'Program'       =>  \&POE::Component::EasyDBI::SubProcess::main,
        'ProgramArgs'   =>  [ $heap->{opts} ],
    );

    if ($heap->{opts}{alt_fork} && $^O ne 'MSWin32') {
        my $X = $heap->{opts}{alt_fork} ne "1" ? $heap->{opts}{alt_fork} : $^X;
        %prog = (
            'Program'   =>  "$X -MPOE::Component::EasyDBI::SubProcess -I".join(' -I',@INC)
                ." -e 'POE::Component::EasyDBI::SubProcess::main(1)'",
        );
    }

#    $kernel->sig_child( $heap->{wheel_pid} )
#        if ( $heap->{wheel_pid} );

    # Set up the SubProcess we communicate with
    $heap->{wheel} = POE::Wheel::Run->new(
        # What we will run in the separate process
        %prog,

        # Kill off existing FD's unless we're running in HELL^H^H^H^HMSWin32
        'CloseOnCall'   =>  ($^O eq 'MSWin32' ? 0 : 1),

        # Redirect errors to our error routine
        'ErrorEvent'    =>  'child_error',

        # Send child died to our child routine
        'CloseEvent'    =>  'child_closed',

        # Send input from child
        'StdoutEvent'   =>  'child_STDOUT',

        # Send input from child STDERR
        'StderrEvent'   =>  'child_STDERR',

        # Set our filters
        # Communicate with child via Storable::nfreeze
        'StdinFilter'   =>  POE::Filter::Reference->new(),
        # Receive input via Storable::nfreeze
        'StdoutFilter'  =>  POE::Filter::Reference->new(),
        # Plain ol' error lines
        'StderrFilter'  =>  POE::Filter::Line->new(),
    );

    $heap->{wheel_pid} = $heap->{wheel}->PID();

    if ( $kernel->can( "sig_child" ) ) {
        $kernel->sig_child( $heap->{wheel_pid} => 'sig_child' );
    } else {
        $kernel->sig( 'CHLD', 'sig_child' );
    }

    # Check for errors
    if (! defined $heap->{wheel}) {
        die 'Unable to create a new wheel!';
    } else {
        # Increment our retry count
        $heap->{retries}++;

        # Set the wheel to inactive
        $heap->{active} = 0;

        if ($heap->{opts}{alt_fork}) {
            $heap->{wheel}->put($heap->{opts});
        }

        # Check for queries
        $kernel->call($_[SESSION], 'check_queue');
    }

    return;
}

sub stop {
    # nothing to see here, move along
}

# Deletes a query from the queue, if it is not active
sub delete_query {
    my ($kernel, $heap) = @_[KERNEL,HEAP];
    # ARG0 = ID
    my $id = $_[ARG0];

    # Validation
    if (!defined($id)) {
        # Debugging
        DEBUG && warn 'In Delete_Query event with no arguments!';
        return;
    }

    # Check if the id exists + not at the top of the queue :)
    if (defined($heap->{queue}->[0])) {
        if ($heap->{queue}->[0]->{id} eq $id) {
            # Query is still active, nothing we can do...
            return undef;
        } else {
            # Search through the rest of the queue and see what we get
            foreach my $count (@{ $heap->{queue} }) {
                if ($heap->{queue}->[$count]->{id} eq $id) {
                    # Found a match, delete it!
                    splice(@{ $heap->{queue} }, $count, 1);

                    # Return success
                    return 1;
                }
            }
        }
    }

    # If we got here, we didn't find anything
    return undef;
}

# Handles child DIE'ing
sub child_closed {
    my ($kernel, $heap) = @_[KERNEL,HEAP];

    DEBUG && warn 'POE::Component::EasyDBI\'s Wheel closed';
    if ($heap->{shutdown}) {
#       if ($heap->{wheel}) {
#           $heap->{wheel}->kill();
#       }
        delete $heap->{wheel};
        return;
    }

    # Emit debugging information
    DEBUG && warn 'Restarting it...';

    # Create the wheel again
    delete $heap->{wheel};
    $kernel->call($_[SESSION], 'setup_wheel');

    return;
}

# Handles child error
sub child_error {
    my $heap = $_[HEAP];

    # Emit warnings only if debug is on
    DEBUG && do {
        # Copied from POE::Wheel::Run manpage
        my ($operation, $errnum, $errstr) = @_[ARG0 .. ARG2];
        warn "POE::Component::EasyDBI got an $operation error $errnum from Subprocess: '$errstr' shutdown: $heap->{shutdown}\n";
    };

    if ($heap->{shutdown}) {
        if ($heap->{wheel}) {
            $heap->{wheel}->kill();
        }
        delete $heap->{wheel};
        return;
    }
}

# Handles child STDOUT output
sub child_STDOUT {
    my ($self, $kernel, $heap, $data) = @_[OBJECT,KERNEL,HEAP,ARG0];

    # Validate the argument
    unless ( _HASH($data) ) {
        warn "POE::Component::EasyDBI did not get a hash from the child ( $data )";
        return;
    }


#   DEBUG && do {
#       require Data::Dumper;
#       print Data::Dumper->Dump([$data,$heap->{queue}[0]]);
#   };

    # Check for special DB messages with ID of 'DBI'
    if ($data->{id} eq 'DBI') {
        # Okay, we received a DBI error -> error in connection...
        if ($heap->{no_connect_failures}) {
            my $qc = {};
            if (defined($heap->{queue}->[0])) {
                $qc = $heap->{queue}[0];
            }
            $qc->{error} = $data->{error};
            if (_ARRAY($heap->{opts}{connect_error})) {
                $kernel->post(@{$heap->{opts}{connect_error}}, $qc);
                if (_HASH($heap->{queue}[0])) {
                    delete $heap->{queue}[0]->{error};
                }
            } elsif ($qc->{session} && $qc->{event}) {
                if ( reftype( $qc->{event} ) && reftype( $qc->{event} ) eq 'CODE' ) {
                    my $callback = delete $qc->{event};
                    $callback->($qc);
                } else {
                    $kernel->post($qc->{session}, $qc->{event}, $qc);
                }
            } else {
                warn "No connect_error defined and no queries in the queue while "
                ."error occurred: $data->{error}";
            }
#           print "DBI error: $data->{error}, retrying\n";
            return;
        }

        # Shutdown ourself!
        $kernel->call($_[SESSION], 'shutdown', 'NOW');

        # Too bad that we have to die...
        croak("Could not connect to DBI or database went away: $data->{error}");
    }

    if ($data->{id} eq 'DBI-CONNECTED') {
        if (_ARRAY($heap->{opts}{connected})) {
            my $query_copy = {};
            if (defined($heap->{queue}->[0])) {
                $query_copy = { %{$heap->{queue}[0]} };
            }
            $kernel->post(@{$heap->{opts}{connected}}, $query_copy);
        }
        return;
    }

    my $query;
    my $refcount_decrement = 0;

    if (exists($data->{chunked})) {
        # Get the query from the queue
        if (exists($data->{last_chunk})) {
            # last chunk, delete it out of the queue
            $query = shift(@{ $heap->{queue} });
            $refcount_decrement = 1;
        } else {
            $query = $heap->{queue}->[0];
        }
    } else {
        # Check to see if the ID matches with the top of the queue
        if ($data->{id} ne $heap->{queue}->[0]->{id}) {
            die "Internal error in queue/child consistency! ( CHILD: $data->{id} "
            ."QUEUE: $heap->{queue}->[0]->{id} )";
        }
        # Get the query from the top of the queue
        $query = shift(@{ $heap->{queue} });
        $refcount_decrement = 1;
    }

    # copy the query data, so we don't clobber the
    # stored data when using chunks
    #my $query_copy = { %{ $query } };

    # marry data from the child to the data from the queue
    #%$query_copy = (%$query_copy, %$data);

#   my $query_copy = { (%$query, %$data) };

#   my $query_copy = $query;
#   foreach (keys %$data) { $query_copy->{$_} = $data->{$_}; }

    my $query_copy = { %$query, %$data };

    # undocumented
    $poe_kernel->call( $self->{query_logger} => _log => $query_copy )
        if ( $self->{query_logger} );

#    my ($ses,$evt) = ("$query_copy->{session}", "$query_copy->{event}");

#    $kernel->call($ses => $evt => $query_copy);

    #undef $query;
    #foreach my $k (keys %$data) {
    #   $query_copy->{$k} = $data->{$k};
    #}

   if ( reftype( $query_copy->{event} ) && reftype( $query_copy->{event} ) eq 'CODE' ) {
       DEBUG && print "calling callback\n";
       my $callback = delete $query_copy->{event};
       $callback->($query_copy);
   } else {
       #DEBUG && print "calling event $query->{event} in session $query->{session} from our session ".$_[SESSION]->ID."\n";
       $kernel->call($query_copy->{session} => $query_copy->{event} => $query_copy);
   }

    # Decrement the refcount for the session that sent us a query
    if ($refcount_decrement) {
        $heap->{active} = 0;
        $kernel->refcount_decrement($query_copy->{session}, 'EasyDBI');

        if (defined($query_copy->{sendersession}) && $query_copy->{sendersession} ne $query_copy->{session}) {
            $kernel->refcount_decrement($query_copy->{sendersession}, 'EasyDBI');
        }

        # Now, that we have got a result, check if we need to send another query
        $kernel->call($_[SESSION], 'check_queue');
    }

    return;
}

# Handles child STDERR output
sub child_STDERR {
    my $input = $_[ARG0];

    # Skip empty lines as the POE::Filter::Line manpage says...
    if ($input eq '') { return }

    return if ($_[HEAP]->{opts}->{no_warnings} && !DEBUG);

    warn "$input\n";
}

sub sig_child {
    my ($kernel, $heap) = @_[KERNEL, HEAP];

    delete $heap->{wheel_pid};
    $kernel->sig_handled();

    if ( $heap->{shutdown} ) {
        # Remove our alias so we can be properly terminated
        $kernel->alias_remove($heap->{alias}) if ($heap->{alias} ne '');
        # and the child
        $kernel->sig( 'CHLD' );
    }
}

# ----------------
# Object methods

sub ID {
    my $self = shift;
    return $self->{ID};
}

# not called directly
sub DESTROY {
    my $self = shift;
    if (ref($self) && $self->ID) {
#        $poe_kernel->post($self->ID => 'shutdown' => @_);
    } else {
        return undef;
    }
}

# End of module
1;

# egg: I like Sealab 2021. This is why you can see lots of 2021 refences in my code.

#=item C
#
#Optional. If set to a true value, it will install a signal handler that will
#call $sth->cancel when a INT signal is received by the sub-process.  You may
#want to read the docs on your driver to see if this is supported.

__END__