// WebSocket adds a websocket route
966879
func (g *routeGroup) WebSocket(upgrader websocket.Upgrader, path string, h Handler) {

	handler := g.lars.wrapHandler(h)
	g.Get(path, func(c Context) {

		ctx := c.BaseContext()
		var err error

		ctx.websocket, err = upgrader.Upgrade(ctx.response, ctx.request, nil)
		if err != nil {
			return
		}

		defer ctx.websocket.Close()
		c.Next()
	}, handler)
}
821562
func (c *Client) DialSocket(socket string, transport *http.Transport) {
	dialSock := func(ctx context.Context, proto, addr string) (conn net.Conn, err error) {
		return defaultDialer.DialContext(ctx, "unix", socket)
	}
	if transport == nil {
		ht := http.DefaultTransport.(*http.Transport)
		transport = &http.Transport{
			Proxy:                 ht.Proxy,
			MaxIdleConns:          ht.MaxIdleConns,
			IdleConnTimeout:       ht.IdleConnTimeout,
			TLSHandshakeTimeout:   ht.TLSHandshakeTimeout,
			ExpectContinueTimeout: ht.ExpectContinueTimeout,
			DialContext:           dialSock,
		}
	}
	if c.Client == nil {
		// need to copy this so we don't modify the default client
		c.Client = &http.Client{
			Timeout: defaultHttpClient.Timeout,
		}
	}
	switch tp := c.Client.Transport.(type) {
	case nil, *Transport:
		c.Client.Transport = &Transport{
			RoundTripper: transport,
			Debug:        DefaultTransport.Debug,
			Output:       DefaultTransport.Output,
		}
	case *http.Transport:
		tp = transport
	default:
		panic(fmt.Sprintf("could not set DialSocket on unknown transport: %#v", tp))
	}
}
125287
func NewSocketTransport(path string) *http.Transport {
	socket := NewSocketRoundTripper(path)
	transport := &http.Transport{}
	transport.RegisterProtocol("unix", socket)
	return transport
}
400422
func (p *Poller) RemoveBySocket(soc *Socket) error {
	for id, s := range p.socks {
		if s == soc {
			return p.Remove(id)
		}
	}
	return ErrorNoSocket
}
72790
func (p DebugPrinter) WebsocketRead(typ int, content []byte, closeCode int) {
	b := &bytes.Buffer{}
	fmt.Fprintf(b, "<- Received: %s", wsMessageTypeName(typ))
	if typ == websocket.CloseMessage {
		fmt.Fprintf(b, " (%d)", closeCode)
	}
	fmt.Fprint(b, "\n")
	if len(content) > 0 {
		if typ == websocket.BinaryMessage {
			fmt.Fprintf(b, "%v\n", content)
		} else {
			fmt.Fprintf(b, "%s\n", content)
		}
	}
	fmt.Fprintf(b, "\n")
	p.logger.Logf(b.String())
}
843244
func (p *Poller) UpdateBySocket(soc *Socket, events State) (previous State, err error) {
	for id, s := range p.socks {
		if s == soc {
			previous = State(p.items[id].events)
			p.items[id].events = C.short(events)
			return previous, nil
		}
	}
	return 0, ErrorNoSocket
}
405123
func (r *Reader) handle(ws *websocket.Conn) {
	// Close the connection when the client requests it, or when we finish streaming, whichever happens first
	closeConnOnce := &sync.Once{}
	closeConn := func() {
		closeConnOnce.Do(func() {
			ws.Close()
		})
	}

	negotiated := ws.Config().Protocol
	r.selectedProtocol = negotiated[0]
	defer close(r.err)
	defer closeConn()

	go func() {
		defer runtime.HandleCrash()
		// This blocks until the connection is closed.
		// Client should not send anything.
		IgnoreReceives(ws, r.timeout)
		// Once the client closes, we should also close
		closeConn()
	}()

	r.err <- messageCopy(ws, r.r, !r.protocols[r.selectedProtocol].Binary, r.ping, r.timeout)
}
833557
func (srv *Server) AddRoute(path string, f Function) {
	if srv.rtable == nil {
		srv.rtable = make(map[string]Function)
	}
	srv.rtable[path] = f
}
510284
func CreateWebSocketServer(address string) (*WebSocketServer, error) {
	listener, err := net.Listen("tcp", address)
	if err != nil {
		return nil, err
	}

	return NewWebSocketServer(listener), nil
}
888580
func (conn *WebsocketConnection) Write(b []byte) (n int, err error) {
	conn.writeMutex.Lock()
	defer conn.writeMutex.Unlock()

	// allocate a writer
	var writer io.WriteCloser
	writer, err = conn.ws.NextWriter(websocket.BinaryMessage)
	if err != nil {
		return
	}

	// write
	n, err = writer.Write(b)
	if err != nil {
		return
	}

	// close it
	err = writer.Close()
	return
}
692580
func (xc *CommonExchange) websocket() (websocketFeed, WebsocketProcessor) {
	xc.mtx.RLock()
	defer xc.mtx.RUnlock()
	return xc.ws, xc.wsProcessor
}
842432
func (f *httpForwarder) serveWebSocket(w http.ResponseWriter, req *http.Request, ctx *handlerContext) {
	if f.log.GetLevel() >= log.DebugLevel {
		logEntry := f.log.WithField("Request", utils.DumpHttpRequest(req))
		logEntry.Debug("vulcand/oxy/forward/websocket: begin ServeHttp on request")
		defer logEntry.Debug("vulcand/oxy/forward/websocket: completed ServeHttp on request")
	}

	outReq := f.copyWebSocketRequest(req)

	dialer := websocket.DefaultDialer

	if outReq.URL.Scheme == "wss" && f.tlsClientConfig != nil {
		dialer.TLSClientConfig = f.tlsClientConfig.Clone()
		// WebSocket is only in http/1.1
		dialer.TLSClientConfig.NextProtos = []string{"http/1.1"}
	}
	targetConn, resp, err := dialer.DialContext(outReq.Context(), outReq.URL.String(), outReq.Header)
	if err != nil {
		if resp == nil {
			ctx.errHandler.ServeHTTP(w, req, err)
		} else {
			f.log.Errorf("vulcand/oxy/forward/websocket: Error dialing %q: %v with resp: %d %s", outReq.Host, err, resp.StatusCode, resp.Status)
			hijacker, ok := w.(http.Hijacker)
			if !ok {
				f.log.Errorf("vulcand/oxy/forward/websocket: %s can not be hijack", reflect.TypeOf(w))
				ctx.errHandler.ServeHTTP(w, req, err)
				return
			}

			conn, _, errHijack := hijacker.Hijack()
			if errHijack != nil {
				f.log.Errorf("vulcand/oxy/forward/websocket: Failed to hijack responseWriter")
				ctx.errHandler.ServeHTTP(w, req, errHijack)
				return
			}
			defer conn.Close()

			errWrite := resp.Write(conn)
			if errWrite != nil {
				f.log.Errorf("vulcand/oxy/forward/websocket: Failed to forward response")
				ctx.errHandler.ServeHTTP(w, req, errWrite)
				return
			}
		}
		return
	}

	// Only the targetConn choose to CheckOrigin or not
	upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool {
		return true
	}}

	utils.RemoveHeaders(resp.Header, WebsocketUpgradeHeaders...)
	utils.CopyHeaders(resp.Header, w.Header())

	underlyingConn, err := upgrader.Upgrade(w, req, resp.Header)
	if err != nil {
		f.log.Errorf("vulcand/oxy/forward/websocket: Error while upgrading connection : %v", err)
		return
	}
	defer func() {
		underlyingConn.Close()
		targetConn.Close()
		if f.websocketConnectionClosedHook != nil {
			f.websocketConnectionClosedHook(req, underlyingConn.UnderlyingConn())
		}
	}()

	errClient := make(chan error, 1)
	errBackend := make(chan error, 1)
	replicateWebsocketConn := func(dst, src *websocket.Conn, errc chan error) {

		forward := func(messageType int, reader io.Reader) error {
			writer, err := dst.NextWriter(messageType)
			if err != nil {
				return err
			}
			_, err = io.Copy(writer, reader)
			if err != nil {
				return err
			}
			return writer.Close()
		}

		src.SetPingHandler(func(data string) error {
			return forward(websocket.PingMessage, bytes.NewReader([]byte(data)))
		})

		src.SetPongHandler(func(data string) error {
			return forward(websocket.PongMessage, bytes.NewReader([]byte(data)))
		})

		for {
			msgType, reader, err := src.NextReader()

			if err != nil {
				m := websocket.FormatCloseMessage(websocket.CloseNormalClosure, fmt.Sprintf("%v", err))
				if e, ok := err.(*websocket.CloseError); ok {
					if e.Code != websocket.CloseNoStatusReceived {
						m = nil
						// Following codes are not valid on the wire so just close the
						// underlying TCP connection without sending a close frame.
						if e.Code != websocket.CloseAbnormalClosure &&
							e.Code != websocket.CloseTLSHandshake {

							m = websocket.FormatCloseMessage(e.Code, e.Text)
						}
					}
				}
				errc <- err
				if m != nil {
					forward(websocket.CloseMessage, bytes.NewReader([]byte(m)))
				}
				break
			}
			err = forward(msgType, reader)
			if err != nil {
				errc <- err
				break
			}
		}
	}

	go replicateWebsocketConn(underlyingConn, targetConn, errClient)
	go replicateWebsocketConn(targetConn, underlyingConn, errBackend)

	var message string
	select {
	case err = <-errClient:
		message = "vulcand/oxy/forward/websocket: Error when copying from backend to client: %v"
	case err = <-errBackend:
		message = "vulcand/oxy/forward/websocket: Error when copying from client to backend: %v"

	}
	if e, ok := err.(*websocket.CloseError); !ok || e.Code == websocket.CloseAbnormalClosure {
		f.log.Errorf(message, err)
	}
}
1135798
func NewWebSocketService(c *Client) *WebSocketService {
	return &WebSocketService{
		client:     c,
		chanMap:    make(map[float64]chan []float64),
		subscribes: make([]subscribeToChannel, 0),
	}
}
990586
func (ws WebSocket) ServeHTTP(w http.ResponseWriter, r *http.Request) (int, error) {
	for _, sockConfig := range ws.Sockets {
		if httpserver.Path(r.URL.Path).Matches(sockConfig.Path) {
			return serveWS(w, r, &sockConfig)
		}
	}

	// Didn't match a websocket path, so pass-through
	return ws.Next.ServeHTTP(w, r)
}
135456
func (c *connection) Write(websocketMessageType int, data []byte) error {
	// for any-case the app tries to write from different goroutines,
	// we must protect them because they're reporting that as bug...
	c.writerMu.Lock()
	if writeTimeout := c.server.config.WriteTimeout; writeTimeout > 0 {
		// set the write deadline based on the configuration
		c.underline.SetWriteDeadline(time.Now().Add(writeTimeout))
	}

	// .WriteMessage same as NextWriter and close (flush)
	err := c.underline.WriteMessage(websocketMessageType, data)
	c.writerMu.Unlock()
	if err != nil {
		// if failed then the connection is off, fire the disconnect
		c.Disconnect()
	}
	return err
}
254375
func Serve(w http.ResponseWriter, req *http.Request, handler func(ws *Conn)) {
	conn, err := websocketUpgrader.Upgrade(w, req, nil)
	if err != nil {
		logger.Errorf("problem initiating websocket: %v", err)
		return
	}
	handler(&Conn{conn})
}
765489
func NewWebSocketClient(uri ...string) (client *WebSocketClient) {
	client = new(WebSocketClient)
	client.InitBaseClient()
	client.limiter.InitLimiter()
	client.closed = false
	client.SetURIList(uri)
	client.SendAndReceive = client.sendAndReceive
	return
}