enh(core): optimize zmq communication (MON-6761) (#110)

centreon · Jul 16, 2024 · 16cec09 · 16cec09
1 parent 83dc3cf
commit 16cec09
Show file tree

Hide file tree

Showing 21 changed files with 295 additions and 326 deletions.
diff --git a/gorgone/gorgone/class/clientzmq.pm b/gorgone/gorgone/class/clientzmq.pm
@@ -230,7 +230,8 @@ sub event {
     $connectors->{$options{identity}}->{ping_time} = time();
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $sockets->{$options{identity}});
-
+        last if (!defined($message));
+
         # in progress
         if ($connectors->{$options{identity}}->{handshake} == 0) {
             $connectors->{$options{identity}}->{handshake} = 1;
@@ -272,9 +273,7 @@ sub event {
             if (defined($callbacks->{$options{identity}})) {
                 $callbacks->{$options{identity}}->(identity => $options{identity}, data => $data);
             }
-        }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $sockets->{$options{identity}}));
+        }        
     }
 }
 

diff --git a/gorgone/gorgone/class/core.pm b/gorgone/gorgone/class/core.pm
@@ -564,7 +564,6 @@ sub router_internal_event {
             code => $code,
             token => $token
         );
-        last unless (gorgone::standard::library::zmq_still_read(socket => $gorgone->{internal_socket}));
     }
 }
 
@@ -713,7 +712,6 @@ sub router_external_event {
                 data => $response
             );
         }
-        last unless (gorgone::standard::library::zmq_still_read(socket => $gorgone->{external_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/centreon/anomalydetection/class.pm b/gorgone/gorgone/modules/centreon/anomalydetection/class.pm
@@ -626,6 +626,7 @@ sub action_saasregister {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
+        last if (!defined($message));
 
         $connector->{logger}->writeLogDebug("[anomalydetection] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
@@ -636,8 +637,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/centreon/autodiscovery/class.pm b/gorgone/gorgone/modules/centreon/autodiscovery/class.pm
@@ -1060,7 +1060,8 @@ sub is_hdisco_synced {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[autodiscovery] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -1070,8 +1071,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/centreon/engine/class.pm b/gorgone/gorgone/modules/centreon/engine/class.pm
@@ -258,14 +258,13 @@ sub create_child {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[engine] Event: $message");
 
         if ($message !~ /^\[ACK\]/) {
             $connector->create_child(message => $message);
-        }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
+        }        
     }
 }
 

diff --git a/gorgone/gorgone/modules/centreon/judge/class.pm b/gorgone/gorgone/modules/centreon/judge/class.pm
@@ -387,7 +387,8 @@ sub action_judgelistener {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[judge] -class- event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -397,8 +398,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/centreon/legacycmd/class.pm b/gorgone/gorgone/modules/centreon/legacycmd/class.pm
@@ -726,6 +726,7 @@ sub action_centreoncommand {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
+        last if (!defined($message));
 
         $connector->{logger}->writeLogDebug("[legacycmd] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
@@ -736,8 +737,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/centreon/nodes/class.pm b/gorgone/gorgone/modules/centreon/nodes/class.pm
@@ -211,7 +211,8 @@ sub action_nodesresync {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[nodes] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -221,8 +222,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/centreon/statistics/class.pm b/gorgone/gorgone/modules/centreon/statistics/class.pm
@@ -585,7 +585,8 @@ sub rrd_update {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[statistics] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -595,8 +596,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/action/class.pm b/gorgone/gorgone/modules/core/action/class.pm
@@ -398,8 +398,9 @@ sub create_child {
 
 sub event {
     while (1) {
-        my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket}); 
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[action] Event: $message");
 
         if ($message !~ /^\[ACK\]/) {
@@ -415,8 +416,6 @@ sub event {
                 $connector->create_child(action => $action, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/cron/class.pm b/gorgone/gorgone/modules/core/cron/class.pm
@@ -373,6 +373,7 @@ sub action_deletecron {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
+        last if (!defined($message));
 
         $connector->{logger}->writeLogDebug("[cron] Event: $message");
         if ($message =~ /^\[ACK\]\s+\[(.*?)\]\s+(.*)$/m) {
@@ -390,9 +391,7 @@ sub event {
                 my $data = JSON::XS->new->utf8->decode($3);
                 $method->($connector, token => $token, data => $data);
             }
-        }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
+        }        
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/dbcleaner/class.pm b/gorgone/gorgone/modules/core/dbcleaner/class.pm
@@ -153,7 +153,8 @@ sub action_dbclean {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[dbcleaner] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -163,8 +164,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/httpserver/class.pm b/gorgone/gorgone/modules/core/httpserver/class.pm
@@ -107,10 +107,9 @@ sub class_handle_HUP {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
+        last if (!defined($message));
 
         $connector->{logger}->writeLogDebug("[httpserver] Event: $message");
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/pipeline/class.pm b/gorgone/gorgone/modules/core/pipeline/class.pm
@@ -215,7 +215,8 @@ sub check_timeout {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[pipeline] -class- event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -225,8 +226,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/proxy/class.pm b/gorgone/gorgone/modules/core/proxy/class.pm
@@ -459,6 +459,7 @@ sub event_internal {
     my $socket = $options{channel} eq 'control' ? $connector->{internal_socket} : $connector->{internal_channels}->{ $options{channel} };
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $socket);
+        last if (!defined($message));
 
         proxy(message => $message, channel => $options{channel});
         if ($connector->{stop} == 1 && (time() - $connector->{exit_timeout}) > $connector->{stop_time}) {
@@ -468,7 +469,6 @@ sub event_internal {
             defined($connector->{clients}->{ $options{channel} }) && 
             ($connector->{clients}->{ $options{channel} }->{com_read_internal} == 0 || $connector->{clients}->{ $options{channel} }->{delete} == 1)
         );
-        last unless (gorgone::standard::library::zmq_still_read(socket => $socket));
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/pull/hooks.pm b/gorgone/gorgone/modules/core/pull/hooks.pm
@@ -168,12 +168,13 @@ sub transmit_back {
 sub from_router {
     while (1) {        
         my $message = transmit_back(message => gorgone::standard::library::zmq_dealer_read_message(socket => $socket_to_internal));
+        last if (!defined($message));
+
         # Only send back SETLOGS and PONG
         if (defined($message)) {
             $logger->writeLogDebug("[pull] Read message from internal: $message");
             $client->send_message(message => $message);
         }
-        last unless (gorgone::standard::library::zmq_still_read(socket => $socket_to_internal));
     }
 }
 

diff --git a/gorgone/gorgone/modules/core/register/class.pm b/gorgone/gorgone/modules/core/register/class.pm
@@ -140,7 +140,8 @@ sub action_registerresync {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[register] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -150,8 +151,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/plugins/newtest/class.pm b/gorgone/gorgone/modules/plugins/newtest/class.pm
@@ -590,7 +590,8 @@ sub action_newtestresync {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("gorgone-newtest: class: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -600,8 +601,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }
 

diff --git a/gorgone/gorgone/modules/plugins/scom/class.pm b/gorgone/gorgone/modules/plugins/scom/class.pm
@@ -478,7 +478,8 @@ sub action_scomresync {
 sub event {
     while (1) {
         my $message = gorgone::standard::library::zmq_dealer_read_message(socket => $connector->{internal_socket});
-
+        last if (!defined($message));
+
         $connector->{logger}->writeLogDebug("[scom] Event: $message");
         if ($message =~ /^\[(.*?)\]/) {
             if ((my $method = $connector->can('action_' . lc($1)))) {
@@ -488,8 +489,6 @@ sub event {
                 $method->($connector, token => $token, data => $data);
             }
         }
-
-        last unless (gorgone::standard::library::zmq_still_read(socket => $connector->{internal_socket}));
     }
 }