public function connectToNSQD($host, $port)
    {
        if ($this->isStopped) {
            throw new NsqException("consumer stopped");
        }

        $addr = "$host:$port";
        if (!isset($this->nsqdTCPAddrsConnNum[$addr])) {
            $this->nsqdTCPAddrsConnNum[$addr] = 0;
        }

        $perNsqdMaxNum = $this->getMaxConnNumPerNsqd();
        $currentNum = $this->nsqdTCPAddrsConnNum[$addr];
        $remainNum = max($perNsqdMaxNum - $currentNum, 0);

        $nsqdConns = [];
        for ($i = 0; $i < $remainNum; $i++) {
            try {
                $this->nsqdTCPAddrsConnNum[$addr]++;

                $conn = new Connection($host, $port);
                $hash = spl_object_hash($conn);

                try {
                    $this->pendingConnections[$hash] = $conn;
                    yield $conn->connect();
                    $this->idleConnections[$hash] = $conn;
                } finally {
                    unset($this->pendingConnections[$hash]);
                }

                $nsqdConns[] = $conn;
            } catch (\Throwable $t) {
                $this->nsqdTCPAddrsConnNum[$addr]--;
                throw $t;
            } catch (\Exception $ex) {
                $this->nsqdTCPAddrsConnNum[$addr]--;
                throw $ex;
            }
        }
        yield $nsqdConns;
    }
public function connectToNSQD($host, $port)
    {
        if ($this->isStopped) {
            throw new NsqException("consumer stopped");
        }

        $addr = "$host:$port";
        if (!isset($this->nsqdTCPAddrsConnNum[$addr])) {
            $this->nsqdTCPAddrsConnNum[$addr] = 0;
        }

        $perNsqdMaxNum = $this->getMaxConnNumPerNsqd();
        $currentNum = $this->nsqdTCPAddrsConnNum[$addr];
        $remainNum = max($perNsqdMaxNum - $currentNum, 0);

        $nsqdConns = [];
        for ($i = 0; $i < $remainNum; $i++) {
            try {
                $this->nsqdTCPAddrsConnNum[$addr]++;

                $conn = new Connection($host, $port);
                $hash = spl_object_hash($conn);

                try {
                    $this->pendingConnections[$hash] = $conn;
                    yield $conn->connect();
                    $this->idleConnections[$hash] = $conn;
                } finally {
                    unset($this->pendingConnections[$hash]);
                }

                $nsqdConns[] = $conn;
            } catch (\Throwable $t) {
                $this->nsqdTCPAddrsConnNum[$addr]--;
                throw $t;
            } catch (\Exception $ex) {
                $this->nsqdTCPAddrsConnNum[$addr]--;
                throw $ex;
            }
        }
        yield $nsqdConns;
    }
public function connectToNSQD ( $ host , $ port ) { if ( $ this -> isStopped ) { throw new NsqException ( "consumer stopped" ) ; } $ addr = "$host:$port" ; if ( ! isset ( $ this -> nsqdTCPAddrsConnNum [ $ addr ] ) ) { $ this -> nsqdTCPAddrsConnNum [ $ addr ] = 0 ; } $ perNsqdMaxNum = $ this -> getMaxConnNumPerNsqd ( ) ; $ currentNum = $ this -> nsqdTCPAddrsConnNum [ $ addr ] ; $ remainNum = max ( $ perNsqdMaxNum - $ currentNum , 0 ) ; $ nsqdConns = [ ] ; for ( $ i = 0 ; $ i < $ remainNum ; $ i ++ ) { try { $ this -> nsqdTCPAddrsConnNum [ $ addr ] ++ ; $ conn = new Connection ( $ host , $ port ) ; $ hash = spl_object_hash ( $ conn ) ; try { $ this -> pendingConnections [ $ hash ] = $ conn ; yield $ conn -> connect ( ) ; $ this -> idleConnections [ $ hash ] = $ conn ; } finally { unset ( $ this -> pendingConnections [ $ hash ] ) ; } $ nsqdConns [ ] = $ conn ; } catch ( \ Throwable $ t ) { $ this -> nsqdTCPAddrsConnNum [ $ addr ] -- ; throw $ t ; } catch ( \ Exception $ ex ) { $ this -> nsqdTCPAddrsConnNum [ $ addr ] -- ; throw $ ex ; } } yield $ nsqdConns ; }
ConnectToNSQD takes a nsqd address to connect directly to.
It is recommended to use ConnectToNSQLookupd so that topics are discovered
automatically.  This method is useful when you want to connect to a single, local, instance.
@param string $host
@param string $port
@return mixed yield List<Connection>
@throws NsqException
@throws \Exception
@throws \Throwable
ConnectToNSQD takes a nsqd address to connect directly to . It is recommended to use ConnectToNSQLookupd so that topics are discovered automatically . This method is useful when you want to connect to a single local instance .