发送buffer数据到网关
@param string $address
@param string $gateway_buffer
@return bool
335662
protected static function sendBufferToGateway($address, $gateway_buffer)
    {
        // 有$businessWorker说明是workerman环境，使用$businessWorker发送数据
        if (static::$businessWorker) {
            if (!isset(static::$businessWorker->gatewayConnections[$address])) {
                return false;
            }
            return static::$businessWorker->gatewayConnections[$address]->send($gateway_buffer, true);
        }
        // 非workerman环境
        $gateway_buffer = static::$secretKey ? static::generateAuthBuffer() . $gateway_buffer : $gateway_buffer;
        $flag           = static::$persistentConnection ? STREAM_CLIENT_PERSISTENT | STREAM_CLIENT_CONNECT : STREAM_CLIENT_CONNECT;
        $client         = stream_socket_client("tcp://$address", $errno, $errmsg, static::$connectTimeout, $flag);
        return strlen($gateway_buffer) == stream_socket_sendto($client, $gateway_buffer);
    }
191970
protected function _execute($address) {
		$this->_Geocoder = new Geocoder();
		try {
			$addresses = $this->_Geocoder->geocode($address);
		} catch (InconclusiveException $e) {
			return null;
		} catch (NotAccurateEnoughException $e) {
			return null;
		}

		if (!$addresses || $addresses->count() < 1) {
			return null;
		}
		$address = $addresses->first();
		return $address;
	}
264228
public function appendProtocolHeader(Buffer $buffer)
    {
        $buffer->append('AMQP');
        $buffer->appendUint8(0);
        $buffer->appendUint8(0);
        $buffer->appendUint8(9);
        $buffer->appendUint8(1);
    }
190552
private function isAddressValid(string $address): bool
    {
        $position = \strrpos($address, ':');
        if ($position === false) {
            return ($address[0] ?? '') === "\0"; // udg socket address.
        }

        $ip = \trim(\substr($address, 0, $position), '[]');
        $port = (int) \substr($address, $position + 1);

        return \inet_pton($ip) !== false && $port > 0 && $port < 65536;
    }
722672
function getCache($address) {
        if(!isset($this->dsn))
            return false;
        
        $_ret = array();
        
        // PEAR DB
        require_once('DB.php');          
        $_db =& DB::connect($this->dsn);
        if (PEAR::isError($_db)) {
            die($_db->getMessage());
        }
        $_res =& $_db->query("SELECT lon,lat FROM {$this->_db_cache_table} where address = ?", $address);
        if (PEAR::isError($_res)) {
            die($_res->getMessage());
        }
        if($_row = $_res->fetchRow()) {            
            $_ret['lon'] = $_row[0];
            $_ret['lat'] = $_row[1];
        }
        
        $_db->disconnect();
        
        return !empty($_ret) ? $_ret : false;
    }
830548
public static long address(final ByteBuffer buffer)
    {
        if (!buffer.isDirect())
        {
            throw new IllegalArgumentException("buffer.isDirect() must be true");
        }

        return UNSAFE.getLong(buffer, BYTE_BUFFER_ADDRESS_FIELD_OFFSET);
    }
243168
public function shouldBuffer($buffer)
    {
        if ('' === $buffer) {
            return false;
        }

        $endOfBuffer = substr($buffer, -1);
        foreach ((array) $this->search as $needle) {
            if (false !== strpos($needle, $endOfBuffer)) {
                return true;
            }
        }

        return false;
    }
891966
protected boolean populateBuffer (Buffer buffer)
    {
        if (_abuf == null) {
            _abuf = ByteBuffer.allocateDirect(getBufferSize()).order(ByteOrder.nativeOrder());
        }
        _abuf.clear();
        int read = 0;
        try {
            read = Math.max(populateBuffer(_abuf), 0);
        } catch (IOException e) {
            log.warning("Error reading audio stream [error=" + e + "].");
        }
        if (read <= 0) {
            return false;
        }
        _abuf.rewind().limit(read);
        buffer.setData(getFormat(), _abuf, getFrequency());
        return true;
    }
1149597
public function isExpectedType(Address $address) {
		$expected = $this->_config['expect'];
		if (!$expected) {
			return true;
		}

		$adminLevels = $address->getAdminLevels();
		$map = [
			static::TYPE_AAL1 => 1,
			static::TYPE_AAL2 => 2,
			static::TYPE_AAL3 => 3,
			static::TYPE_AAL4 => 4,
			static::TYPE_AAL5 => 5,
		];

		foreach ($expected as $expect) {
			switch ($expect) {
				case static::TYPE_COUNTRY:
					if ($address->getCountry() !== null) {
						return true;
					}
					break;
				case (static::TYPE_AAL1):
				case (static::TYPE_AAL2):
				case (static::TYPE_AAL3):
				case (static::TYPE_AAL4):
				case (static::TYPE_AAL5):
					if ($adminLevels->has($map[$expect])) {
						return true;
					}
					break;
				case static::TYPE_LOC:
					if ($address->getLocality() !== null) {
						return true;
					}
					break;
				case static::TYPE_SUBLOC:
					if ($address->getSubLocality() !== null) {
						return true;
					}
					break;
				case static::TYPE_POSTAL:
					if ($address->getPostalCode() !== null) {
						return true;
					}
					break;
				case static::TYPE_ADDRESS:
					if ($address->getStreetName() !== null) {
						return true;
					}
					break;
				case static::TYPE_NUMBER:
					if ($address->getStreetNumber() !== null) {
						return true;
					}
					break;
			}
		}
		return false;
	}
1353812
protected function processRules(Buffer $buffer)
    {

        // Set the result to a new result instance
        $result = new Result();

        // Named values
        $inc = -1;
        while ($buffer->getLength()) {
            // Grab the key
            $key = $buffer->readPascalString(1);

            // Make sure mutators don't overwrite each other
            if ($key === 'Mutator') {
                $key .= ++$inc;
            }

            $result->add(strtolower($key), utf8_encode($buffer->readPascalString(1)));
        }

        unset($buffer);

        return $result->fetch();
    }
1097042
public static Slice wrappedBuffer(ByteBuffer buffer)
    {
        if (buffer.isDirect()) {
            long address = bufferAddress(buffer);
            return new Slice(null, address + buffer.position(), buffer.remaining(), buffer.capacity(), buffer);
        }

        if (buffer.hasArray()) {
            return new Slice(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining());
        }

        throw new IllegalArgumentException("cannot wrap " + buffer.getClass().getName());
    }
808133
protected static function sendToAllGateway($gateway_data)
    {
        $buffer = GatewayProtocol::encode($gateway_data);
        // 如果有businessWorker实例，说明运行在workerman环境中，通过businessWorker中的长连接发送数据
        if (self::$businessWorker) {
            foreach (self::$businessWorker->gatewayConnections as $gateway_connection) {
                /** @var TcpConnection $gateway_connection */
                $gateway_connection->send($buffer, true);
            }
        } // 运行在其它环境中，通过注册中心得到gateway地址
        else {
            $all_addresses = self::getAllGatewayAddressesFromRegister();
            if (!$all_addresses) {
                throw new Exception('Gateway::getAllGatewayAddressesFromRegister() with registerAddress:' .
                    self::$registerAddress . '  return ' . var_export($all_addresses, true));
            }
            foreach ($all_addresses as $address) {
                self::sendBufferToGateway($address, $buffer);
            }
        }
    }
497976
public static boolean compact(ByteBuffer buffer) {
        if (buffer.position() == 0)
            return false;
        boolean full = buffer.limit() == buffer.capacity();
        buffer.compact().flip();
        return full && buffer.limit() < buffer.capacity();
    }
1170645
protected static function getBufferFromAllGateway($gateway_data)
    {
        $gateway_buffer = GatewayProtocol::encode($gateway_data);
        $gateway_buffer = self::$secretKey ? self::generateAuthBuffer() . $gateway_buffer : $gateway_buffer;
        if (isset(self::$businessWorker)) {
            $all_addresses = self::$businessWorker->getAllGatewayAddresses();
            if (empty($all_addresses)) {
                throw new Exception('businessWorker::getAllGatewayAddresses return empty');
            }
        } else {
            $all_addresses = self::getAllGatewayAddressesFromRegister();
            if (empty($all_addresses)) {
                return array();
            }
        }
        $client_array = $status_data = $client_address_map = $receive_buffer_array = $recv_length_array = array();
        // 批量向所有gateway进程发送请求数据
        foreach ($all_addresses as $address) {
            $client = stream_socket_client("tcp://$address", $errno, $errmsg, self::$connectTimeout);
            if ($client && strlen($gateway_buffer) === stream_socket_sendto($client, $gateway_buffer)) {
                $socket_id                        = (int)$client;
                $client_array[$socket_id]         = $client;
                $client_address_map[$socket_id]   = explode(':', $address);
                $receive_buffer_array[$socket_id] = '';
            }
        }
        // 超时5秒
        $timeout    = 5;
        $time_start = microtime(true);
        // 批量接收请求
        while (count($client_array) > 0) {
            $write = $except = array();
            $read  = $client_array;
            if (@stream_select($read, $write, $except, $timeout)) {
                foreach ($read as $client) {
                    $socket_id = (int)$client;
                    $buffer    = stream_socket_recvfrom($client, 65535);
                    if ($buffer !== '' && $buffer !== false) {
                        $receive_buffer_array[$socket_id] .= $buffer;
                        $receive_length = strlen($receive_buffer_array[$socket_id]);
                        if (empty($recv_length_array[$socket_id]) && $receive_length >= 4) {
                            $recv_length_array[$socket_id] = current(unpack('N', $receive_buffer_array[$socket_id]));
                        }
                        if (!empty($recv_length_array[$socket_id]) && $receive_length >= $recv_length_array[$socket_id] + 4) {
                            unset($client_array[$socket_id]);
                        }
                    } elseif (feof($client)) {
                        unset($client_array[$socket_id]);
                    }
                }
            }
            if (microtime(true) - $time_start > $timeout) {
                break;
            }
        }
        $format_buffer_array = array();
        foreach ($receive_buffer_array as $socket_id => $buffer) {
            $local_ip                                    = ip2long($client_address_map[$socket_id][0]);
            $local_port                                  = $client_address_map[$socket_id][1];
            $format_buffer_array[$local_ip][$local_port] = unserialize(substr($buffer, 4));
        }
        return $format_buffer_array;
    }
855532
function buffer_empty(buf) {
  for (var i = buf.length - 1; i > 0; i--) {
    if (buf[i]) { return false; }
  }

  return true;
}
415893
protected Connection getConnection(InetSocketAddress address) {
        Connection conn = null;
        try {
            // 获取连接
            conn = pool.borrowObject(address);
        } catch (FdfsException e) {
            throw e;
        } catch (Exception e) {
            LOGGER.error("Unable to borrow buffer from pool", e);
            throw new RuntimeException("Unable to borrow buffer from pool", e);
        }
        return conn;
    }
1333732
public function contains(AddressInterface $address)
    {
        $result = false;
        if ($address->getAddressType() === $this->getAddressType()) {
            if ($address->toString(false) === $this->address->toString(false)) {
                $result = true;
            }
        }

        return $result;
    }