}

// monitorState will monitor the state of the connection and close the net.Conn if it changes.
func (c *Connection) monitorState(conn net.Conn, cancel context.CancelCauseFunc) {
	c.connChange.L.Lock()
	defer c.connChange.L.Unlock()
	for {
		newState := c.State()
		if newState != StateConnected {
			conn.Close()
			cancel(ErrDisconnected)
			return
		}
		// Unlock and wait for state change.

        when:
        def acceptor = incomingConnector.accept({ ConnectCompletion event ->
            def conn = event.create(serializer)
            conn.dispatch("bye")
            conn.stop()
            instant.closed
        } as Action, false)

        def connection = outgoingConnector.connect(acceptor.address).create(serializer)

pkg database/sql, method (*Conn) BeginTx(context.Context, *TxOptions) (*Tx, error)
pkg database/sql, method (*Conn) Close() error
pkg database/sql, method (*Conn) ExecContext(context.Context, string, ...interface{}) (Result, error)
pkg database/sql, method (*Conn) PingContext(context.Context) error
pkg database/sql, method (*Conn) PrepareContext(context.Context, string) (*Stmt, error)

		_, err := httpstream.Handshake(req, w, []string{constants.PortForwardV1Name})
		require.NoError(t, err)
		upgrader := spdy.NewResponseUpgrader()
		conn := upgrader.UpgradeResponse(w, req, justQueueStream(streamChan))
		require.NotNil(t, conn)
		defer conn.Close() //nolint:errcheck
		<-stopServerChan
	}))
	defer spdyServer.Close()
	// Create UpgradeAwareProxy handler, with url/transport pointing to upstream SPDY. Then

						if src == dst {
							dst = (dst + 1) % len(managers)
						}
						local := managers[src]
						conn := local.Connection(hosts[dst])
						if conn == nil {
							b.Fatal("No connection")
						}
						// Send the payload.
						t := time.Now()
						resp, err := conn.Request(ctx, handlerTest, payload)
						latency += time.Since(t).Nanoseconds()
						if err != nil {
							if debugReqs {

	return buckets, nil
}

func (client *remotePeerS3Client) HealBucket(ctx context.Context, bucket string, opts madmin.HealOpts) (madmin.HealResultItem, error) {
	conn := client.gridConn()
	if conn == nil {
		return madmin.HealResultItem{}, nil
	}

	mss := grid.NewMSSWith(map[string]string{
		peerS3Bucket:        bucket,
		peerS3BucketDeleted: strconv.FormatBool(opts.Remove),
	})

}

func startTestSocketPeer(t testing.TB, conn Conn, op string, chunkSize, totalSize int) (func(t testing.TB), error) {
	t.Helper()

	if runtime.GOOS == "windows" {
		// TODO(panjf2000): Windows has not yet implemented FileConn,
		//		remove this when it's implemented in https://go.dev/issues/9503.
		t.Fatalf("startTestSocketPeer is not supported on %s", runtime.GOOS)
	}

	f, err := conn.(interface{ File() (*os.File, error) }).File()

	conn := client.gridConn()
	if conn == nil {
		return nil
	}
	_, err := reloadPoolMetaRPC.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
	return err
}

func (client *peerRESTClient) StopRebalance(ctx context.Context) error {
	conn := client.gridConn()
	if conn == nil {
		return nil
	}
	_, err := stopRebalanceRPC.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))

	// MinKMS, MinKES or Builtin.
	Type Type

	// The default key, used for generating new data keys
	// if no explicit GenerateKeyRequest.Name is provided.
	DefaultKey string

	conn conn // Connection to the KMS

	// Metrics
	reqOK, reqErr, reqFail atomic.Uint64
	latencyBuckets         []time.Duration // expected to be sorted
	latency                []atomic.Uint64
}

			proxyURL, _ := url.Parse(proxy.URL)

			conn, err := net.Dial("tcp", proxyURL.Host)
			require.NoError(t, err)
			bufferedReader := bufio.NewReader(conn)

			// Send upgrade request resulting in a non-101 response from the backend
			req, _ := http.NewRequest("GET", "/", nil)
			req.Header.Set(httpstream.HeaderConnection, httpstream.HeaderUpgrade)
			require.NoError(t, req.Write(conn))