// The worst connection is one that hasn't been sent, or sent anything useful
// for the longest. A bad connection is one that usually sends us unwanted
// pieces, or has been in worser half of the established connections for more
// than a minute.
429727
func (t *Torrent) worstBadConn() *connection {
	wcs := worseConnSlice{t.unclosedConnsAsSlice()}
	heap.Init(&wcs)
	for wcs.Len() != 0 {
		c := heap.Pop(&wcs).(*connection)
		if c.stats.ChunksReadWasted.Int64() >= 6 && c.stats.ChunksReadWasted.Int64() > c.stats.ChunksReadUseful.Int64() {
			return c
		}
		// If the connection is in the worst half of the established
		// connection quota and is older than a minute.
		if wcs.Len() >= (t.maxEstablishedConns+1)/2 {
			// Give connections 1 minute to prove themselves.
			if time.Since(c.completedHandshake) > time.Minute {
				return c
			}
		}
	}
	return nil
}
241883
func (s *Store) Connection(id uint64) (*Connection, bool) {
	s.connsMu.RLock()
	defer s.connsMu.RUnlock()
	c, ok := s.conns[id]
	return c, ok
}
697172
func (d *JSONDecoder) DecodeDisconnect(data []byte) (*DisconnectRequest, error) {
	var p DisconnectRequest
	err := json.Unmarshal(data, &p)
	if err != nil {
		return nil, err
	}
	return &p, nil
}
879602
func (c *DefaultClient) Connect() error {
	_, err := c.connect(false)
	return err
}
216582
func (s *Service) connect(fail chan struct{}) (*Client, bool) {
	// prepare new client
	client := New()
	client.Session = s.Session
	client.Logger = s.Logger
	client.futureStore = s.futureStore

	// set callback
	client.Callback = func(msg *packet.Message, err error) error {
		if err != nil {
			s.err("Client", err)
			close(fail)
			return nil
		}

		// call the handler
		if s.MessageCallback != nil {
			return s.MessageCallback(msg)
		}

		return nil
	}

	// attempt to connect
	connectFuture, err := client.Connect(s.config)
	if err != nil {
		s.err("Connect", err)
		return nil, false
	}

	// wait for connack
	err = connectFuture.Wait(s.ConnectTimeout)

	// check if future has been canceled
	if err == future.ErrCanceled {
		s.err("Connect", err)
		return nil, false
	}

	// check if future has timed out
	if err == future.ErrTimeout {
		client.Close()

		s.err("Connect", err)
		return nil, false
	}

	return client, connectFuture.SessionPresent()
}
693973
func (ctx *Context) BadRequest(err ...error) {
	e := badRequest
	if len(err) > 0 {
		e = err[0].Error()
	}

	ctx.Error(e, fasthttp.StatusBadRequest)
}
166773
func (c *Connection) Send(ctx context.Context, messageParts ...[]byte) (<-chan Message, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	msgID := atomic.AddUint64(&c.lastMessageID, 1)
	chunks, err := buildChunks(msgID, c.maxChunkSize, messageParts...)
	if err != nil {
		return nil, driver.WithStack(err)
	}
	// Prepare for receiving a response
	m := c.msgStore.Add(msgID)
	responseChan := m.responseChan

	//panic(fmt.Sprintf("chunks: %d, messageParts: %d, first: %s", len(chunks), len(messageParts), hex.EncodeToString(messageParts[0])))

	// Send all chunks
	sendErrors := make(chan error)
	deadline, ok := ctx.Deadline()
	if !ok {
		deadline = time.Time{}
	}
	go func() {
		defer close(sendErrors)
		for _, chunk := range chunks {
			if err := c.sendChunk(deadline, chunk); err != nil {
				// Cancel response
				c.msgStore.Remove(msgID)
				// Return error
				sendErrors <- driver.WithStack(err)
				return
			}
		}
	}()

	// Wait for sending to be ready, or context to be cancelled.
	select {
	case err := <-sendErrors:
		if err != nil {
			return nil, driver.WithStack(err)
		}
		return responseChan, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}
406795
func (t *Torrent) addConnection(c *connection) (err error) {
	defer func() {
		if err == nil {
			torrent.Add("added connections", 1)
		}
	}()
	if t.closed.IsSet() {
		return errors.New("torrent closed")
	}
	for c0 := range t.conns {
		if c.PeerID != c0.PeerID {
			continue
		}
		if !t.cl.config.dropDuplicatePeerIds {
			continue
		}
		if left, ok := c.hasPreferredNetworkOver(c0); ok && left {
			c0.Close()
			t.deleteConnection(c0)
		} else {
			return errors.New("existing connection preferred")
		}
	}
	if len(t.conns) >= t.maxEstablishedConns {
		c := t.worstBadConn()
		if c == nil {
			return errors.New("don't want conns")
		}
		c.Close()
		t.deleteConnection(c)
	}
	if len(t.conns) >= t.maxEstablishedConns {
		panic(len(t.conns))
	}
	t.conns[c] = struct{}{}
	return nil
}
640739
func (cp *ProviderFactory) Connection() Connection {
	cp.mtx.RLock()
	defer cp.mtx.RUnlock()
	return cp.connection
}
383096
func (m *model) Connection(deviceID protocol.DeviceID) (connections.Connection, bool) {
	m.pmut.RLock()
	cn, ok := m.conn[deviceID]
	m.pmut.RUnlock()
	if ok {
		m.deviceWasSeen(deviceID)
	}
	return cn, ok
}
1180884
func (p *Proxy) downConnection(ctx context.Context, ip net.IP, port int) (net.Conn, error) {

	raddr := &net.TCPAddr{
		IP:   ip,
		Port: port,
	}

	return markedconn.DialMarkedWithContext(ctx, "tcp4", raddr.String(), proxyMarkInt)

}
218040
func ConnectionProblem(err error, message string, args ...interface{}) error {
	return WrapWithMessage(&ConnectionProblemError{
		Message: fmt.Sprintf(message, args...),
		Err:     err,
	}, message, args...)
}
842674
func (c *BaseConn) Receive() (packet.Generic, error) {
	c.rMutex.Lock()
	defer c.rMutex.Unlock()

	// read next packet
	pkt, err := c.stream.Read()
	if err != nil {
		// ensure connection gets closed
		_ = c.carrier.Close()

		return nil, err
	}

	// reset timeout
	err = c.resetTimeout()
	if err != nil {
		return nil, err
	}

	return pkt, nil
}
736398
func BadRequest(id, format string, a ...interface{}) error {
	return &Error{
		Id:     id,
		Code:   400,
		Detail: fmt.Sprintf(format, a...),
		Status: http.StatusText(400),
	}
}
283609
func (t *Transport) cleanup(all bool) {
	staleTime := time.Now().Add(-t.expire)
	for proto, stack := range t.conns {
		if len(stack) == 0 {
			continue
		}
		if all {
			t.conns[proto] = nil
			// now, the connections being passed to closeConns() are not reachable from
			// transport methods anymore. So, it's safe to close them in a separate goroutine
			go closeConns(stack)
			continue
		}
		if stack[0].used.After(staleTime) {
			continue
		}

		// connections in stack are sorted by "used"
		good := sort.Search(len(stack), func(i int) bool {
			return stack[i].used.After(staleTime)
		})
		t.conns[proto] = stack[good:]
		// now, the connections being passed to closeConns() are not reachable from
		// transport methods anymore. So, it's safe to close them in a separate goroutine
		go closeConns(stack[:good])
	}
}
195359
func peerURLFromConnectionFailedStatus(details []interface{}) string {
	if len(details) != 0 {
		url, ok := details[0].(string)
		if ok {
			return endpoint.ToAddress(url)
		}
	}
	return ""
}
430747
func (cn *connection) downloadRate() float64 {
	return float64(cn.stats.BytesReadUsefulData.Int64()) / cn.cumInterest().Seconds()
}