registerResource registers a new resource object with a given type t and name.  It returns the auto-generated
    URN and the ID that will resolve after the deployment has completed.  All properties will be initialized to property
    objects that the registration operation will resolve at the right time (or remain unresolved for deployments).

def register_resource(res: 'Resource', ty: str, name: str, custom: bool, props: 'Inputs', opts: Optional['ResourceOptions']):
    """
    registerResource registers a new resource object with a given type t and name.  It returns the auto-generated
    URN and the ID that will resolve after the deployment has completed.  All properties will be initialized to property
    objects that the registration operation will resolve at the right time (or remain unresolved for deployments).
    """
    log.debug(f"registering resource: ty={ty}, name={name}, custom={custom}")
    monitor = settings.get_monitor()

    # Prepare the resource.

    # Simply initialize the URN property and get prepared to resolve it later on.
    # Note: a resource urn will always get a value, and thus the output property
    # for it can always run .apply calls.
    log.debug(f"preparing resource for RPC")
    urn_future = asyncio.Future()
    urn_known = asyncio.Future()
    urn_known.set_result(True)
    resolve_urn = urn_future.set_result
    resolve_urn_exn = urn_future.set_exception
    res.urn = known_types.new_output({res}, urn_future, urn_known)

    # If a custom resource, make room for the ID property.
    resolve_id: Optional[Callable[[Any, str, Optional[Exception]], None]] = None
    if custom:
        resolve_value = asyncio.Future()
        resolve_perform_apply = asyncio.Future()
        res.id = known_types.new_output({res}, resolve_value, resolve_perform_apply)

        def do_resolve(value: Any, perform_apply: bool, exn: Optional[Exception]):
            if exn is not None:
                resolve_value.set_exception(exn)
                resolve_perform_apply.set_exception(exn)
            else:
                resolve_value.set_result(value)
                resolve_perform_apply.set_result(perform_apply)

        resolve_id = do_resolve

    # Now "transfer" all input properties into unresolved futures on res.  This way,
    # this resource will look like it has all its output properties to anyone it is
    # passed to.  However, those futures won't actually resolve until the RPC returns
    resolvers = rpc.transfer_properties(res, props)

    async def do_register():
        try:
            log.debug(f"preparing resource registration: ty={ty}, name={name}")
            resolver = await prepare_resource(res, ty, custom, props, opts)
            log.debug(f"resource registration prepared: ty={ty}, name={name}")

            property_dependencies = {}
            for key, deps in resolver.property_dependencies.items():
                property_dependencies[key] = resource_pb2.RegisterResourceRequest.PropertyDependencies(urns=deps)

            ignore_changes = opts.ignore_changes
            if res.translate_input_property is not None and opts.ignore_changes is not None:
                ignore_changes = map(res.translate_input_property, opts.ignore_changes)

            req = resource_pb2.RegisterResourceRequest(
                type=ty,
                name=name,
                parent=resolver.parent_urn,
                custom=custom,
                object=resolver.serialized_props,
                protect=opts.protect,
                provider=resolver.provider_ref,
                dependencies=resolver.dependencies,
                propertyDependencies=property_dependencies,
                deleteBeforeReplace=opts.delete_before_replace,
                ignoreChanges=ignore_changes,
                version=opts.version or "",
            )

            def do_rpc_call():
                if monitor:
                    # If there is a monitor available, make the true RPC request to the engine.
                    try:
                        return monitor.RegisterResource(req)
                    except grpc.RpcError as exn:
                        # See the comment on invoke for the justification for disabling
                        # this warning
                        # pylint: disable=no-member
                        if exn.code() == grpc.StatusCode.UNAVAILABLE:
                            sys.exit(0)

                        details = exn.details()
                    raise Exception(details)
                else:
                    # If no monitor is available, we'll need to fake up a response, for testing.
                    return RegisterResponse(create_test_urn(ty, name), None, resolver.serialized_props)

            resp = await asyncio.get_event_loop().run_in_executor(None, do_rpc_call)
        except Exception as exn:
            log.debug(f"exception when preparing or executing rpc: {traceback.format_exc()}")
            rpc.resolve_outputs_due_to_exception(resolvers, exn)
            resolve_urn_exn(exn)
            if resolve_id is not None:
                resolve_id(None, False, exn)
            raise

        log.debug(f"resource registration successful: ty={ty}, urn={resp.urn}")
        resolve_urn(resp.urn)
        if resolve_id:
            # The ID is known if (and only if) it is a non-empty string. If it's either None or an empty string,
            # we should treat it as unknown. TFBridge in particular is known to send the empty string as an ID when
            # doing a preview.
            is_known = bool(resp.id)
            resolve_id(resp.id, is_known, None)

        await rpc.resolve_outputs(res, props, resp.object, resolvers)

    asyncio.ensure_future(RPC_MANAGER.do_rpc("register resource", do_register)())