// condRelease releases this connection if the error pointed to by err
// is nil (not an error) or is only a protocol level error (e.g. a
// cache miss).  The purpose is to not recycle TCP connections that
// are bad.
504764
func (cn *conn) condRelease(err *error) {
	if *err == nil || resumableError(*err) {
		cn.release()
	} else {
		cn.nc.Close()
	}
}
349100
func ConnLabel(source, target enode.ID) string {
	var first, second enode.ID
	if bytes.Compare(source.Bytes(), target.Bytes()) > 0 {
		first = target
		second = source
	} else {
		first = source
		second = target
	}
	return fmt.Sprintf("%v-%v", first, second)
}
615413
func (conn *Connection) Connect(ctx context.Context) (err error) {
	obs := prometheus.NewTimer(connInitialConnectTime)
	defer obs.ObserveDuration()

	conn.conMu.Lock()
	defer conn.conMu.Unlock()

	var tlsc *tls.Config

	if !conn.choria.Config.DisableTLS {
		tlsc, err = conn.choria.TLSConfig()
		if err != nil {
			err = fmt.Errorf("Could not create TLS Config: %s", err)
			return err
		}
	}

	urls := []string{}
	var url *url.URL

	servers, err := conn.servers()
	if err != nil {
		err = fmt.Errorf("Could not resolve servers during initial connect: %s", err)
	}

	for _, server := range servers {
		url, err = server.URL()
		if err != nil {
			err = fmt.Errorf("Could not determine URL for server %#v", server)
			return
		}

		urls = append(urls, url.String())
	}

	options := []nats.Option{
		nats.MaxReconnects(-1),
		nats.Name(conn.name),

		// This is specifically set quite small, just about enough to handle short
		// reconnects rather than the 8MB long buffer that's default.  30 000 nodes
		// each sending several MB after reconnect is not what anyone wants
		nats.ReconnectBufSize(10 * 1024),

		// nats.SetPend
		nats.DisconnectHandler(func(nc *nats.Conn) {
			err = nc.LastError()

			if err != nil {
				conn.logger.Warnf("NATS client connection got disconnected: %s", nc.LastError())
			}
		}),

		nats.ReconnectHandler(func(nc *nats.Conn) {
			conn.logger.Warnf("NATS client reconnected after a previous disconnection, connected to %s", nc.ConnectedUrl())
			connReconnectCtr.Inc()
		}),

		nats.ClosedHandler(func(nc *nats.Conn) {
			err = nc.LastError()
			if err != nil {
				conn.logger.Warnf("NATS client connection closed: %s", nc.LastError())
			}
			connClosedCtr.Inc()
		}),

		nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
			conn.logger.Errorf("NATS client on %s encountered an error: %s", nc.ConnectedUrl(), err)
			connErrorCtr.Inc()
		}),
	}

	if !conn.choria.Config.DisableTLS {
		options = append(options, nats.Secure(tlsc))
	}

	if !conn.choria.Config.Choria.RandomizeMiddlewareHosts {
		options = append(options, nats.DontRandomize())
	}

	try := 0

	for {
		try++

		conn.nats, err = nats.Connect(strings.Join(urls, ", "), options...)
		if err != nil {
			connInitialConnectCtr.Inc()

			conn.logger.Warnf("Initial connection to the NATS broker cluster failed: %s", err)

			if ctx.Err() != nil {
				err = fmt.Errorf("Initial connection cancelled due to shut down")
				return
			}

			s := backoff.FiveSec.Duration(try)
			conn.logger.Infof("Sleeping %s after failed connection attempt %d", s, try)
			time.Sleep(s)

			continue
		}

		conn.logger.Debugf("Connected to %s", conn.nats.ConnectedUrl())

		break
	}

	return
}
1142456
func (cn *conn) release() {
	cn.c.putFreeConn(cn.srv.Addr, cn)
}
140542
func (kea *KamEvapi) Disconnect() (err error) {
	kea.connMutex.Lock()
	defer kea.connMutex.Unlock()
	if kea.conn != nil {
		kea.logger.Printf(fmt.Sprintf(" Disconnecting from %s", kea.kamaddr))
		err = kea.conn.Close()
		kea.conn = nil
	}
	return
}
715231
func (p *ConnPool) Release(conn *Conn) {
	if conn.ctxInProgress {
		panic("should never release when context is in progress")
	}

	if conn.txStatus != 'I' {
		conn.Exec("rollback")
	}

	if len(conn.channels) > 0 {
		if err := conn.Unlisten("*"); err != nil {
			conn.die(err)
		}
		conn.channels = make(map[string]struct{})
	}
	conn.notifications = nil

	p.cond.L.Lock()

	if conn.poolResetCount != p.resetCount {
		conn.Close()
		p.cond.L.Unlock()
		p.cond.Signal()
		return
	}

	if conn.IsAlive() {
		p.availableConnections = append(p.availableConnections, conn)
	} else {
		p.removeFromAllConnections(conn)
	}
	p.cond.L.Unlock()
	p.cond.Signal()
}
917132
func (s *storageLeases) RemoveLease(ip string) error {
	return s.storage.Delete(apirequest.NewDefaultContext(), s.baseKey+"/"+ip, &corev1.Endpoints{}, nil)
}
698517
func (s *RPCServer) Detach(arg DetachIn, out *DetachOut) error {
	err := s.debugger.Detach(arg.Kill)
	if s.config.DisconnectChan != nil {
		close(s.config.DisconnectChan)
		s.config.DisconnectChan = nil
	}
	return err
}
62931
func (d *ProtobufDecoder) DecodeDisconnect(data []byte) (*DisconnectRequest, error) {
	var p DisconnectRequest
	err := p.Unmarshal(data)
	if err != nil {
		return nil, err
	}
	return &p, nil
}
1184600
func (cm *ConnManager) Connect(c *ConnReq) {
	if atomic.LoadInt32(&cm.stop) != 0 {
		return
	}

	// During the time we wait for retry there is a chance that
	// this connection was already cancelled
	if c.State() == ConnCanceled {
		log.Debugf("Ignoring connect for canceled connreq=%v", c)
		return
	}

	if atomic.LoadUint64(&c.id) == 0 {
		atomic.StoreUint64(&c.id, atomic.AddUint64(&cm.connReqCount, 1))

		// Submit a request of a pending connection attempt to the
		// connection manager. By registering the id before the
		// connection is even established, we'll be able to later
		// cancel the connection via the Remove method.
		done := make(chan struct{})
		select {
		case cm.requests <- registerPending{c, done}:
		case <-cm.quit:
			return
		}

		// Wait for the registration to successfully add the pending
		// conn req to the conn manager's internal state.
		select {
		case <-done:
		case <-cm.quit:
			return
		}
	}

	log.Debugf("Attempting to connect to %v", c)

	conn, err := cm.cfg.Dial(c.Addr)
	if err != nil {
		select {
		case cm.requests <- handleFailed{c, err}:
		case <-cm.quit:
		}
		return
	}

	select {
	case cm.requests <- handleConnected{c, conn}:
	case <-cm.quit:
	}
}
476659
func WrapNetConn(nc net.Conn) net.Conn {
	if !enabled {
		return nc
	}
	if nc == nil {
		return nil
	}
	return wrappedNetConn{
		nc,
		newCloseWrapper(nc),
	}
}
1022251
func (p *Peer) Disconnect(reason DiscReason) {
	select {
	case p.disc <- reason:
	case <-p.closed:
	}
}
1321162
func ReleaseSlice(ctx context.Context, identities []*identity.Identity) error {
	var err error
	for _, id := range identities {
		if id == nil {
			continue
		}
		if _, err2 := Release(ctx, id); err2 != nil {
			log.WithError(err2).WithFields(logrus.Fields{
				logfields.Identity: id,
			}).Error("Failed to release identity")
			err = err2
		}
	}
	return err
}
388368
func (c *client) Disconnect(quiesce uint) {
	status := atomic.LoadUint32(&c.status)
	if status == connected {
		DEBUG.Println(CLI, "disconnecting")
		c.setConnected(disconnected)

		dm := packets.NewControlPacket(packets.Disconnect).(*packets.DisconnectPacket)
		dt := newToken(packets.Disconnect)
		c.oboundP <- &PacketAndToken{p: dm, t: dt}

		// wait for work to finish, or quiesce time consumed
		dt.WaitTimeout(time.Duration(quiesce) * time.Millisecond)
	} else {
		WARN.Println(CLI, "Disconnect() called but not connected (disconnected/reconnecting)")
		c.setConnected(disconnected)
	}

	c.disconnect()
}
950296
func (c *Client) DeleteRelease(appID, releaseID string) (*ct.ReleaseDeletion, error) {
	events := make(chan *ct.Event)
	stream, err := c.StreamEvents(ct.StreamEventsOptions{
		AppID:       appID,
		ObjectID:    releaseID,
		ObjectTypes: []ct.EventType{ct.EventTypeReleaseDeletion},
	}, events)
	if err != nil {
		return nil, err
	}
	defer stream.Close()

	if err := c.Delete(fmt.Sprintf("/apps/%s/releases/%s", appID, releaseID), nil); err != nil {
		return nil, err
	}

	select {
	case event, ok := <-events:
		if !ok {
			return nil, stream.Err()
		}
		var e ct.ReleaseDeletionEvent
		if err := json.Unmarshal(event.Data, &e); err != nil {
			return nil, err
		}
		if e.Error != "" {
			return nil, errors.New(e.Error)
		}
		return e.ReleaseDeletion, nil
	case <-time.After(60 * time.Second):
		return nil, errors.New("timed out waiting for release deletion")
	}
}
663869
func (cli *Client) NetworkDisconnect(ctx context.Context, networkID, containerID string, force bool) error {
	nd := types.NetworkDisconnect{Container: containerID, Force: force}
	resp, err := cli.post(ctx, "/networks/"+networkID+"/disconnect", nil, nd, nil)
	ensureReaderClosed(resp)
	return err
}
805255
func (cp *Connack) Decode(src []byte) (int, error) {
	total := 0

	// decode header
	hl, _, rl, err := headerDecode(src, CONNACK)
	total += hl
	if err != nil {
		return total, err
	}

	// check remaining length
	if rl != 2 {
		return total, makeError(cp.Type(), "expected remaining length to be 2")
	}

	// read connack flags
	connackFlags := src[total]
	cp.SessionPresent = connackFlags&0x1 == 1
	total++

	// check flags
	if connackFlags&254 != 0 {
		return 0, makeError(cp.Type(), "bits 7-1 in acknowledge flags are not 0")
	}

	// read return code
	cp.ReturnCode = ConnackCode(src[total])
	total++

	// check return code
	if !cp.ReturnCode.Valid() {
		return 0, makeError(cp.Type(), "invalid return code (%d)", cp.ReturnCode)
	}

	return total, nil
}