update

seata · Oct 31, 2021 · 491bf69 · 491bf69
1 parent 3955f87
commit 491bf69
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 6 deletions.
diff --git a/src/Core/Rpc/Runtime/Swoole/SocketChannel.php b/src/Core/Rpc/Runtime/Swoole/SocketChannel.php
@@ -15,6 +15,7 @@
 use Hyperf\Seata\Core\Protocol\MessageType;
 use Hyperf\Seata\Core\Protocol\RpcMessage;
 use Hyperf\Seata\Core\Rpc\Address;
+use Hyperf\Seata\Core\Rpc\Runtime\ProcessorManager;
 use Hyperf\Seata\Core\Rpc\Runtime\SocketChannelInterface;
 use Hyperf\Seata\Core\Rpc\Runtime\V1\ProtocolV1Decoder;
 use Hyperf\Seata\Core\Rpc\Runtime\V1\ProtocolV1Encoder;
@@ -75,22 +76,29 @@ public function getAddress(): Address
     protected function createRecvLoop()
     {
         Coroutine::create(function () {
+            $processorManger = ApplicationContext::getContainer()->get(ProcessorManager::class);
             while (true) {
                 try {
                     $data = $this->socket->recvAll();
                     if (! $data) {
+                        // Coroutines give up
+                        usleep(1);
                         continue;
                     }
                     $byteBuffer = ByteBuffer::wrapBinary($data);
                     $rpcMessage = $this->protocolDecoder->decode($byteBuffer);
+
+                    $processorManger->dispatch($this, $rpcMessage);
+
                     if (isset($this->responses[$rpcMessage->getId()])) {
                         $responseChannel = $this->responses[$rpcMessage->getId()];
                         $responseChannel->push($rpcMessage);
-                    } elseif ($rpcMessage->getMessageType() === MessageType::TYPE_HEARTBEAT_MSG) {
-                        var_dump('heartbeat', $rpcMessage);
-                    } else {
+                    }  else {
                         var_dump('else', $rpcMessage);
                     }
+//                    elseif ($rpcMessage->getMessageType() === MessageType::TYPE_HEARTBEAT_MSG) {
+//                        var_dump('heartbeat', $rpcMessage);
+//                    }
                 } catch (\InvalidArgumentException $exception) {
                     var_dump($exception->getMessage());
                 }

diff --git a/src/Core/Rpc/Runtime/Swow/SocketChannel.php b/src/Core/Rpc/Runtime/Swow/SocketChannel.php
@@ -42,16 +42,13 @@ class SocketChannel implements SocketChannelInterface
 
     protected Channel $sendChannel;
 
-    protected ProcessorManager $processorManger;
-
     public function __construct(Socket $socket, Address $address)
     {
         $this->socket = $socket;
         $this->address = $address;
         $container = ApplicationContext::getContainer();
         $this->protocolEncoder = $container->get(ProtocolV1Encoder::class);
         $this->protocolDecoder = $container->get(ProtocolV1Decoder::class);
-//        $this->processorManger = $container->get(ProcessorManager::class);
         $this->sendChannel = new Channel();
         $this->createRecvLoop();
         //$this->createSendLoop();