Code

msgPool
[gosa.git] / gosa-si / gosa-si-server
index e99285b80f8f4ef3eec607ed78bd86e73b7e4375..e0299693216f9db80a1657e67783690b72d67000 100755 (executable)
@@ -44,7 +44,7 @@ use File::Basename;
 use File::Path;
 use GOSA::DBsqlite;
 use GOSA::GosaSupportDaemon;
-use POE qw(Component::Server::TCP);
+use POE qw(Component::Server::TCP Wheel::Run Filter::Reference);
 use Net::LDAP;
 use Net::LDAP::Util qw(:escape);
 
@@ -77,7 +77,7 @@ our $no_arp;
 our $verbose;
 our $forground;
 our $cfg_file;
-our ($ldap_handle, $ldap_uri, $ldap_base, $ldap_admin_dn, $ldap_admin_password );
+our ($ldap_handle, $ldap_uri, $ldap_base, $ldap_admin_dn, $ldap_admin_password, $ldap_server_dn);
 
 
 # specifies the verbosity of the daemon_log
@@ -125,7 +125,7 @@ my @login_users_col_names = ('client', 'user', 'timestamp');
 our $fai_server_db;
 our $fai_server_tn = "fai_server"; 
 my $fai_server_file_name;
-our @fai_server_col_names = ('timestamp', 'server', 'release', 'tag'); 
+our @fai_server_col_names = ('timestamp', 'server', 'release', 'sections', 'tag'); 
 our $fai_release_tn = "fai_release"; 
 our @fai_release_col_names = ('timestamp', 'release', 'class', 'type', 'state'); 
 
@@ -146,6 +146,10 @@ my $messaging_file_name;
 # path to directory to store client install log files
 our $client_fai_log_dir = "/var/log/fai"; 
 
+# queue which stores taskes while max_clients are reached
+my @tasks = qw();
+
+
 %cfg_defaults = (
 "general" => {
     "log-file" => [\$log_file, "/var/run/".$prg.".log"],
@@ -313,7 +317,8 @@ sub check_pid {
             chomp( $pid );
             if( -f "/proc/$pid/stat" ) {
                 my($stat) = `cat /proc/$pid/stat` =~ m/$pid \((.+)\).*/;
-                if( $0 eq $stat ) {
+                if( $stat ) {
+                                       daemon_log("ERROR: Already running",1);
                     close( LOCK_FILE );
                     exit -1;
                 }
@@ -401,8 +406,12 @@ sub import_modules {
 sub sig_int_handler {
     my ($signal) = @_;
 
+       if(defined($ldap_handle)) {
+               $ldap_handle->disconnect;
+       }
+
     daemon_log("shutting down gosa-si-server", 1);
-    exit(1);
+    system("killall gosa-si-server");
 }
 $SIG{INT} = \&sig_int_handler;
 
@@ -844,7 +853,7 @@ sub send_msg_to_target {
         print $socket $crypted_msg."\n";
 
         daemon_log("INFO: send ".$header."msg to $address", 5);
-        daemon_log("message:\n$msg", 8);
+        daemon_log("DEBUG: message:\n$msg", 9);
         
     }
 
@@ -924,28 +933,71 @@ sub update_jobdb_status_for_send_msgs {
     }
 }
 
-
 sub _start {
     my ($kernel) = $_[KERNEL];
     &trigger_db_loop($kernel);
-    $kernel->yield('create_fai_server_db', $fai_server_tn );
+       $kernel->yield('create_fai_server_db', $fai_server_tn );
        $kernel->yield('create_fai_release_db', $fai_release_tn );
-    #$kernel->yield('create_packages_list_db', $sources_list );
 }
 
+sub next_task {
+    my ($session, $heap) = @_[SESSION, HEAP];
+
+    while ( keys( %{ $heap->{task} } ) < $max_clients ) {
+        my $next_task = shift @tasks;
+        last unless defined $next_task;
+
+        my $task = POE::Wheel::Run->new(
+                Program => sub { process_task($session, $heap, $next_task) },
+                StdioFilter => POE::Filter::Reference->new(),
+                StdoutEvent  => "task_result",
+                StderrEvent  => "task_debug",
+                CloseEvent   => "task_done",
+               );
+
+        $heap->{task}->{ $task->ID } = $task;
+    }
+}
+
+sub handle_task_result {
+    my ($kernel, $heap, $result) = @_[KERNEL, HEAP, ARG0];
+    my $client_answer = $result->{'answer'};
+    if( $client_answer =~ s/session_id=(\d+)$// ) {
+        my $session_id = $1;
+        if( defined $session_id ) {
+            my $session_reference = $kernel->ID_id_to_session($session_id);
+            if( defined $session_reference ) {
+                $heap = $session_reference->get_heap();
+            }
+        }
+        $heap->{'client'}->put($client_answer);
+    }
+    $kernel->sig(CHLD => "child_reap");
+}
+
+sub handle_task_debug {
+    my $result = $_[ARG0];
+    print STDERR "$result\n";
+}
 
-sub client_input {
+sub handle_task_done {
+    my ( $kernel, $heap, $task_id ) = @_[ KERNEL, HEAP, ARG0 ];
+    delete $heap->{task}->{$task_id};
+    $kernel->yield("next_task");
+}
+
+sub process_task {
     no strict "refs";
-    my ($kernel, $session, $heap, $input, $wheel) = @_[KERNEL, SESSION, HEAP, ARG0, ARG1];
+    my ($session, $heap, $input) = @_;
     my $session_id = $session->ID;
     my ($msg, $msg_hash, $module);
     my $error = 0;
     my $answer_l;
     my ($answer_header, @answer_target_l, $answer_source);
-    my $client_answer;
+    my $client_answer = "";
 
     daemon_log("", 5); 
-    daemon_log("INFO: Incoming msg from '".$heap->{'remote_ip'}."'", 5);
+    daemon_log("INFO: Incoming msg with session ID $session_id from '".$heap->{'remote_ip'}."'", 5);
     daemon_log("DEBUG: Incoming msg:\n$input", 9);
 
     ####################
@@ -962,6 +1014,15 @@ sub client_input {
     }
     # an error occurred
     if(( !$msg ) || ( !$msg_hash ) || ( !$module )){
+        my $sql_statement = "SELECT * FROM $main::known_clients_tn WHERE (hostname LIKE '".$heap->{'remote_ip'}."%')";
+        my $query_res = $known_clients_db->select_dbentry( $sql_statement ); 
+        while( my ($hit_num, $hit) = each %{ $query_res } ) {    
+            my $host_name = $hit->{'hostname'};
+            my $host_key = $hit->{'hostkey'};
+            my $ping_msg = "<xml> <header>gosa_ping</header> <source>$server_address</source> <target>$host_name</target></xml>";
+            my $error = &send_msg_to_target($ping_msg, $host_name, $host_key, "gosa_ping");
+            &update_jobdb_status_for_send_msgs($ping_msg, $error);
+        }
         $error++;
     }
 
@@ -972,9 +1033,9 @@ sub client_input {
         daemon_log("DEBUG: Processing module ".$module, 7);
         $answer_l = &{ $module."::process_incoming_msg" }($msg, $msg_hash, $session_id);
 
-        if ( 0 > @{$answer_l} ) {
+        if ( 0 < @{$answer_l} ) {
             my $answer_str = join("\n", @{$answer_l});
-            daemon_log("DEGUB: $module: Got answer from module: \n".$answer_str,8);
+            daemon_log("DEBUG: $module: Got answer from module: \n".$answer_str,8);
         }
     }
     if( !$answer_l ) { $error++ };
@@ -1027,10 +1088,9 @@ sub client_input {
                 }
 
                 # target of msg is GOsa
-                elsif( $answer_target eq "GOSA" ) {
-                    $answer =~ /<session_id>(\d+)<\/session_id>/;
-                    my $session_id = $1;
-                    my $add_on = "";
+                               elsif( $answer_target eq "GOSA" ) {
+                                       my $session_id = ($1) if $answer =~ s/<session_id>(\d+?)<\/session_id>//;
+                                       my $add_on = "";
                     if( defined $session_id ) {
                         $add_on = ".session_id=$session_id";
                     }
@@ -1051,7 +1111,7 @@ sub client_input {
                 # target of msg is a mac address
                 elsif( $answer_target =~ /^([0-9a-f]{2}:[0-9a-f]{2}:[0-9a-f]{2}:[0-9a-f]{2}:[0-9a-f]{2}:[0-9a-f]{2})$/i ) {
                     daemon_log("INFO: target is mac address '$answer_target', looking for host in known_clients", 5);
-                    my $sql_statement = "SELECT * FROM known_clients WHERE macaddress LIKE '$answer_target'";
+                    my $sql_statement = "SELECT * FROM known_clients WHERE macaddress='$answer_target'";
                     my $query_res = $known_clients_db->select_dbentry( $sql_statement );
                     my $found_ip_flag = 0;
                     while( my ($hit_num, $hit) = each %{ $query_res } ) {    
@@ -1113,23 +1173,18 @@ sub client_input {
         }
     }
 
-    if( $client_answer ) {
-        if( $client_answer =~ s/session_id=(\d+)$// ) {
-            my $session_id = $1;
-            if( defined $session_id ) {
-                my $session_reference = $kernel->ID_id_to_session($session_id);
-                if( defined $session_reference ) {
-                    $heap = $session_reference->get_heap();
-                }
-            }     
-        }
-        $heap->{client}->put($client_answer);
-    }
+    my $filter = POE::Filter::Reference->new();
+    my %result = ( 
+            status => "seems ok to me",
+            answer => $client_answer,
+            );
 
-    return;
-}
+    my $output = $filter->put( [ \%result ] );
+    print @$output;
 
 
+}
+
 
 sub trigger_db_loop {
        my ($kernel) = @_ ;
@@ -1168,7 +1223,7 @@ sub watch_for_new_jobs {
         my $job_msg = $hit->{'xmlmessage'};
         daemon_log("DEBUG: its time to execute $job_msg", 7); 
         my $header = $hit->{'headertag'};
-        my $sql_statement = "SELECT * FROM known_clients WHERE macaddress='$macaddress'";
+        my $sql_statement = "SELECT * FROM known_clients WHERE macaddress LIKE '$macaddress'";
                my $res_hash = $known_clients_db->select_dbentry( $sql_statement );
                # expect macaddress is unique!!!!!!
                my $target = $res_hash->{1}->{hostname};
@@ -1370,11 +1425,11 @@ sub change_goto_state {
 
 
 sub create_fai_server_db {
-    my ($table_name) = $_[ARG0];
+    my ($table_name, $kernel) = @_;
        my $result;
 
        if(defined($ldap_handle)) {
-               daemon_log("Creating fai_server_db",4);
+               daemon_log("INFO: create_fai_server_db: start", 5);
                my $mesg= $ldap_handle->search(
                        base   => $ldap_base,
                        scope  => 'sub',
@@ -1394,25 +1449,42 @@ sub create_fai_server_db {
                                                                   primkey => ['server', 'release', 'tag'],
                                                                   server => $tmp_url,
                                                                   release => $tmp_release,
+                                                                  sections => $tmp_sections,
                                                                   tag => (length($tmp_tag)>0)?$tmp_tag:"",
                                                           } );
                                           }
                                   }
                           }
                   }
-               daemon_log("Done with creating fai_server_db",4);
+               daemon_log("INFO: create_fai_server_db: finished", 5);
+
+               # TODO: Find a way to post the 'create_packages_list_db' event
+               &create_packages_list_db();
        }       
-   
+
        return $result;
 }
 
+sub run_create_fai_server_db {
+    my ($kernel, $session, $heap, $table_name) = @_[KERNEL, SESSION, HEAP, ARG0];
+    my $task = POE::Wheel::Run->new(
+            Program => sub { &create_fai_server_db($table_name,$kernel) },
+            StdoutEvent  => "session_run_result",
+            StderrEvent  => "session_run_debug",
+            CloseEvent   => "session_run_done",
+            );
+
+    $heap->{task}->{ $task->ID } = $task;
+    return;
+}
+
 
 sub create_fai_release_db {
-       my ($table_name) = $_[ARG0];
+       my ($table_name) = @_;
        my $result;
 
        if(defined($ldap_handle)) {
-               daemon_log("Creating fai_release_db",4);
+               daemon_log("INFO: create_fai_release_db: start",5);
                my $mesg= $ldap_handle->search(
                        base   => $ldap_base,
                        scope  => 'sub',
@@ -1445,14 +1517,59 @@ sub create_fai_release_db {
                                        }
                                }
                        }
-                       daemon_log("Inserting entries to DB",6);
+                       daemon_log("DEBUG: Inserting ".scalar @sql_list." entries to DB",6);
                        if(@sql_list) {
                                unshift @sql_list, "DELETE FROM $table_name";
                                $fai_server_db->exec_statementlist(\@sql_list);
                        }
-                       daemon_log("Done with inserting",6);
+                       daemon_log("DEBUG: Done with inserting",6);
+               }
+               daemon_log("INFO: create_fai_release_db: finished",5);
+       }
+
+       return $result;
+}
+sub run_create_fai_release_db {
+    my ($session, $heap, $table_name) = @_[SESSION, HEAP, ARG0];
+    my $task = POE::Wheel::Run->new(
+            Program => sub { &create_fai_release_db($table_name) },
+            StdoutEvent  => "session_run_result",
+            StderrEvent  => "session_run_debug",
+            CloseEvent   => "session_run_done",
+            );
+
+    $heap->{task}->{ $task->ID } = $task;
+    return;
+}
+
+sub get_fai_types {
+       my $tmp_classes = shift || return undef;
+       my @result;
+
+       foreach my $type(keys %{$tmp_classes}) {
+               if(defined($tmp_classes->{$type}[0]) && (!($tmp_classes->{$type}[0] =~ /^.*?removed.*?$/))) {
+                       my $entry = {
+                               type => $type,
+                               state => $tmp_classes->{$type}[0],
+                       };
+                       push @result, $entry;
+               }
+       }
+
+       return @result;
+}
+
+sub get_fai_state {
+       my $result = "";
+       my $tmp_classes = shift || return $result;
+
+       foreach my $type(keys %{$tmp_classes}) {
+               if(defined($tmp_classes->{$type}[0])) {
+                       $result = $tmp_classes->{$type}[0];
+                       
+               # State is equal for all types in class
+                       last;
                }
-               daemon_log("Done with creating fai_release_db",4);
        }
 
        return $result;
@@ -1465,14 +1582,14 @@ sub resolve_fai_classes {
        my $fai_filter= "(|(&(objectClass=FAIclass)(|(objectClass=".join(")(objectClass=", @possible_fai_classes).")))(objectClass=FAIbranch))";
        my $fai_classes;
 
-       daemon_log("Searching for FAI entries in base $fai_base",6);
+       daemon_log("DEBUG: Searching for FAI entries in base $fai_base",6);
        my $mesg= $ldap_handle->search(
                base   => $fai_base,
                scope  => 'sub',
                attrs  => ['cn','objectClass','FAIstate'],
                filter => $fai_filter,
        );
-       daemon_log("Found ".$mesg->count()." FAI entries",6);
+       daemon_log("DEBUG: Found ".$mesg->count()." FAI entries",6);
 
        if($mesg->{'resultCode'} == 0 &&
                $mesg->count != 0) {
@@ -1554,6 +1671,53 @@ sub resolve_fai_classes {
        return $result;
 }
 
+sub apply_fai_inheritance {
+       my $fai_classes = shift || return {};
+       my $tmp_classes;
+
+       # Get the classes from the branch
+       foreach my $class (keys %{$fai_classes}) {
+               # Skip subreleases
+               if($class =~ /^ou=.*$/) {
+                       next;
+               } else {
+                       $tmp_classes->{$class}= deep_copy($fai_classes->{$class});
+               }
+       }
+
+       # Apply to each subrelease
+       foreach my $subrelease (keys %{$fai_classes}) {
+               if($subrelease =~ /ou=/) {
+                       foreach my $tmp_class (keys %{$tmp_classes}) {
+                               if(!exists($fai_classes->{$subrelease}->{$tmp_class})) {
+                                       $fai_classes->{$subrelease}->{$tmp_class} =
+                                       deep_copy($tmp_classes->{$tmp_class});
+                               } else {
+                                       foreach my $type (keys %{$tmp_classes->{$tmp_class}}) {
+                                               if(!exists($fai_classes->{$subrelease}->{$tmp_class}->{$type})) {
+                                                       $fai_classes->{$subrelease}->{$tmp_class}->{$type}=
+                                                       deep_copy($tmp_classes->{$tmp_class}->{$type});
+                                               }
+                                       }
+                               }
+                       }
+               }
+       }
+
+       # Find subreleases in deeper levels
+       foreach my $subrelease (keys %{$fai_classes}) {
+               if($subrelease =~ /ou=/) {
+                       foreach my $subsubrelease (keys %{$fai_classes->{$subrelease}}) {
+                               if($subsubrelease =~ /ou=/) {
+                                       apply_fai_inheritance($fai_classes->{$subrelease});
+                               }
+                       }
+               }
+       }
+
+       return $fai_classes;
+}
+
 sub get_fai_release_entries {
        my $tmp_classes = shift || return;
        my $parent = shift || "";
@@ -1589,87 +1753,72 @@ sub get_fai_release_entries {
        return @result;
 }
 
-sub get_fai_types {
-       my $tmp_classes = shift || return undef;
-       my @result;
+sub deep_copy {
+       my $this = shift;
+       if (not ref $this) {
+               $this;
+       } elsif (ref $this eq "ARRAY") {
+               [map deep_copy($_), @$this];
+       } elsif (ref $this eq "HASH") {
+               +{map { $_ => deep_copy($this->{$_}) } keys %$this};
+       } else { die "what type is $_?" }
+}
 
-       foreach my $type(keys %{$tmp_classes}) {
-               if(defined($tmp_classes->{$type}[0]) && (!($tmp_classes->{$type}[0] =~ /^.*?removed.*?$/))) {
-                       my $entry = {
-                               type => $type,
-                               state => $tmp_classes->{$type}[0],
-                       };
-                       push @result, $entry;
-               }
-       }
 
-       return @result;
+sub session_run_result {
+    my ($kernel, $heap, $client_answer) = @_[KERNEL, HEAP, ARG0];    
+    $kernel->sig(CHLD => "child_reap");
 }
 
-sub apply_fai_inheritance {
-       my $fai_classes = shift || return {};
-       my $tmp_classes;
+sub session_run_debug {
+    my $result = $_[ARG0];
+    print STDERR "$result\n";
+}
 
-       # Get the classes from the branch
-       foreach my $class (keys %{$fai_classes}) {
-               # Skip subreleases
-               if($class =~ /^ou=.*$/) {
-                       next;
-               } else {
-                       $tmp_classes->{$class}= deep_copy($fai_classes->{$class});
-               }
-       }
+sub session_run_done {
+    my ( $kernel, $heap, $task_id ) = @_[ KERNEL, HEAP, ARG0 ];
+    delete $heap->{task}->{$task_id};
+}
 
-       # Apply to each subrelease
-       foreach my $subrelease (keys %{$fai_classes}) {
-               if($subrelease =~ /ou=/) {
-                       foreach my $tmp_class (keys %{$tmp_classes}) {
-                               if(!exists($fai_classes->{$subrelease}->{$tmp_class})) {
-                                       $fai_classes->{$subrelease}->{$tmp_class} = 
-                                       deep_copy($tmp_classes->{$tmp_class});
-                               } else {
-                                       foreach my $type (keys %{$tmp_classes->{$tmp_class}}) {
-                                               if(!exists($fai_classes->{$subrelease}->{$tmp_class}->{$type})) {
-                                                       $fai_classes->{$subrelease}->{$tmp_class}->{$type}=
-                                                       deep_copy($tmp_classes->{$tmp_class}->{$type});
-                                               }
-                                       }
-                               }
-                       }
-               }
+sub create_sources_list {
+       my $result="/tmp/gosa_si_tmp_sources_list";
+
+       # Remove old file
+       if(stat($result)) {
+               unlink($result);
        }
 
-       # Find subreleases in deeper levels
-       foreach my $subrelease (keys %{$fai_classes}) {
-               if($subrelease =~ /ou=/) {
-                       foreach my $subsubrelease (keys %{$fai_classes->{$subrelease}}) {
-                               if($subsubrelease =~ /ou=/) {
-                                       apply_fai_inheritance($fai_classes->{$subrelease});
-                               }
+       my $fh;
+       open($fh, ">$result") or return undef;
+       if(defined($ldap_server_dn) && length($ldap_server_dn)>0) {
+               my $mesg=$ldap_handle->search(
+                       base    => $ldap_server_dn,
+                       scope   => 'base',
+                       attrs   => 'FAIrepository',
+               );
+
+               if($mesg->count) {
+                       foreach my $entry(@{$mesg->{'entries'}}) {
+                               my ($server, $tag, $release, $sections)= split /\|/, $entry->get_value('FAIrepository');
+                               my $line = "deb $server $release";
+                               $sections =~ s/,/ /g;
+                               $line.= " $sections";
+                               print $fh $line."\n";
                        }
                }
        }
-       
-       return $fai_classes;
-}
+       close($fh);
 
-sub deep_copy {
-       my $this = shift;
-       if (not ref $this) {
-               $this;
-       } elsif (ref $this eq "ARRAY") {
-               [map deep_copy($_), @$this];
-       } elsif (ref $this eq "HASH") {
-               +{map { $_ => deep_copy($this->{$_}) } keys %$this};
-       } else { die "what type is $_?" }
+       return $result;
 }
 
 sub create_packages_list_db {
-    my ($sources_file) = $_[ARG0] ;
+    my ($sources_file) = @_ || &create_sources_list;
     my $line;
+    daemon_log("INFO: create_packages_list_db: start", 5); 
 
     open(CONFIG, "<$sources_file") or do {
-        daemon_log( "ERROR: Failed to open '$sources_file', creating packages.db stopped", 1);
+        daemon_log( "ERROR: create_packages_list_db: Failed to open '$sources_file'", 1);
         return;
     };
     
@@ -1700,9 +1849,19 @@ sub create_packages_list_db {
 
     close (CONFIG);
 
-
+    daemon_log("INFO: create_packages_list_db: finished", 5); 
     return;
 }
+sub run_create_packages_list_db {
+    my ($session, $heap) = @_[SESSION, HEAP];
+    my $task = POE::Wheel::Run->new(
+            Program => sub {&create_packages_list_db},
+            StdoutEvent  => "session_run_result",
+            StderrEvent  => "session_run_debug",
+            CloseEvent   => "session_run_done",
+            );
+    $heap->{task}->{ $task->ID } = $task;
+}
 sub parse_package_info {
   my ($baseurl, $dist, $section)= @_;
   my ($package);
@@ -1710,13 +1869,9 @@ sub parse_package_info {
   my ($path) = ($baseurl =~ m%://[^/]*(.*)$%);
 
   foreach $package ("Packages.gz"){
-    print "getting $baseurl , $dist , $section... \n";
+    daemon_log("DEBUG: create_packages_list: fetch $baseurl, $dist, $section", 7);
     get_package( "$baseurl/dists/$dist/$section/binary-$arch/$package", "$outdir/$dist/$section" );
-    print "done\n";
-    print "parsing packages... \n"; 
     parse_package( "$outdir/$dist/$section", $dist, $path );
-    print "done\n";
-    last;
   }
 }
 sub get_package {
@@ -1726,22 +1881,34 @@ sub get_package {
   -d "$tpath" || mkpath "$tpath";
 
   # This is ugly, but I've no time to take a look at "how it works in perl"
-  if( system("wget '$url' -O '$dest' 2>/dev/null") ) {
-    system("gzip -cd '$dest' > '$dest.in'");
-    system("rm -f '$dest'");
+  if(0 == system("wget '$url' -O '$dest' 2>/dev/null") ) {
+      system("gunzip -cd '$dest' > '$dest.in'");
+      system("rm -f '$dest'");
+  } else {
+      daemon_log("ERROR: create_packages_list_db: get_packages: fetching '$url' failed!", 1);
   }
-
   return 0;
 }
 sub parse_package {
     my ($path, $dist, $srv_path )= @_;
     my ($package, $version, $section, $description);
     my @sql_list;
+    my $PACKAGES;
+
+    if(not stat("$path.in")) {
+        daemon_log("ERROR: create_packages_list: parse_package: file '$path.in' is not readable",1);
+        return;
+    }
 
-    open(PACKAGES, "<$path.in") or return;
+    open($PACKAGES, "<$path.in");
+       if(not defined($PACKAGES)) {
+        daemon_log("ERROR: create_packages_list_db: parse_package: can not open '$path.in'",1); 
+        return;
+    }
 
     # Read lines
-    while (my $line = <PACKAGES>){
+    while (<$PACKAGES>){
+        my $line = $_;
         # Unify
         chop($line);
 
@@ -1782,7 +1949,7 @@ sub parse_package {
 
     }
 
-    close( PACKAGES );
+    close( $PACKAGES );
     unlink( "$path.in" );
     
     $packages_list_db->exec_statementlist(\@sql_list);
@@ -1801,19 +1968,6 @@ sub parse_package {
 #  $repo_files{ "${srvdir}${path}/$file" } = \%fileinfo;
 #}
 
-sub next_task {
-    my ($heap) = $_[HEAP];
-
-    if( keys(%{$heap->{task}}) < $max_clients ) {
-        
-    }
-    
-    #$heap->{task}->{$task->ID} = $task;
-}
-
-sub task_result {
-
-}
 
 #==== MAIN = main ==============================================================
 #  parse commandline options
@@ -1907,23 +2061,38 @@ $messaging_db->create_table($messaging_tn, \@messaging_col_names);
 $xml = new XML::Simple();
 
 # create socket for incoming xml messages
+
 POE::Component::Server::TCP->new(
        Port => $server_port,
-       ClientInput => \&client_input,
+       ClientInput => sub {
+        my ($kernel, $input) = @_[KERNEL, ARG0];
+        push(@tasks, $input);
+        $kernel->yield("next_task");
+        },
+    InlineStates => {
+        next_task => \&next_task,
+        task_result => \&handle_task_result,
+        task_done   => \&handle_task_done,
+        task_debug  => \&handle_task_debug,
+        child_reap => sub { "Do nothing special. I'm just a comment, but i'm necessary!"  },
+    }
 );
+
 daemon_log("start socket for incoming xml messages at port '$server_port' ", 1);
 
 # create session for repeatedly checking the job queue for jobs
 POE::Session->create(
        inline_states => {
                _start => \&_start,
-        next_task => \&next_task,
-        task_result => \&task_result,
                watch_for_new_jobs => \&watch_for_new_jobs,
         watch_for_done_jobs => \&watch_for_done_jobs,
-        create_packages_list_db => \&create_packages_list_db,
-        create_fai_server_db => \&create_fai_server_db,
-        create_fai_release_db => \&create_fai_release_db,
+        create_packages_list_db => \&run_create_packages_list_db,
+        create_fai_server_db => \&run_create_fai_server_db,
+        create_fai_release_db => \&run_create_fai_release_db,
+        session_run_result => \&session_run_result,
+        session_run_debug => \&session_run_debug,
+        session_run_done => \&session_run_done,
+        child_reap => sub { "Do nothing special. I'm just a comment, but i'm necessary!"  },
        }
 );