async def METHOD_NAME(self):
    # TODO: extract json request data stuff into mixin?
    request_data = tornado.escape.json_decode(self.request.body)
    connection = request_data['connection']
    pool = AsyncAdapter.make_pool()
    with tracer.start_as_current_span("executor setup"):
        if connection["type"].lower() == "tcp":
            try:
                sync_executor = await sync_to_async(partial(DaskJobExecutor.connect,
                    scheduler_uri=connection['address'],
                ), pool=pool)
            except Exception as e:
                msg = Message(self.state).cluster_conn_error(msg=str(e))
                log_message(msg)
                self.write(msg)
                return None
        elif connection["type"].lower() == "local":
            devices = detect()
            options = {
                "local_directory": self.state.get_local_directory()
            }
            try:
                if "numWorkers" in connection:
                    num_workers = connection["numWorkers"]
                    if not isinstance(num_workers, int) or num_workers < 1:
                        raise ValueError('Number of workers must be positive integer')
                    devices["cpus"] = range(num_workers)
                raw_cudas = connection.get("cudas", {})
                cudas = _convert_device_map(raw_cudas)
                devices["cudas"] = cudas
                sync_executor = await sync_to_async(partial(DaskJobExecutor.make_local,
                    spec=cluster_spec(**devices,
                                      options=options,
                                      preload=self.state.get_preload())
                ), pool=pool)
            except Exception as e:
                msg = Message(self.state).cluster_conn_error(msg=str(e))
                log_message(msg)
                self.write(msg)
                return None
        else:
            raise ValueError("unknown connection type")
        executor = AsyncAdapter(wrapped=sync_executor, pool=pool)
        await self.state.executor_state.set_executor(executor, request_data)
    await self.state.dataset_state.verify()
    datasets = await self.state.dataset_state.serialize_all()
    msg = Message(self.state).initial_state(
        jobs=self.state.job_state.serialize_all(),
        datasets=datasets, analyses=self.state.analysis_state.serialize_all(),
        compound_analyses=self.state.compound_analysis_state.serialize_all(),
    )
    log_message(msg)
    # FIXME: don't broadcast, only send to the websocket that matches this HTTP connection
    # (is this even possible?)
    self.event_registry.broadcast_event(msg)
    await self.engine.send_existing_job_results()
    self.write({
        "status": "ok",
        "connection": connection,
    })