async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)


manager = ConnectionManager()


@app.get("/")
async def get():
    return HTMLResponse(html)


@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:

/**
 * A {@code MessagingClient} maintains a single bi-directional uni-cast object connection with some peer.
 */
@ServiceScope(Scope.Global.class)
public interface MessagingClient {
    /**
     * Creates a connection to the given address. Blocks until the connection has been established.
     *
     * @param address The address to connect to.
     */
    ObjectConnection getConnection(Address address);

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)


manager = ConnectionManager()


@app.get("/")
async def get():
    return HTMLResponse(html)


@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:

import spock.lang.Specification

class MessageHubBackedClientTest extends Specification {
    final OutgoingConnector connector = Mock()
    final ExecutorFactory executorFactory = Mock()
    final MessageHubBackedClient client = new MessageHubBackedClient(connector, executorFactory)

    def "creates connection and cleans up on stop"() {
        Address address = Stub()
        ConnectCompletion connectCompletion = Mock()

		// We can't guarantee the Pilot we are connected to has a connection to the proxy we requested
		// There isn't a great way around this, but for debugging purposes its suitable to have the caller retry.
		w.WriteHeader(http.StatusNotFound)
		_, _ = w.Write([]byte("Proxy not connected to this Pilot instance. It may be connected to another instance.\n"))
		return
	}
	var connections []*Connection
	if con != nil {

### [Connections](https://square.github.io/okhttp/4.x/okhttp/okhttp3/-connection/)

When you request a URL with OkHttp, here's what it does:

 1. It uses the URL and configured OkHttpClient to create an **address**. This address specifies how we'll connect to the webserver.
 2. It attempts to retrieve a connection with that address from the **connection pool**.

import okhttp3.internal.assertThreadHoldsLock
import okhttp3.internal.closeQuietly
import okhttp3.internal.concurrent.Task
import okhttp3.internal.concurrent.TaskRunner
import okhttp3.internal.connection.Exchange
import okhttp3.internal.connection.RealCall
import okhttp3.internal.okHttpName
import okhttp3.internal.ws.WebSocketProtocol.CLOSE_CLIENT_GOING_AWAY
import okhttp3.internal.ws.WebSocketProtocol.CLOSE_MESSAGE_MAX

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)
}

func (conn *Connection) Close() {
	conn.s.closeConnection(conn)
}

func (conn *Connection) Watcher() xds.Watcher {
	return conn.proxy
}

  }

  /**
   * Invoked after a connection has been acquired for the `call`.
   *
   * This can be invoked more than 1 time for a single [Call]. For example, if the response
   * to the [Call.request] is a redirect to a different address.
   */
  open fun connectionAcquired(
    call: Call,
    connection: Connection,
  ) {
  }

  /**
   * Invoked after a connection has been released for the `call`.
   *

            }
        """
    }

    def "build fails if debugger is not ready"() {
        setup:
        // port is assigned but not connected
        def port = jdwpClient.port

        when:
        withConnection { connection ->
            connection.newTestLauncher()
                .withJvmTestClasses("example.MyTest")
                .debugTestsOn(port)
                .run()
        }