Creates a subscriber socket.

@param string       $mode
@param string       $subscribe
@param string|array $dsn
@param array        $options

@return Socket
341413
public function createSubscriber($mode, $dsn = null, $subscribe = '', $options = array())
    {
        $options[ZMQ::SOCKOPT_SUBSCRIBE] = $subscribe;

        return $this->createSocket(ZMQ::SOCKET_SUB, $mode, $dsn, $options);
    }
499285
public function createConnection($dsn, array $config, array $options)
    {
        [$username, $password] = [
            $config['username'] ?? null, $config['password'] ?? null,
        ];

        try {
            return $this->createPdoConnection(
                $dsn, $username, $password, $options
            );
        } catch (Exception $e) {
            return $this->tryAgainIfCausedByLostConnection(
                $e, $dsn, $username, $password, $options
            );
        }
    }
1058512
protected function createPdoConnection($dsn, $username, $password, $options)
    {
        if (class_exists(PDOConnection::class) && ! $this->isPersistentConnection($options)) {
            return new PDOConnection($dsn, $username, $password, $options);
        }

        return new PDO($dsn, $username, $password, $options);
    }
701314
public function createReply($mode, $dsn = null, $options = array())
    {
        return $this->createSocket(ZMQ::SOCKET_REP, $mode, $dsn, $options);
    }
633696
public function add_subscriber( EventSubscriber $subscriber ) {

		foreach ( $subscriber->get_subscribed_events() as $event => $params ) {

			if ( is_string( $params ) ) {
				$this->add_listener( $event, array( $subscriber, $params ) );
			} elseif ( is_array( $params ) && isset( $params[0] ) ) {

				$method        = $params[0];
				$priority      = isset( $params[1] ) ? $params[1] : self::PRIORITY;
				$accepted_args = isset( $params[2] ) ? $params[2] : self::ACCEPTED;

				$this->add_listener( $event, array( $subscriber, $method ), $priority, $accepted_args );
			} else {
				throw new \InvalidArgumentException( 'Invalid return format from EventSubscriber.' );
			}
		}

		return $this;
	}
1069550
public function subscriberByRequest(array $input, array $headers): ?Subscriber
    {
        $channel = Arr::get($input, 'channel_name');

        return $channel
            ? $this->subscriberByChannel($channel)
            : null;
    }
836661
function subscribe(conn, channels) {
  return this.doSubscribe(
    PUBSUB.SUBSCRIBE,
    PUBSUB.CHANNELS,
    this._channels, conn, channels);
}
241905
public static function Connect ($dsn) {
		try {
			$dsn = static::AdjustDsn($dsn);

			static::ValidateDsn($dsn);

			return new static (
				static::MakeDsn($dsn),
				$dsn['username'] ?? null,
				$dsn['password'] ?? null,
				static::MakeOptionList($dsn['options'] ?? [])
			);
		} catch (\PDOException $pdo_e) {
 			throw new \Exception(sprintf(
 				'%s dsn=%s user=%s password=%s options=%s',
 				$pdo_e->getMessage(),
 				static::MakeDsn($dsn),
 				$dsn['username'] ?? '',
 				$dsn['password'] ?? '',
 				implode(', ', static::MakeOptionList($dsn['options'] ?? []))
 			), 0, $pdo_e);
		}
	}
387255
public function channelsSubscriptions($channel, $options = [], $token = null)
    {
        $availableOptions = ['limit', 'offset', 'direction'];

        return $this->sendRequest('GET', 'channels/'.$channel.'/subscriptions', $this->getToken($token), $options, $availableOptions);
    }
933377
public function createConnection($dsn, array $config, array $options)
    {
        list($username, $password) = [
            Arr::get($config, 'username'), Arr::get($config, 'password'),
        ];

        try {
            return $this->createPdoConnection(
                $dsn,
                $username,
                $password,
                $options
            );
        } catch (Exception $e) {
            return $this->tryAgainIfCausedByLostConnection(
                $e,
                $dsn,
                $username,
                $password,
                $options
            );
        }
    }
858189
public function newSubscription(string $subscription, string $plan): SubscriptionBuilder
    {
        return new SubscriptionBuilder($this, $subscription, $plan);
    }
1053526
protected function mergeOptions($src, $dst, &$options) {
    if (!isset($options['width']) && !isset($options['height'])) {
      $info = $this->getVideoInfo($src);
      $options['width'] = $info['width'];
      $options['height'] = $info['height'];
      if (!$info['audioStreams']) {
        $options['audio'] = false;
      }
    }
    if (!isset($options['targetFormat'])) {
      $ext = strtolower(substr($dst, strrpos($dst, '.') + 1));
      $options['targetFormat'] = $ext;
    }
  }
355027
protected function setupMongoDB($dsn, $options)
    {
        try {
            $this->client = new \MongoDB\Client($dsn, $options);
            $this->dbh    = $this->client->selectDatabase($this->dbName);
        } catch (\MongoDB\Driver\Exception $e) {
            throw new ModuleException($this, sprintf('Failed to open Mongo connection: %s', $e->getMessage()));
        }
    }
1014873
function register(array $options = array())
    {
        if ($this->alive()) {
            return $this->renew($options);
        } else {
            return $this->subscribe($options);
        }
    }
940174
public static function tcp(string $host, array $options = []) : Socket
    {
        return self::create($host, \array_merge($options, ['transport' => 'tcp']));
    }
315238
public function createPublisher($mode, $dsn = null, $options = array())
    {
        return $this->createSocket(ZMQ::SOCKET_PUB, $mode, $dsn, $options);
    }
1025974
protected function configure(array $options): void
    {
        $options = array_merge([
            'socket_class' => ClientSocket::class,
            'on_data_callback' => null,
            'socket_options' => [],
        ], $options);

        parent::configure($options);
    }