Construct a WebSocket URL.
 @example Construct
   url = LibWebSocket::URL.new
   url.host = 'example.com'
   url.port = '3000'
   url.secure = true
   url.to_s # => 'wss://example.com:3000'
515284
def to_s
      string = ''

      string += 'ws'
      string += 's' if self.secure
      string += '://'
      string += self.host
      string += ':' + self.port.to_s if self.port
      string += self.resource_name || '/'

      return string
    end
1219636
function Socket(uri, opts){
  if (!(this instanceof Socket)) return new Socket(uri, opts);

  opts = opts || {};

  if ('object' == typeof uri) {
    opts = uri;
    uri = null;
  }

  if (uri) {
    uri = util.parseUri(uri);
    opts.host = uri.host;
    opts.secure = uri.protocol == 'https' || uri.protocol == 'wss';
    opts.port = uri.port;
    if (uri.query) opts.query = uri.query;
  }

  this.secure = null != opts.secure ? opts.secure :
    (global.location && 'https:' == location.protocol);

  if (opts.host) {
    var pieces = opts.host.split(':');
    opts.hostname = pieces.shift();
    if (pieces.length) opts.port = pieces.pop();
  }

  this.hostname = opts.hostname ||
    (global.location ? location.hostname : 'localhost');
  this.port = opts.port || (global.location && location.port ?
       location.port :
       (this.secure ? 443 : 80));
  this.query = opts.query || {};
  if ('string' == typeof this.query) this.query = util.qsParse(this.query);
  this.upgrade = false !== opts.upgrade;
  this.path = (opts.path || '/engine.io').replace(/\/$/, '') + '/';
  this.forceJSONP = !!opts.forceJSONP;
  this.timestampParam = opts.timestampParam || 't';
  this.timestampRequests = !!opts.timestampRequests;
  this.flashPath = opts.flashPath || '';
  this.transports = opts.transports || ['polling', 'websocket', 'flashsocket'];
  this.readyState = '';
  this.writeBuffer = [];
  this.callbackBuffer = [];
  this.policyPort = opts.policyPort || 843;
  this.open();

  Socket.sockets.push(this);
  Socket.sockets.evs.emit('add', this);
}
61154
def url_for(params)
      host_with_port = host || location.host
      host_with_port += ":#{port}" if port && port != 80
      scheme = scheme || location.scheme

      unless RUBY_PLATFORM == 'opal'
        # lazy load routes and views on the server
        if !@router && @routes_loader
          # Load the templates
          @routes_loader.call
        end
      end

      path, params = @router.params_to_url(params)

      if path == nil
        raise "No route matched, make sure you have the base route defined last: `client '/', {}`"
      end

      new_url    = "#{scheme}://#{host_with_port}#{path.chomp('/')}"

      # Add query params
      params_str = ''
      unless params.empty?
        query_parts = []
        nested_params_hash(params).each_pair do |key, value|
          # remove the _ from the front
          value = Volt::Parsing.encodeURI(value)
          query_parts << "#{key}=#{value}"
        end

        if query_parts.size > 0
          query = query_parts.join('&')
          new_url += "?#{query}"
        end
      end

      # Add fragment
      frag    = fragment
      new_url += '#' + frag if frag.present?

      new_url
    end
528638
def _build_base_url(self, host, port):
        
        parsed = urlparse(host)
        if not parsed.scheme:
            parsed = parsed._replace(scheme="http")
            parsed = parsed._replace(path="")
            netloc, parts = host, host.partition(":")
            if parts[1] == "" and port is not None:
                netloc = "{}:{}".format(parts[0], port)
            parsed = parsed._replace(netloc=netloc)
        parsed = parsed._replace(path=parsed.path.rstrip("/"))
        return parsed.geturl()
1301952
def ws_connect(
            self,
            url: StrOrURL, *,
            method: str=hdrs.METH_GET,
            protocols: Iterable[str]=(),
            timeout: float=10.0,
            receive_timeout: Optional[float]=None,
            autoclose: bool=True,
            autoping: bool=True,
            heartbeat: Optional[float]=None,
            auth: Optional[BasicAuth]=None,
            origin: Optional[str]=None,
            headers: Optional[LooseHeaders]=None,
            proxy: Optional[StrOrURL]=None,
            proxy_auth: Optional[BasicAuth]=None,
            ssl: Union[SSLContext, bool, None, Fingerprint]=None,
            proxy_headers: Optional[LooseHeaders]=None,
            compress: int=0,
            max_msg_size: int=4*1024*1024) -> '_WSRequestContextManager':
        """"""
        return _WSRequestContextManager(
            self._ws_connect(url,
                             method=method,
                             protocols=protocols,
                             timeout=timeout,
                             receive_timeout=receive_timeout,
                             autoclose=autoclose,
                             autoping=autoping,
                             heartbeat=heartbeat,
                             auth=auth,
                             origin=origin,
                             headers=headers,
                             proxy=proxy,
                             proxy_auth=proxy_auth,
                             ssl=ssl,
                             proxy_headers=proxy_headers,
                             compress=compress,
                             max_msg_size=max_msg_size))
812996
def build_url_parts (self):
        
        # check userinfo@host:port syntax
        self.userinfo, host = urllib.splituser(self.urlparts[1])
        port = urlutil.default_ports.get(self.scheme, 0)
        host, port = urlutil.splitport(host, port=port)
        if port is None:
            raise LinkCheckerError(_("URL host %(host)r has invalid port") %
                    {"host": host})
        self.port = port
        # set host lowercase
        self.host = host.lower()
        if self.scheme in scheme_requires_host:
            if not self.host:
                raise LinkCheckerError(_("URL has empty hostname"))
            self.check_obfuscated_ip()
        if not self.port or self.port == urlutil.default_ports.get(self.scheme):
            host = self.host
        else:
            host = "%s:%d" % (self.host, self.port)
        if self.userinfo:
            self.urlparts[1] = "%s@%s" % (self.userinfo, host)
        else:
            self.urlparts[1] = host
        # safe anchor for later checking
        self.anchor = self.urlparts[4]
        if self.anchor is not None:
            assert isinstance(self.anchor, unicode), repr(self.anchor)
262615
def _build_url(self, host, handler):
        
        scheme = 'https' if self.use_https else 'http'
        return '%s://%s/%s' % (scheme, host, handler)
440942
def build_http_object(uri)
      http = Net::HTTP.new(uri.host, uri.port)
      http.use_ssl = true if uri.scheme == 'https'
      if @ssl_enabled
        http.cert_store = @cert_store if @cert_store
      else http.verify_mode = OpenSSL::SSL::VERIFY_NONE
      end
      http.read_timeout = @timeout if @timeout # Timeout for a request
      http.open_timeout = @timeout if @timeout # Timeout for a connection
      http
    end
1000997
def http_construct(args, unknown):
    

    str_http    = ''
    b_httpSpecd = False

    if '--http' in unknown:
        try:
            str_httpArg = unknown[unknown.index('--http')+1]
            unknown.remove('--http')
            unknown.remove(str_httpArg)
        except:
            str_httpArg = ""
        str_http    = '--http %s' % str_httpArg
        b_httpSpecd = True

    if not b_httpSpecd:
        str_serverIP  = "172.17.0.2"
        str_serverPort  = '5010'
        try:
            if args.b_pman:
                str_serverIP    = os.environ['PMAN_PORT_5010_TCP_ADDR']
                str_serverPort  = os.environ['PMAN_PORT_5010_TCP_PORT']
            if args.b_pfioh:
                str_serverIP    = os.environ['PFIOH_PORT_5055_TCP_ADDR']
                str_serverPort  = os.environ['PFIOH_PORT_5055_TCP_PORT']
        except:
            pass
        str_http    = '--http %s:%s/api/v1/cmd/' % (str_serverIP, str_serverPort)

    return str_http
342598
def base_ws_uri():
    

    scheme = config['wsserver']['advertised_scheme']
    host = config['wsserver']['advertised_host']
    port = config['wsserver']['advertised_port']
    return '{}://{}:{}'.format(scheme, host, port)
74808
def route(self, uri, methods=frozenset({'GET'}), host=None,
              strict_slashes=False, stream=False, websocket=False):
        '''
        '''

        # Fix case where the user did not prefix the URL with a /
        # and will probably get confused as to why it's not working
        if not uri.startswith('/'):
            uri = '/' + uri

        def response(handler):
            if websocket:
                handler.is_websocket = True
            elif stream:
                handler.is_stream = True
            self.router.add(uri=uri, methods=methods, handler=handler,
                            host=host, strict_slashes=strict_slashes)
            return handler

        return response
180837
def build_url(self):
        
        url = "%s/1.0/%s/%s/%s" % (
            self.request.endpoint.get_url(),
            self.request.api_object,
            self.request.api_method,
            to_hex(self.request.nonce)
        )
        return url
891827
async def create_websocket_client(sock: anyio.abc.SocketStream,
                                  addr,
                                  path: str,
                                  headers: Optional[List] = None,
                                  subprotocols: Optional[List[str]] = None):
    
    ws = Websocket()
    await ws.start_client(
        sock, addr=addr, path=path, headers=headers, subprotocols=subprotocols)
    return ws
797195
def http(&block)
      http = Net::HTTP.new(@uri.host, @uri.port)
      if @uri.port == 443
        http.use_ssl = true
        http.verify_mode = OpenSSL::SSL::VERIFY_NONE
      end
      return http.start(&block) if block_given?
      http
    end
517591
def server_url
      @server_url ||= begin
        scheme  = request.env["rack.url_scheme"]
        address = request.env["SERVER_NAME"]
        port    = request.env["SERVER_PORT"]

        Pathname.new("#{scheme}://#{address}:#{port}")
      end
    end
198488
def build_socket(self):
        
        try:
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.sock.settimeout(self.SOCKET_TIMEOUT)
            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            # Check if TLS
            if self.request_object.protocol == 'https':
                self.sock = ssl.wrap_socket(self.sock, ciphers=self.CIPHERS)
            self.sock.connect(
                (self.request_object.dest_addr, self.request_object.port))
        except socket.error as msg:
            raise errors.TestError(
                'Failed to connect to server',
                {
                    'host': self.request_object.dest_addr,
                    'port': self.request_object.port,
                    'proto': self.request_object.protocol,
                    'message': msg,
                    'function': 'http.HttpUA.build_socket'
                })
1364855
protected String getWebSocketURL(ChannelHandlerContext ctx, HttpRequest req) {
        boolean isSecure = ctx.pipeline().get(SslHandler.class) != null;
        return (isSecure ? SCHEME_SECURE_WEBSOCKET : SCHEME_WEBSOCKET) + req.getHeaders().get(HttpHeaderNames.HOST) + req.getUri() ;
    }