// NatsConn returns the underlying NATS conn. Use this with care. For example,
// closing the wrapped NATS conn will put the NATS Streaming Conn in an invalid
// state.
351231
func (sc *conn) NatsConn() *nats.Conn {
	sc.RLock()
	nc := sc.nc
	sc.RUnlock()
	return nc
}
254682
func (c *Client) Conn() *grpc.ClientConn {
	c.connMu.Lock()
	defer c.connMu.Unlock()
	return c.conn
}
1060794
func NewConn(conn io.ReadWriteCloser, opts ...ConnOption) (*Conn, error) {
	return newConn(genericTransport{conn}, opts...)
}
835340
func NewNetConn(conn net.Conn, maxWriteDelay time.Duration) *NetConn {
	return &NetConn{
		BaseConn: NewBaseConn(conn, maxWriteDelay),
		conn:     conn,
	}
}
472040
func (r *rpcHandler) handleConn(ctx context.Context, conn net.Conn, rpcCtx *RPCContext) {
	// Read a single byte
	buf := make([]byte, 1)
	if _, err := conn.Read(buf); err != nil {
		if err != io.EOF {
			r.logger.Error("failed to read first RPC byte", "error", err)
		}
		conn.Close()
		return
	}

	// Enforce TLS if EnableRPC is set
	if r.config.TLSConfig.EnableRPC && !rpcCtx.TLS && pool.RPCType(buf[0]) != pool.RpcTLS {
		if !r.config.TLSConfig.RPCUpgradeMode {
			r.logger.Warn("non-TLS connection attempted with RequireTLS set", "remote_addr", conn.RemoteAddr())
			conn.Close()
			return
		}
	}

	// Switch on the byte
	switch pool.RPCType(buf[0]) {
	case pool.RpcNomad:
		// Create an RPC Server and handle the request
		server := rpc.NewServer()
		r.setupRpcServer(server, rpcCtx)
		r.handleNomadConn(ctx, conn, server)

		// Remove any potential mapping between a NodeID to this connection and
		// close the underlying connection.
		r.removeNodeConn(rpcCtx)

	case pool.RpcRaft:
		metrics.IncrCounter([]string{"nomad", "rpc", "raft_handoff"}, 1)
		r.raftLayer.Handoff(ctx, conn)

	case pool.RpcMultiplex:
		r.handleMultiplex(ctx, conn, rpcCtx)

	case pool.RpcTLS:
		if r.rpcTLS == nil {
			r.logger.Warn("TLS connection attempted, server not configured for TLS")
			conn.Close()
			return
		}
		conn = tls.Server(conn, r.rpcTLS)

		// Force a handshake so we can get information about the TLS connection
		// state.
		tlsConn, ok := conn.(*tls.Conn)
		if !ok {
			r.logger.Error("expected TLS connection", "got", log.Fmt("%T", conn))
			conn.Close()
			return
		}

		if err := tlsConn.Handshake(); err != nil {
			r.logger.Warn("failed TLS handshake", "remote_addr", tlsConn.RemoteAddr(), "error", err)
			conn.Close()
			return
		}

		// Update the connection context with the fact that the connection is
		// using TLS
		rpcCtx.TLS = true

		// Store the verified chains so they can be inspected later.
		state := tlsConn.ConnectionState()
		rpcCtx.VerifiedChains = state.VerifiedChains

		r.handleConn(ctx, conn, rpcCtx)

	case pool.RpcStreaming:
		r.handleStreamingConn(conn)

	case pool.RpcMultiplexV2:
		r.handleMultiplexV2(ctx, conn, rpcCtx)

	default:
		r.logger.Error("unrecognized RPC byte", "byte", buf[0])
		conn.Close()
		return
	}
}
473978
func (s *StanServer) createNatsClientConn(name string) (*nats.Conn, error) {
	var err error
	ncOpts := nats.DefaultOptions

	for _, o := range s.opts.NATSClientOpts {
		o(&ncOpts)
	}

	ncOpts.Servers, err = s.buildServerURLs()
	if err != nil {
		return nil, err
	}
	ncOpts.User = s.natsOpts.Username
	ncOpts.Password = s.natsOpts.Password
	ncOpts.Token = s.natsOpts.Authorization

	ncOpts.Name = fmt.Sprintf("_NSS-%s-%s", s.opts.ID, name)

	if err = nats.ErrorHandler(s.stanErrorHandler)(&ncOpts); err != nil {
		return nil, err
	}
	if err = nats.ReconnectHandler(s.stanReconnectedHandler)(&ncOpts); err != nil {
		return nil, err
	}
	if err = nats.ClosedHandler(s.stanClosedHandler)(&ncOpts); err != nil {
		return nil, err
	}
	if err = nats.DisconnectHandler(s.stanDisconnectedHandler)(&ncOpts); err != nil {
		return nil, err
	}
	if s.opts.Secure {
		if err = nats.Secure()(&ncOpts); err != nil {
			return nil, err
		}
	}
	if s.opts.ClientCA != "" {
		if err = nats.RootCAs(s.opts.ClientCA)(&ncOpts); err != nil {
			return nil, err
		}
	}
	if s.opts.ClientCert != "" {
		if err = nats.ClientCert(s.opts.ClientCert, s.opts.ClientKey)(&ncOpts); err != nil {
			return nil, err
		}
	}
	// Shorten the time we wait to try to reconnect.
	// Don't make it too often because it may exhaust the number of FDs.
	ncOpts.ReconnectWait = 250 * time.Millisecond
	// Make it try to reconnect for ever.
	ncOpts.MaxReconnect = -1
	// To avoid possible duplicate redeliveries, etc.., set the reconnect
	// buffer to -1 to avoid any buffering in the nats library and flush
	// on reconnect.
	ncOpts.ReconnectBufSize = -1

	var nc *nats.Conn
	if nc, err = ncOpts.Connect(); err != nil {
		return nil, err
	}
	return nc, err
}
1032172
func Conn() *MockConn {
	return &MockConn{
		first:    true,
		Written:  make([][]byte, 0, 4),
		readings: make([][]byte, 0, 4),
	}
}
800550
func (sc *conn) closeDueToPing(err error) {
	sc.Lock()
	if sc.nc == nil {
		sc.Unlock()
		return
	}
	// Stop timer, unsubscribe, fail the pubs, etc..
	sc.cleanupOnClose(err)
	// No need to send Close protocol, so simply close the underlying
	// NATS connection (if we own it, and if not already closed)
	if sc.ncOwned && !sc.nc.IsClosed() {
		sc.nc.Close()
	}
	// Mark this streaming connection as closed. Do this under pingMu lock.
	sc.pingMu.Lock()
	sc.nc = nil
	sc.pingMu.Unlock()
	// Capture callback (even though this is immutable).
	cb := sc.connLostCB
	sc.Unlock()
	if cb != nil {
		// Execute in separate go routine.
		go cb(sc, err)
	}
}
223735
func NewConn(c netConn) *Conn {
	conn, ok := c.(*Conn)
	if ok {
		return conn
	}
	conn, ok = connPool.Get().(*Conn)
	if !ok {
		conn = new(Conn)
	}
	conn.netConn = c
	return conn
}
75443
func (m *MockConnector) Nats() *go_nats.Conn {
	ret := m.ctrl.Call(m, "Nats")
	ret0, _ := ret[0].(*go_nats.Conn)
	return ret0
}
1312205
func (c *Client) handleConn(conn net.Conn) {
	// Read a single byte
	buf := make([]byte, 1)
	if _, err := conn.Read(buf); err != nil {
		if err != io.EOF {
			c.rpcLogger.Error("error reading byte", "error", err)
		}
		conn.Close()
		return
	}

	// Switch on the byte
	switch pool.RPCType(buf[0]) {
	case pool.RpcNomad:
		c.handleNomadConn(conn)

	case pool.RpcStreaming:
		c.handleStreamingConn(conn)

	default:
		c.rpcLogger.Error("unrecognized RPC byte", "byte", buf[0])
		conn.Close()
		return
	}
}
1374083
func (c *streamClient) closeConn() {
	conn := c.decoupleConn()
	if conn != nil {
		if err := conn.Close(); err != nil {
			log.Fields{
				log.ErrorKey: err,
			}.Warningf(c, "Error on connection close.")
		}
	}
}
576782
func newWSConn(conn *websocket.Conn) (*wsConn, error) {
	c := &wsConn{conn: conn}

	t, r, err := conn.NextReader()
	if err != nil {
		return nil, err
	}

	c.typ = t
	c.reader = r

	return c, nil
}
708681
func (f *fifo) SyscallConn() (syscall.RawConn, error) {
	// deterministic check for closed
	select {
	case <-f.closed:
		return nil, errors.New("fifo closed")
	default:
	}

	select {
	case <-f.closed:
		return nil, errors.New("fifo closed")
	case <-f.opened:
		return f.file.SyscallConn()
	default:
	}

	// Not opened and not closed, this means open is non-blocking AND it's not open yet
	// Use rawConn to deal with non-blocking open.
	rc := &rawConn{f: f, ready: make(chan struct{})}
	go func() {
		select {
		case <-f.closed:
			return
		case <-f.opened:
			rc.raw, rc.err = f.file.SyscallConn()
			close(rc.ready)
		}
	}()

	return rc, nil
}
1030414
func FromChan(ch <-chan *master.Call) RequestStreamingFunc {
	if ch == nil {
		// avoid blocking forever if we're handed a nil chan
		return func() *master.Call { return nil }
	}
	return func() *master.Call {
		if m, ok := <-ch; ok {
			return m
		}
		return nil
	}
}
415734
func (v *VoiceConnection) wsListen(wsConn *websocket.Conn, close <-chan struct{}) {

	v.log(LogInformational, "called")

	for {
		_, message, err := v.wsConn.ReadMessage()
		if err != nil {
			// Detect if we have been closed manually. If a Close() has already
			// happened, the websocket we are listening on will be different to the
			// current session.
			v.RLock()
			sameConnection := v.wsConn == wsConn
			v.RUnlock()
			if sameConnection {

				v.log(LogError, "voice endpoint %s websocket closed unexpectantly, %s", v.endpoint, err)

				// Start reconnect goroutine then exit.
				go v.reconnect()
			}
			return
		}

		// Pass received message to voice event handler
		select {
		case <-close:
			return
		default:
			go v.onEvent(message)
		}
	}
}
826781
func (c *Client) streamingRpcConn(server *servers.Server, method string) (net.Conn, error) {
	// Dial the server
	conn, err := net.DialTimeout("tcp", server.Addr.String(), 10*time.Second)
	if err != nil {
		return nil, err
	}

	// Cast to TCPConn
	if tcp, ok := conn.(*net.TCPConn); ok {
		tcp.SetKeepAlive(true)
		tcp.SetNoDelay(true)
	}

	// Check if TLS is enabled
	c.tlsWrapLock.RLock()
	tlsWrap := c.tlsWrap
	c.tlsWrapLock.RUnlock()

	if tlsWrap != nil {
		// Switch the connection into TLS mode
		if _, err := conn.Write([]byte{byte(pool.RpcTLS)}); err != nil {
			conn.Close()
			return nil, err
		}

		// Wrap the connection in a TLS client
		tlsConn, err := tlsWrap(c.Region(), conn)
		if err != nil {
			conn.Close()
			return nil, err
		}
		conn = tlsConn
	}

	// Write the multiplex byte to set the mode
	if _, err := conn.Write([]byte{byte(pool.RpcStreaming)}); err != nil {
		conn.Close()
		return nil, err
	}

	// Send the header
	encoder := codec.NewEncoder(conn, structs.MsgpackHandle)
	decoder := codec.NewDecoder(conn, structs.MsgpackHandle)
	header := structs.StreamingRpcHeader{
		Method: method,
	}
	if err := encoder.Encode(header); err != nil {
		conn.Close()
		return nil, err
	}

	// Wait for the acknowledgement
	var ack structs.StreamingRpcAck
	if err := decoder.Decode(&ack); err != nil {
		conn.Close()
		return nil, err
	}

	if ack.Error != "" {
		conn.Close()
		return nil, errors.New(ack.Error)
	}

	return conn, nil
}