Function connect

internal/codespaces/rpc/invoker.go:76–147

Finds a free port to listen on and creates a new RPC invoker that connects to that port

(ctx context.Context, fwd portforwarder.PortForwarder)

// Finds a free port to listen on and creates a new RPC invoker that connects to that port
func connect(ctx context.Context, fwd portforwarder.PortForwarder) (Invoker, error) {
	listener, err := listenTCP()
	if err != nil {
		return nil, err
	}
	localAddress := listener.Addr().String()

	invoker := &invoker{
		fwd:      fwd,
		listener: listener,
	}

	// Create a cancelable context to be able to cancel background tasks
	// if we encounter an error while connecting to the gRPC server
	connectctx, cancel := context.WithCancel(context.Background())
	defer func() {
		if err != nil {
			cancel()
		}
	}()

	ch := make(chan error, 2) // Buffered channel to ensure we don't block on the goroutine

	// Ensure we close the port forwarder if we encounter an error
	// or once the gRPC connection is closed. pfcancel is retained
	// to close the PF whenever we close the gRPC connection.
	pfctx, pfcancel := context.WithCancel(connectctx)
	invoker.cancelPF = pfcancel

	// Tunnel the remote gRPC server port to the local port
	go func() {
		// Start forwarding the port locally
		opts := portforwarder.ForwardPortOpts{
			Port:     codespacesInternalPort,
			Internal: true,
		}
		ch <- fwd.ForwardPortToListener(pfctx, opts, listener)
	}()

	var conn *grpc.ClientConn
	go func() {
		// Attempt to connect to the port
		opts := []grpc.DialOption{
			grpc.WithTransportCredentials(insecure.NewCredentials()),
		}
		conn, err = grpc.NewClient(localAddress, opts...)
		ch <- err // nil if we successfully connected
	}()

	// Wait for the connection to be established or for the context to be cancelled
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case err := <-ch:
		if err != nil {
			return nil, err
		}
	}