Golang code-google-com-p-go-net-websocket.NewConfig类(方法)实例源码

func TestProxyWithoutAuthorization(t *testing.T) {
	proxy := NewProxy(
		"localhost:62061",
		[]*hasher.Hasher{
			hasher.NewHasher([]string{"localhost:62032"}),
		},
		testhelpers.SuccessfulAuthorizer,
		loggertesthelper.Logger(),
	)
	go proxy.Start()
	time.Sleep(time.Millisecond * 50)

	config, err := websocket.NewConfig("ws://localhost:62061/?app=myApp", "http://localhost")
	assert.NoError(t, err)
	receivedChan := ClientWithAuth(t, "62061", "/?app=myApp", config)

	select {
	case data := <-receivedChan:
		messagetesthelpers.AssertProtoBufferMessageEquals(t, "Error: Authorization not provided", data)
	case <-time.After(1 * time.Second):
		t.Error("Did not receive response within one second")
	}

	_, stillOpen := <-receivedChan
	assert.False(t, stillOpen)
}

func New(key, secret string, currencies ...string) (*StreamingApi, error) {
	url := fmt.Sprintf("%s%s?Currency=%s", api_host, api_path, strings.Join(currencies, ","))
	config, _ := websocket.NewConfig(url, origin_url)
	ws, err := websocket.DialConfig(config)

	if err != nil {
		return nil, err
	}

	api := &StreamingApi{
		ws:     ws,
		Ticker: make(chan Ticker),
		Info:   make(chan Info),
		Depth:  make(chan Depth),
		Trade:  make(chan Trade),
		Orders: make(chan []Order),
	}

	api.key, err = hex.DecodeString(strings.Replace(key, "-", "", -1))
	if err != nil {
		return nil, err
	}

	api.secret, err = base64.StdEncoding.DecodeString(secret)
	if err != nil {
		return nil, err
	}

	return api, err
}

func Client(t *testing.T, port string, path string) chan []byte {
	config, err := websocket.NewConfig("ws://localhost:"+port+path, "http://localhost")
	config.Header.Add("Authorization", testhelpers.VALID_AUTHENTICATION_TOKEN)
	assert.NoError(t, err)

	return ClientWithAuth(t, port, path, config)
}

func (repo LoggregatorLogsRepository) connectToWebsocket(location string, app cf.Application, onConnect func(), outputChan chan *logmessage.Message, stopLoggingChan chan bool, printTimeBuffer time.Duration) (err error) {

	trace.Logger.Printf("\n%s %s\n", terminal.HeaderColor("CONNECTING TO WEBSOCKET:"), location)

	config, err := websocket.NewConfig(location, "http://localhost")
	if err != nil {
		return
	}

	config.Header.Add("Authorization", repo.config.AccessToken)
	config.TlsConfig = &tls.Config{InsecureSkipVerify: true}

	ws, err := websocket.DialConfig(config)
	if err != nil {
		return
	}

	onConnect()

	inputChan := make(chan *logmessage.Message, 1000)

	go repo.sendKeepAlive(ws)
	go repo.listenForMessages(ws, inputChan, stopLoggingChan)
	go makeAndStartMessageSorter(inputChan, outputChan, stopLoggingChan, printTimeBuffer)

	return
}

func (c *WebsocketClient) ConnectOnce() error {
	c.logger.Debug("ConnectOnce:call")
	defer c.logger.Debug("ConnectOnce:return")

	// Make websocket connection.  If this fails, either API is down or the ws
	// address is wrong.
	link := c.api.AgentLink(c.link)
	c.logger.Debug("ConnectOnce:link:" + link)
	config, err := websocket.NewConfig(link, c.api.Origin())
	if err != nil {
		return err
	}
	config.Header.Add("X-Percona-API-Key", c.api.ApiKey())

	c.logger.Debug("ConnectOnce:websocket.DialConfig")
	c.status.Update(c.name, "Connecting "+link)
	conn, err := websocket.DialConfig(config)
	if err != nil {
		return err
	}

	c.mux.Lock()
	defer c.mux.Unlock()
	c.connected = true
	c.conn = conn
	c.status.Update(c.name, "Connected "+link)

	return nil
}

func (proxy *Proxy) HandleWebSocket(clientWS *websocket.Conn) {
	req := clientWS.Request()
	req.ParseForm()
	req.Form.Get("app")
	clientAddress := clientWS.RemoteAddr()

	appId := req.Form.Get("app")

	extractAuthTokenFromUrl := func(u *url.URL) string {
		authorization := ""
		queryValues := u.Query()
		if len(queryValues["authorization"]) == 1 {
			authorization = queryValues["authorization"][0]
		}
		return authorization
	}

	authToken := clientWS.Request().Header.Get("Authorization")
	if authToken == "" {
		authToken = extractAuthTokenFromUrl(req.URL)
	}

	if authorized, errorMessage := proxy.isAuthorized(appId, authToken, clientAddress); !authorized {
		data, err := proto.Marshal(errorMessage)
		if err != nil {
			proxy.logger.Errorf("Error marshalling log message: %s", err)
		}
		websocket.Message.Send(clientWS, data)
		clientWS.Close()
		return
	}

	defer clientWS.Close()

	proxy.logger.Debugf("Output Proxy: Request for app: %v", req.Form.Get("app"))
	serverWSs := make([]*websocket.Conn, len(proxy.hashers))
	for index, hasher := range proxy.hashers {
		proxy.logger.Debugf("Output Proxy: Servers in group [%v]: %v", index, hasher.LoggregatorServers())

		server := hasher.GetLoggregatorServerForAppId(appId)
		proxy.logger.Debugf("Output Proxy: AppId is %v. Using server: %v", appId, server)

		config, err := websocket.NewConfig("ws://"+server+req.URL.RequestURI(), "http://localhost")

		if err != nil {
			proxy.logger.Errorf("Output Proxy: Error creating config for websocket - %v", err)
		}

		serverWS, err := websocket.DialConfig(config)
		if err != nil {
			proxy.logger.Errorf("Output Proxy: Error connecting to loggregator server - %v", err)
		}

		if serverWS != nil {
			serverWSs[index] = serverWS
		}
	}
	proxy.forwardIO(serverWSs, clientWS)

}

func (repo LoggregatorLogsRepository) connectToWebsocket(location string, onConnect func(), outputChan chan *logmessage.Message, stopLoggingChan chan bool, printTimeBuffer time.Duration) (err error) {
	trace.Logger.Printf("\n%s %s\n", terminal.HeaderColor("CONNECTING TO WEBSOCKET:"), location)

	config, err := websocket.NewConfig(location, "http://localhost")
	if err != nil {
		return
	}

	config.Header.Add("Authorization", repo.config.AccessToken)
	config.TlsConfig = &tls.Config{InsecureSkipVerify: true}

	ws, err := websocket.DialConfig(config)
	if err != nil {
		return
	}
	defer ws.Close()

	onConnect()

	go repo.sendKeepAlive(ws)

	inputChan := make(chan *logmessage.Message, LogBufferSize)
	stopInputChan := make(chan bool, 1)
	go func() {
		defer close(stopInputChan)
		defer close(inputChan)
		repo.listenForMessages(ws, inputChan, stopInputChan)
	}()

	messageQueue := &SortedMessageQueue{printTimeBuffer: printTimeBuffer}

	repo.processMessages(messageQueue, inputChan, outputChan, stopLoggingChan, stopInputChan)

	return
}

func New(key, secret string, currencies ...string) (*StreamingApi, error) {
	url := fmt.Sprintf("%s%s?Currency=%s", api_host, api_path, strings.Join(currencies, ","))
	config, _ := websocket.NewConfig(url, origin_url)
	ws, err := jsonws.New(config)

	if err != nil {
		return nil, err
	}

	api := &StreamingApi{
		MsgWatcher: msgwatch.New(ws.RecvForever()),
		ws:         ws,
	}

	api.key, err = hex.DecodeString(strings.Replace(key, "-", "", -1))
	if err != nil {
		return nil, err
	}

	api.secret, err = base64.StdEncoding.DecodeString(secret)
	if err != nil {
		return nil, err
	}

	remarks := api.Listen("remark")
	go func() {
		for remark := range remarks {
			fmt.Println("Remark:", remark)
		}
	}()

	return api, err
}

func TestProxyWhenAuthorizationFailsThroughQueryParams(t *testing.T) {
	proxy := NewProxy(
		"localhost:62062",
		[]*hasher.Hasher{
			hasher.NewHasher([]string{"localhost:62032"}),
		},
		testhelpers.SuccessfulAuthorizer,
		loggertesthelper.Logger(),
	)
	go proxy.Start()
	time.Sleep(time.Millisecond * 50)

	config, err := websocket.NewConfig("ws://localhost:62061/?app=myApp&authorization="+url.QueryEscape(testhelpers.INVALID_AUTHENTICATION_TOKEN), "http://localhost")
	assert.NoError(t, err)
	receivedChan := ClientWithAuth(t, "62062", "/?app=myApp", config)

	select {
	case data := <-receivedChan:
		messagetesthelpers.AssertProtoBufferMessageEquals(t, "Error: Invalid authorization", data)
	case <-time.After(1 * time.Second):
		t.Error("Did not receive response within one second")
	}

	_, stillOpen := <-receivedChan
	assert.False(t, stillOpen)
}

func newConfig(t *testing.T, path string) *websocket.Config {
	wsaddr := fmt.Sprintf("ws://%s%s", serverAddr, path)
	lh := "http://localhost"
	c, err := websocket.NewConfig(wsaddr, lh)
	if err != nil {
		t.Fatalf("NewConfig(%q, %q) got error: %s, want nil", wsaddr, lh, err.Error())
	}
	return c
}

func KeepAliveClient(t *testing.T, port string, path string) {
	config, err := websocket.NewConfig("ws://localhost:"+port+path, "http://localhost")
	config.Header.Add("Authorization", testhelpers.VALID_AUTHENTICATION_TOKEN)
	assert.NoError(t, err)
	ws, err := websocket.DialConfig(config)
	assert.NoError(t, err)

	websocket.Message.Send(ws, []byte("keep alive"))
}

// Wrapper funcs for websocket
func DialTimeout(url_, protocol, origin string, timeout time.Duration) (ws *websocket.Conn, err error) {
	config, err := websocket.NewConfig(url_, origin)
	if err != nil {
		return nil, err
	}
	if protocol != "" {
		config.Protocol = []string{protocol}
	}
	return DialConfigTimeout(config, timeout)
}

func wsConnSetup(srvAddr string) (config *websocket.Config, err error) {
	tcpAddr, err := net.ResolveTCPAddr("tcp", srvAddr)
	if err != nil {
		return nil, err
	}

	config, _ = websocket.NewConfig(fmt.Sprintf("ws://%s%s", tcpAddr, "/ws"), "http://localhost/ws")

	return
}

func (s *debugLogSuite) dialWebsocketFromURL(c *gc.C, server string, header http.Header) (*websocket.Conn, error) {
	c.Logf("dialing %v", server)
	config, err := websocket.NewConfig(server, "http://localhost/")
	c.Assert(err, gc.IsNil)
	config.Header = header
	caCerts := x509.NewCertPool()
	c.Assert(caCerts.AppendCertsFromPEM([]byte(testing.CACert)), jc.IsTrue)
	config.TlsConfig = &tls.Config{RootCAs: caCerts, ServerName: "anything"}
	return websocket.DialConfig(config)
}

func createHandler(tag, s string, info *HttpEndpointInfo) http.Handler {
	// TODO: hook gadget in as HTTP handler
	// if _, ok := flow.Registry[s]; ok {
	// 	return http.Handler(reqHandler)
	// }
	if s == "<websocket>" {
		var wsConfig *websocket.Config
		var err error
		//TODO: use wss:// and TlsConfig if wanting secure websockets outside https
		wsproto := "ws://"
		if info.uri.Scheme == "https" {
			wsproto = "wss://"
		}
		if wsConfig, err = websocket.NewConfig(wsproto+info.uri.Host+tag, info.uri.String()); err != nil {
			glog.Fatal(err)
		}

		hsfunc := func(ws *websocket.Config, req *http.Request) error {

			tag := ""
			for _, v := range ws.Protocol { //check for first supported WebSocket- (circuit) protocol
				if flow.Registry["WebSocket-"+v] != nil {
					tag = v
					break
				}
			}
			ws.Protocol = []string{tag} //let client know we picked one

			return nil //errors.New("Protocol Unsupported")
		}
		wsHandshaker := websocket.Server{Handler: wsHandler,
			Config:    *wsConfig,
			Handshake: hsfunc,
		}
		return wsHandshaker
	}

	if !strings.ContainsAny(s, "./") {
		glog.Fatalln("cannot create handler for:", s)
	}
	h := http.FileServer(http.Dir(s))
	if s != "/" {
		h = http.StripPrefix(tag, h)
	}
	if tag != "/" {
		return h
	}
	// special-cased to return main page unless the URL has an extension
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if path.Ext(r.URL.Path) == "" {
			r.URL.Path = "/"
		}
		h.ServeHTTP(w, r)
	})
}

func TestGetContainersAttachWebsocket(t *testing.T) {
	runCmd := exec.Command(dockerBinary, "run", "-dit", "busybox", "cat")
	out, _, err := runCommandWithOutput(runCmd)
	if err != nil {
		t.Fatalf(out, err)
	}
	defer deleteAllContainers()

	rwc, err := sockConn(time.Duration(10 * time.Second))
	if err != nil {
		t.Fatal(err)
	}

	cleanedContainerID := stripTrailingCharacters(out)
	config, err := websocket.NewConfig(
		"/containers/"+cleanedContainerID+"/attach/ws?stream=1&stdin=1&stdout=1&stderr=1",
		"http://localhost",
	)
	if err != nil {
		t.Fatal(err)
	}

	ws, err := websocket.NewClient(config, rwc)
	if err != nil {
		t.Fatal(err)
	}
	defer ws.Close()

	expected := []byte("hello")
	actual := make([]byte, len(expected))
	outChan := make(chan string)
	go func() {
		if _, err := ws.Read(actual); err != nil {
			t.Fatal(err)
		}
		outChan <- "done"
	}()

	inChan := make(chan string)
	go func() {
		if _, err := ws.Write(expected); err != nil {
			t.Fatal(err)
		}
		inChan <- "done"
	}()

	<-inChan
	<-outChan

	if !bytes.Equal(expected, actual) {
		t.Fatal("Expected output on websocket to match input")
	}

	logDone("container attach websocket - can echo input via cat")
}

// Create a New SocketCluster Client
func NewClient(auth AuthDetails, dbpath string) (*Client, error) {
	Info = log.New(os.Stdout, "SOCKETCLUSTER: ", log.Ltime|log.Lshortfile)

	origin := "http://localhost"
	prefix := "ws"
	if auth.SecureWS {
		prefix = "wss"
	}

	url := fmt.Sprintf("%s://%s/socketcluster/", prefix, auth.Host)

	config, _ := websocket.NewConfig(url, origin)
	config.Header.Add("User-Agent", auth.UserAgent)

	Info.Println("Connecting: " + url)
	ws, err := websocket.DialConfig(config)
	if err != nil {
		Info.Println(err)
		return nil, err
	}

	c := &Client{
		ws:       ws,
		id:       0,
		mutex:    &sync.Mutex{},
		quitChan: make(chan int),
	}

	c.setupDB(dbpath)

	// Connection succeded. Send a handshake event.
	c.emit(c.NewEvent("#handshake", makeHandshakeData()))

	rEvent, err := c.recieve()

	if err != nil {
		Info.Println(err)
		return nil, errors.New("#handshake recieve error")
	}

	// Start listening to events
	go c.listen()

	if rEvent.Rid == 1 {
		if !isAuthenticated(rEvent) {
			c.emit(c.NewEvent("clearoldsessions", makeClearOldSessionsData(auth)))

			c.loginEvent = c.NewEvent("login", makeLoginData(auth))
			c.emit(c.loginEvent)
		}
	}

	return c, nil
}

func newJWS() *JsonWebsocket {
	cfg, err := websocket.NewConfig("ws://localhost:12345/test", "http://localhost")
	if err != nil {
		panic(err)
	}
	jws, err := New(cfg)
	if err != nil {
		panic(err)
	}
	return jws
}

func AssertConnectionFails(t *testing.T, port string, path string, expectedErrorCode uint16) {
	config, err := websocket.NewConfig("ws://localhost:"+port+path, "http://localhost")
	assert.NoError(t, err)

	ws, err := websocket.DialConfig(config)
	assert.NoError(t, err)
	data := make([]byte, 2)
	_, err = ws.Read(data)
	errorCode := binary.BigEndian.Uint16(data)
	assert.Equal(t, expectedErrorCode, errorCode)
	assert.Equal(t, "EOF", err.Error())
}

func RunClient(url string, id string, userKey string) {
	rootPath, _ = filepath.Abs(rootPath)
	ListenForSignals()
	socketUrl := fmt.Sprintf("%s/clientsocket", url)
	var ws *websocket.Conn
	var timeout time.Duration = 1e8
	config, err := websocket.NewConfig(socketUrl, socketUrl)
	if err != nil {
		fmt.Println(err)
		return
	}
	config.TlsConfig = new(tls.Config)
	// Disable this when getting a proper certificate
	config.TlsConfig.InsecureSkipVerify = true
	for {
		time.Sleep(timeout)
		var err error
		ws, err = websocket.DialConfig(config)
		timeout *= 2
		if err != nil {
			fmt.Println("Could not yet connect:", err.Error(), ", trying again in", timeout)
		} else {
			break
		}
	}

	buffer, _ := json.Marshal(HelloMessage{"0.1", id, userKey})

	if _, err := ws.Write(buffer); err != nil {
		log.Fatal(err)
		return
	}
	connectUrl := strings.Replace(url, "ws://", "http://", 1)
	connectUrl = strings.Replace(connectUrl, "wss://", "https://", 1)
	multiplexer := NewRPCMultiplexer(ws, handleRequest)

	if userKey == "" {
		fmt.Print("In the Zed application copy and paste following URL to edit:\n\n")
		fmt.Printf("  %s/fs/%s\n\n", connectUrl, id)
	} else {
		fmt.Println("A Zed window should now open. If not, make sure Zed is running and configured with the correct userKey.")
	}
	fmt.Println("Press Ctrl-c to quit.")
	err = multiplexer.Multiplex()
	if err != nil {
		// TODO do this in a cleaner way (reconnect, that is)
		if err.Error() == "no-client" {
			fmt.Printf("ERROR: Your Zed editor is not currently connected to zedrem server %s.\nBe sure Zed is running and the project picker is open.\n", url)
		} else {
			RunClient(url, id, userKey)
		}
	}
}