Get the name of the queue connection to listen on.

@param  string  $connection
@return string
327397
protected function getQueue($connection)
	{
		if (is_null($connection))
		{
			$connection = $this->flyphp['config']['queue.default'];
		}

		$queue = $this->flyphp['config']->get("queue.connections.{$connection}.queue", 'default');

		return $this->input->getOption('queue') ?: $queue;
	}
1027664
protected function parseConnection($connection)
    {
        $pos  = strpos($connection, '=');
        $name = substr($connection, 0, $pos);
        $dsn  = substr($connection, $pos + 1, strlen($connection));

        $pos  = strpos($dsn, ':');
        $adapter = substr($dsn, 0, $pos);

        $extras = [];
        foreach (explode(';', $dsn) as $element) {
            $parts = preg_split('/=/', $element);

            if (2 === count($parts)) {
                $extras[strtolower($parts[0])] = urldecode($parts[1]);
            }
        }
        $extras['adapter'] = $adapter;

        return [$name, $dsn, $extras];
    }
18059
public static function getConnectionString($key)
    {
        Validate::isString($key, 'key');

        self::_init();
        $value = null;

        foreach (self::$_sources as $source) {
            $value = call_user_func_array($source, [$key]);

            if (!empty($value)) {
                break;
            }
        }

        return $value;
    }
1203231
protected function getNextJob($connection, $queue)
    {
        try {
            foreach (explode(',', $queue) as $queue) {
                if (!is_null($job = $connection->pop($queue))) {
                    return $job;
                }
            }
        } catch (Exception $e) {
            $this->report($e);

            $this->stopWorkerIfLostConnection($e);

            $this->sleep(1);
        } catch (Throwable $e) {
            $e = new FatalThrowableError($e);
            $this->report($e);

            $this->stopWorkerIfLostConnection($e);

            $this->sleep(1);
        }
    }
183243
public static function GetConnectionName () {
		$class_name = static::class;
		if (!isset(static::$_connectionNameList[$class_name])) {
			if (defined($class_name.'::CONNECTION_NAME')) {
				$connection_name = $class_name::CONNECTION_NAME;
			} else {
				$connection_name = basename(dirname(str_replace("\\", '/', $class_name)));
			}
			static::$_connectionNameList[$class_name] = $connection_name;
		}
		return static::$_connectionNameList[$class_name];
	}
1055004
protected function getPath($connection)
	{
		if ($this->option('path')) return $this->option('path');

		if (is_null($connection))
		{
			return base_path().'/app/storage/logs/flyphp.log';
		}
		else
		{
			return $this->getRoot($connection).'/app/storage/logs/flyphp.log';
		}
	}
378655
private function getConnectionByQueue(): ConnectionInterface
    {
        if($this->queue == null){
            $this->queue = new \SplQueue();
        }
        if (!$this->queue->isEmpty()) {
            return $this->getEffectiveConnection($this->queue->count(), false);
        }

        if ($this->currentCount >= $this->poolConfig->getMaxActive()) {
            throw new ConnectionException('Connection pool queue is full');
        }

        $connect = $this->createConnection();
        $this->currentCount++;

        return $connect;
    }
40916
protected function checkSlave(Connection $connection)
    {
        $paused = 0;

        while (($lag = $this->getLag($connection)) > $this->options['max_lag']) {
            $this->paused = true;
            $for = $this->options['pause_lag'];

            $this->logger->notice('Chunk detected lag of {lag} on {database}, pausing for {for}', [
                'lag'      => $lag,
                'database' => sprintf('%s/%s', $connection->getHost(), $connection->getDatabase()),
                'for'      => $for
            ]);

            usleep($for);
            $paused += $for;

            if ($paused > $this->options['max_pause_lag']) {
                if ($this->options['continue_lag']) {
                    break;
                } else {
                    throw new RuntimeException('Slave lag did not recover after processing chunk. Aborting.');
                }
            }
        }
    }
624291
protected function prepare(Connection $connection)
	{
		$connection->setFetchMode($this->app['config']['database.fetch']);

		if ($this->app->bound('events'))
		{
			$connection->setEventDispatcher($this->app['events']);
		}

		// The database connection can also utilize a cache manager instance when cache
		// functionality is used on queries, which provides an expressive interface
		// to caching both fluent queries and FlyORM queries that are executed.
		$app = $this->app;

		$connection->setCacheManager(function() use ($app)
		{
			return $app['cache'];
		});

		// We will setup a Closure to resolve the paginator instance on the connection
		// since the Paginator isn't sued on every request and needs quite a few of
		// our dependencies. It'll be more efficient to lazily resolve instances.
		$connection->setPaginator(function() use ($app)
		{
			return $app['paginator'];
		});

		return $connection;
	}
1163062
public function connection($connection = null)
    {
        if ($connection === null) {
            if ($this->connection === null) {
                $this->connection = ConnectionManager::get($this->options['connection']);
            }
            return $this->connection;
        }

        return $this->connection = $connection;
    }
409773
public function getPresenceVerifierFor($connection)
    {
        return tap($this->getPresenceVerifier(), function ($verifier) use ($connection) {
            $verifier->setConnection($connection);
        });
    }
1250458
protected function getPath($connection)
    {
        if ($this->option('path')) {
            return $this->option('path');
        }

        if (is_null($connection) && $this->option('path')) {
            return storage_path($this->option('path'));
        } elseif (is_null($connection) && !$this->option('path')) {
            return storage_path('logs');
        }

        return $this->getRoot($connection).str_replace(base_path(), '', storage_path('logs'));
    }
877649
protected function runWorker($connection, $queue)
    {
        foreach (explode(',', $queue) as $tube) {
            $this->worker->eyeQueues[] = Queue::firstOrCreate(['connection' => $connection,
                                                               'tube' => $tube,
                                                               'driver' => $this->getDriverName($connection),
                                                               'healthy' => 1]);
        }

        return parent::runWorker($connection, $queue);
    }
475651
protected function checkConnections($connection)
    {
        $connections = $connection->getConnectionManager()->count();

        if ($connections > $this->options['connections']) {
            $this->limit($connection, 'Max connections');
        }
    }
955841
protected function selectSQLDriver($connection)
    {
        switch (strtolower($connection)) {
            case 'mysql':
                $this->db = [
                    'driver'   => 'pdo_mysql',
                    'host'     => env('DB_HOST', '127.0.0.1'),
                    'port'     => env('DB_PORT', '3306'),
                    'dbname'   => env('DB_DATABASE', 'teamelf'),
                    'user'     => env('DB_USERNAME', 'teamelf'),
                    'password' => env('DB_PASSWORD', 'teamelf'),
                    'charset'  => 'UTF8'
                ];
                break;
            default:
                throw new Exception($connection . 'has not been supported yet! use [mysql] instead!');
        }
    }
1119794
public function return_connection($connection) {
        if ($connection->op_count >= $this->recycle) {
            $this->stats['recycled'] += 1;
            $connection->close();
            $this->make_conn();
            $connection = $this->get();
        }
        array_push($this->queue, $connection);
    }
1056464
public function onMessage($connection, $buffer)
    {
        if ($buffer === 'ping') {
            return;
        }
        $data = unserialize($buffer);
        if (!$buffer || !isset($data['cmd']) || !isset($data['key'])) {
            return $connection->close(serialize('bad request'));
        }
        $cmd = $data['cmd'];
        $key = $data['key'];
        switch ($cmd) {
            case 'get':
                if (!isset($this->_dataArray[$key])) {
                    return $connection->send('N;');
                }
                return $connection->send(serialize($this->_dataArray[$key]));
                break;
            case 'set':
                $this->_dataArray[$key] = $data['value'];
                $connection->send('b:1;');
                break;
            case 'add':
                if (isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                $this->_dataArray[$key] = $data['value'];
                return $connection->send('b:1;');
                break;
            case 'increment':
                if (!isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                if (!is_numeric($this->_dataArray[$key])) {
                    $this->_dataArray[$key] = 0;
                }
                $this->_dataArray[$key] = $this->_dataArray[$key] + $data['step'];
                return $connection->send(serialize($this->_dataArray[$key]));
                break;
            case 'cas':
                if (isset($this->_dataArray[$key]) && md5(serialize($this->_dataArray[$key])) === $data['md5']) {
                    $this->_dataArray[$key] = $data['value'];
                    return $connection->send('b:1;');
                }
                $connection->send('b:0;');
                break;
            case 'push':
                if (!isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                if (!is_array($this->_dataArray[$key])) {
                    $this->_dataArray[$key] = array();
                }
                return $connection->send(serialize(array_push($this->_dataArray[$key], $data['value'])));
                break;
            case 'pushIfNotExist':
                if (!isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                if (!is_array($this->_dataArray[$key])) {
                    $this->_dataArray[$key] = array();
                }
                if (in_array($data['value'], $this->_dataArray[$key])) {
                    return $connection->send('b:1;');
                }
                return $connection->send(serialize(array_push($this->_dataArray[$key], $data['value'])));
                break;
            case 'pushToLeftIfNotExist':
                if (!isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                if (!is_array($this->_dataArray[$key])) {
                    $this->_dataArray[$key] = array();
                }
                if (in_array($data['value'], $this->_dataArray[$key])) {
                    return $connection->send('b:1;');
                }
                return $connection->send(serialize(array_unshift($this->_dataArray[$key], $data['value'])));
                break;
            case 'shift':
                if (!isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                if (!is_array($this->_dataArray[$key])) {
                    $this->_dataArray[$key] = array();
                }
                return $connection->send(serialize(array_shift($this->_dataArray[$key])));
                break;
            case 'pop':
                if (!isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                if (!is_array($this->_dataArray[$key])) {
                    $this->_dataArray[$key] = array();
                }
                return $connection->send(serialize(array_pop($this->_dataArray[$key])));
                break;
            case 'count':
                if (!isset($this->_dataArray[$key]) || !is_array($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                return $connection->send(serialize(count($this->_dataArray[$key])));
                break;
            case 'rand':
                if (!isset($this->_dataArray[$key]) || !is_array($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                return $connection->send(serialize($this->_dataArray[$key][array_rand($this->_dataArray[$key])]));
                break;
            case 'in':
                if (!isset($this->_dataArray[$key]) || !is_array($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                return $connection->send(serialize(in_array($data['value'], $this->_dataArray[$key])));
                break;
            case 'delete':
                unset($this->_dataArray[$key]);
                $connection->send('b:1;');
                break;
            case 'bfNew':
                if (isset($this->_dataArray[$key])) {
                    return $connection->send('b:0;');
                }
                $m = isset($data['value'][0]) ? $data['value'][0] : 40000;
                $k = isset($data['value'][1]) ? $data['value'][1] : 14;
                $this->_dataArray[$key] = new BloomFilter($m, $k);
                return $connection->send('b:1;');
                break;
            case 'bfAdd':
                if (!isset($this->_dataArray[$key]) || !($this->_dataArray[$key] instanceof BloomFilter)) {
                    return $connection->send('b:0;');
                }
                $this->_dataArray[$key]->add($data['value']);
                return $connection->send('b:1;');
                break;
            case 'bfIn':
                if (!isset($this->_dataArray[$key]) || !($this->_dataArray[$key] instanceof BloomFilter)) {
                    return $connection->send('b:0;');
                }
                return $connection->send(serialize($this->_dataArray[$key]->maybeInSet($data['value'])));
                break;
            default:
                $connection->close(serialize('bad cmd ' . $cmd));
        }
    }