fixed socket loop null

Author: nmred

src/CommonSocket.php

@@ -91,7 +91,7 @@ abstract class CommonSocket
      */
     private $saslMechanismProvider = null;
 
-    private $loop = null;
+    protected $loop = null;
 
     /**
      * __construct

src/Consumer/State.php

@@ -24,7 +24,7 @@ class State
     const STATUS_PROCESS = 8;
     const STATUS_FINISH  = 16;
 
-    private const CLEAN_REQUEST_STATE = [
+    private static $cleanRequestState = [
         self::REQUEST_METADATA      => [],
         self::REQUEST_GETGROUP      => [],
         self::REQUEST_JOINGROUP     => [],
@@ -38,7 +38,17 @@ class State
 
     private $callStatus = [];
 
-    private $requests = self::CLEAN_REQUEST_STATE;
+	private $requests = [
+		self::REQUEST_METADATA      => [],
+		self::REQUEST_GETGROUP      => [],
+		self::REQUEST_JOINGROUP     => [],
+		self::REQUEST_SYNCGROUP     => [],
+		self::REQUEST_HEARTGROUP    => [],
+		self::REQUEST_OFFSET        => ['interval' => 2000],
+		self::REQUEST_FETCH         => ['interval' => 100],
+		self::REQUEST_FETCH_OFFSET  => ['interval' => 2000],
+		self::REQUEST_COMMIT_OFFSET => ['norepeat' => true],
+	];
 
     private $loop = null;
 
@@ -103,10 +113,10 @@ public function stop()
         $this->removeWatchers();
 
         $this->callStatus = [];
-        $this->requests   = self::CLEAN_REQUEST_STATE;
+        $this->requests   = self::$cleanRequestState;
     }
 
-    private function removeWatchers(): void
+    private function removeWatchers()
     {
         foreach (array_keys($this->requests) as $request) {
             if ($this->requests[$request]['watcher'] === null) {

src/Producer.php

@@ -1,13 +1,17 @@
 <?php
 namespace Kafka;
 
+use \Kafka\Loop;
+
 class Producer
 {
     use \Psr\Log\LoggerAwareTrait;
     use \Kafka\LoggerTrait;
 
     private $process = null;
 
+    private $loop = null;
+
     /**
      * __construct
      *
@@ -22,6 +26,7 @@ public function __construct(callable $producer = null)
         } else {
             $this->process = new \Kafka\Producer\Process($producer);
         }
+		$this->loop = Loop::getInstance();
     }
 
     /**
@@ -39,7 +44,7 @@ public function send($data = true)
         if (is_bool($data)) {
             $this->process->start();
             if ($data) {
-                \Amp\Loop::run();
+                $this->loop->run();
             }
         } else {
             return $this->process->send($data);

tests/Base/SocketTest.php

@@ -446,7 +446,6 @@ private function mockStreamSocketClient($host, $port, $config = null, $sasl = nu
             ->setMethods($mockMethod)
             ->setConstructorArgs([$host, $port, $config, $sasl])
             ->getMock();
-        $socket->loop = \Kafka\Loop::getInstance();
 
         $socket->method('createSocket')
                ->will($this->returnCallback(function ($remoteSocket, $context, &$errno, &$error) {