assertThat(c1.socket().isClosed).isFalse()
  }

  @Test fun cleanupPrioritizesEarliestEviction() {
    val pool = factory.newConnectionPool()
    val c1 = factory.newConnection(pool, routeA1, 75L)
    val c2 = factory.newConnection(pool, routeB1, 50L)

    // Running at time 75, the pool returns that nothing can be evicted until time 150.
    assertThat(pool.closeConnections(75L)).isEqualTo(75L)

	if err != nil {
		conn.Close()
		return nil, err
	}

	s.conn = conn

	return resp, nil
}

// NewConnection validates the upgrade response, creating and returning a new
// httpstream.Connection if there were no errors.
func (s *SpdyRoundTripper) NewConnection(resp *http.Response) (httpstream.Connection, error) {
	connectionHeader := strings.ToLower(resp.Header.Get(httpstream.HeaderConnection))

	pushRequest *model.PushRequest

	// function to call once a push is finished. This must be called or future changes may be blocked.
	done func()
}

func newConnection(peerAddr string, stream DiscoveryStream) *Connection {
	return &Connection{
		Connection: xds.NewConnection(peerAddr, stream),
	}
}

func (conn *Connection) Initialize(node *core.Node) error {
	return conn.s.initConnection(node, conn, conn.ids)
}

// StreamSecrets serves SDS discovery requests and SDS push requests
func (s *sdsservice) StreamSecrets(stream sds.SecretDiscoveryService_StreamSecretsServer) error {
	return xds.Stream(&Context{
		BaseConnection: xds.NewConnection("", stream),
		s:              s,
		w:              &Watch{},
	})
}

func (s *sdsservice) DeltaSecrets(stream sds.SecretDiscoveryService_DeltaSecretsServer) error {

	t *testing.T
}

func (u *fakeUpgrader) RoundTrip(req *http.Request) (*http.Response, error) {
	u.called = true
	u.req = req
	return u.resp, u.err
}

func (u *fakeUpgrader) NewConnection(resp *http.Response) (httpstream.Connection, error) {
	if u.checkResponse && u.resp != resp {
		u.t.Errorf("response objects passed did not match: %#v", resp)
	}
	return u.conn, u.connErr
}

		if host == o.Local {
			if found {
				return nil, fmt.Errorf("grid: local host found multiple times")
			}
			found = true
			// No connection to local.
			continue
		}
		m.targets[host] = newConnection(connectionParams{
			ctx:           ctx,
			id:            m.ID,
			local:         o.Local,
			remote:        host,
			dial:          o.Dialer,
			handlers:      &m.handlers,
			auth:          o.AddAuth,

				}
			}

			client := &http.Client{Transport: spdyTransport}

			resp, err := client.Do(req)
			var conn httpstream.Connection
			if err == nil {
				conn, err = spdyTransport.NewConnection(resp)
			}
			haveErr := err != nil
			if e, a := testCase.shouldError, haveErr; e != a {
				t.Fatalf("shouldError=%t, got %t: %v", e, a, err)
			}
			if testCase.shouldError {
				return
			}

	reqChan chan *discovery.DiscoveryRequest

	// errorChan is used to process error during discovery request processing.
	errorChan chan error
}

func NewConnection(peerAddr string, stream DiscoveryStream) Connection {
	return Connection{
		pushChannel: make(chan any),
		initialized: make(chan struct{}),
		stop:        make(chan struct{}),

	outgoingBytes func(n int64) // Record outgoing bytes.
	publisher     *pubsub.PubSub[madmin.TraceInfo, madmin.TraceType]

	blockConnect chan struct{}
}

// newConnection will create an unconnected connection to a remote.
func newConnection(o connectionParams) *Connection {
	c := &Connection{
		state:              StateUnconnected,
		Remote:             o.remote,
		Local:              o.local,

	}
	// Else fallback based on type
	return xds.IsWildcardTypeURL(w.TypeUrl)
}

func newDeltaConnection(peerAddr string, stream DeltaDiscoveryStream) *Connection {
	return &Connection{
		Connection:   xds.NewConnection(peerAddr, nil),
		deltaStream:  stream,
		deltaReqChan: make(chan *discovery.DeltaDiscoveryRequest, 1),
	}
}

// To satisfy methods that need DiscoveryRequest. Not suitable for real usage