diff --git a/Server/Manager/OnPipeMessage.php b/Server/Manager/OnPipeMessage.php
index d84dd869..4467ca49 100644
--- a/Server/Manager/OnPipeMessage.php
+++ b/Server/Manager/OnPipeMessage.php
@@ -3,6 +3,7 @@
 namespace Server\Manager;
 
 use Annotation\Inject;
+use Kafka\Message;
 use Server\Abstracts\Server;
 use Server\Constant;
 use Exception;
@@ -33,14 +34,9 @@ class OnPipeMessage extends Server
 		if (!is_object($message) || !($message instanceof PipeMessage)) {
 			return;
 		}
-		defer(fn() => $this->eventDispatch->dispatch(new OnAfterRequest()));
-		$this->runEvent(Constant::PIPE_MESSAGE,
-			function (\Swoole\Server $server, $src_worker_id, $message) {
-				call_user_func([$message, 'execute']);
-			}, [
-				$server, $src_worker_id, $message
-			]
-		);
+
+		call_user_func([$message, 'process'], $server, $src_worker_id);
+		$this->eventDispatch->dispatch(new OnAfterRequest());
 	}
 
 
diff --git a/Server/ServerManager.php b/Server/ServerManager.php
index a45887ca..79e60aa3 100644
--- a/Server/ServerManager.php
+++ b/Server/ServerManager.php
@@ -5,6 +5,7 @@ namespace Server;
 use Closure;
 use Exception;
 use ReflectionException;
+use Server\Manager\OnPipeMessage;
 use Server\SInterface\CustomProcess;
 use Server\SInterface\TaskExecute;
 use Server\Task\OnServerTask;
@@ -77,7 +78,7 @@ class ServerManager extends Abstracts\Server
 	 */
 	public function addListener(string $type, string $host, int $port, int $mode, array $settings = [])
 	{
-	    if ($this->portIsAready($port)) $this->stopServer($port);
+		if ($this->portIsAready($port)) $this->stopServer($port);
 		if (!$this->server) {
 			$this->createBaseServer($type, $host, $port, $mode, $settings);
 		} else {
@@ -99,6 +100,7 @@ class ServerManager extends Abstracts\Server
 		foreach ($this->sortService($configs['ports']) as $config) {
 			$this->startListenerHandler($context, $config);
 		}
+		$this->bindCallback($this->server, [Constant::PIPE_MESSAGE => [OnPipeMessage::class, 'onPipeMessage']]);
 		$this->bindCallback($this->server, $this->getSystemEvents($configs));
 	}
 
@@ -182,7 +184,6 @@ class ServerManager extends Abstracts\Server
 	private function getSystemEvents(array $configs): array
 	{
 		return array_intersect_key($configs['events'] ?? [], [
-			Constant::PIPE_MESSAGE  => '',
 			Constant::SHUTDOWN      => '',
 			Constant::WORKER_START  => '',
 			Constant::WORKER_ERROR  => '',