/**
     * Connects to the emitter service.
     */
    public connect(request?: ConnectRequest, handler?: () => void): Emitter {
        request = request || {};

        // auto-resolve the security level
        if (request.secure == null) {
            if (typeof window !== "undefined" && window != null && window.location != null && window.location.protocol != null) {
                request.secure = window.location.protocol == "https:";
            } else {
                request.secure = false;
            }
        }

        // default options
        var defaultConnectOptions = {
            host: "api.emitter.io",
            port: request.secure ? 443 : 8080,
            keepalive: 30,
            secure: false
        };

        // apply defaults
        for (var k in defaultConnectOptions) {
            request[k] = "undefined" === typeof request[k] ? defaultConnectOptions[k] : request[k];
        }

        request.host = request.host.replace(/.*?:\/\//g, "");
        var brokerUrl = `${request.secure ? "wss://" : "ws://"}${request.host}:${request.port}`;

        this._callbacks = {"connect": [handler]};
        this._mqtt = mqtt.connect(brokerUrl, request);

        this._mqtt.on(EmitterEvents.connect, () => this._tryInvoke(EmitterEvents.connect, this));
        this._mqtt.on("close", () => this._tryInvoke(EmitterEvents.disconnect, this));
        this._mqtt.on("offline", () => this._tryInvoke(EmitterEvents.offline, this));
        this._mqtt.on("error", error => this._tryInvoke(EmitterEvents.error, error));
        this._mqtt.on("message", (topic, msg, packet) => {
            var message = new EmitterMessage(packet);
            if (this._startsWith(message.channel, "emitter/keygen")) {
                // This is keygen message.
                this._tryInvoke(EmitterEvents.keygen, message.asObject())
            } else if (this._startsWith(message.channel, "emitter/presence")) {
                // This is presence message.
                this._tryInvoke(EmitterEvents.presence, message.asObject())
            } else if (this._startsWith(message.channel, "emitter/me")) {
                // This is a message requesting info on the connection.
                this._tryInvoke(EmitterEvents.me, message.asObject());
            } else {
                // Do we have a message callback?
                this._tryInvoke(EmitterEvents.message, message);
            }
        });
        return this;
    }

export default async (
  { channelId, options },
  context,
  callback,
  chromeInstance
): Promise<void> => {
  // used to block requests from being processed while we're exiting
  let endingInvocation = false
  let timeout
  let executionCheckInterval

  debug('Invoked with data: ', channelId, options)

  const chrome = new LocalChrome({
    ...options,
    remote: false,
    cdp: { closeTab: true },
  })

  const queue = new Queue(chrome)

  const TOPIC_CONNECTED = `chrome/${channelId}/connected`
  const TOPIC_REQUEST = `chrome/${channelId}/request`
  const TOPIC_RESPONSE = `chrome/${channelId}/response`
  const TOPIC_END = `chrome/${channelId}/end`

  const channel = mqtt(createPresignedURL())

  if (process.env.DEBUG) {
    channel.on('error', error => debug('WebSocket error', error))
    channel.on('offline', () => debug('WebSocket offline'))
  }

  /*
    Clean up function whenever we want to end the invocation.
    Importantly we publish a message that we're disconnecting, and then
    we kill the running Chrome instance.
  */
  const end = (topic_end_data = {}) => {
    if (!endingInvocation) {
      endingInvocation = true
      clearInterval(executionCheckInterval)
      clearTimeout(timeout)

      channel.unsubscribe(TOPIC_END, () => {
        channel.publish(TOPIC_END, JSON.stringify({ channelId, chrome: true, ...topic_end_data }), {
          qos: 0,
        }, async () => {
          channel.end()

          await chrome.close()
          await chromeInstance.kill()

          callback()
        })
      })
    }
  }

  const newTimeout = () =>
    setTimeout(async () => {
      debug('Timing out. No requests received for 30 seconds.')
      await end({ inactivity: true })
    }, 30000)

  /*
    When we're almost out of time, we clean up.
    Importantly this makes sure that Chrome isn't running on the next invocation
    and publishes a message to the client letting it know we're disconnecting.
  */
  executionCheckInterval = setInterval(async () => {
    if (context.getRemainingTimeInMillis() < 5000) {
      debug('Ran out of execution time.')
      await end({ outOfTime: true })
    }
  }, 1000)

  channel.on('connect', () => {
    debug('Connected to AWS IoT broker')

    /*
      Publish that we've connected. This lets the client know that
      it can start sending requests (commands) for us to process.
    */
    channel.publish(TOPIC_CONNECTED, JSON.stringify({}), { qos: 1 })

    /*
      The main bit. Listen for requests from the client, handle them
      and respond with the result.
    */
    channel.subscribe(TOPIC_REQUEST, () => {
      debug(`Subscribed to ${TOPIC_REQUEST}`)

      timeout = newTimeout()

      channel.on('message', async (topic, buffer) => {
        if (TOPIC_REQUEST === topic && !endingInvocation) {
          const message = buffer.toString()

          clearTimeout(timeout)
//.........这里部分代码省略.........

    await new Promise(async (resolve, reject) => {
      const timeout = setTimeout(() => {
        if (this.channel) {
          this.channel.end()
        }

        reject(
          new Error(
            "Timed out after 30sec. Connection couldn't be established.",
          ),
        )
      }, 30000)

      try {
        const { endpointUrl, apiKey } = getEndpoint(this.options.remote)
        const { body: { url, channelId } } = await got(endpointUrl, {
          headers: apiKey
            ? {
                'x-api-key': apiKey,
              }
            : undefined,
          json: true,
        })

        this.channelId = channelId

        this.TOPIC_NEW_SESSION = 'chrome/new-session'
        this.TOPIC_CONNECTED = `chrome/${channelId}/connected`
        this.TOPIC_REQUEST = `chrome/${channelId}/request`
        this.TOPIC_RESPONSE = `chrome/${channelId}/response`
        this.TOPIC_END = `chrome/${channelId}/end`

        const channel = mqtt(url, {
          will: {
            topic: 'chrome/last-will',
            payload: JSON.stringify({ channelId }),
            qos: 1,
            retain: false,
          },
        })

        this.channel = channel

        if (this.options.debug) {
          channel.on('error', error => console.log('WebSocket error', error))
          channel.on('offline', () => console.log('WebSocket offline'))
        }

        channel.on('connect', () => {
          if (this.options.debug) {
            console.log('Connected to message broker.')
          }

          channel.subscribe(this.TOPIC_CONNECTED, { qos: 1 }, () => {
            channel.on('message', async topic => {
              if (this.TOPIC_CONNECTED === topic) {
                clearTimeout(timeout)
                resolve()
              }
            })

            channel.publish(
              this.TOPIC_NEW_SESSION,
              JSON.stringify({ channelId, options: this.options }),
              { qos: 1 },
            )
          })

          channel.subscribe(this.TOPIC_END, () => {
            channel.on('message', async (topic, buffer) => {
              if (this.TOPIC_END === topic) {
                const message = buffer.toString()
                const data = JSON.parse(message)

                if (data.outOfTime) {
                  console.warn(
                    `Chromeless Proxy disconnected because it reached it's execution time limit (5 minutes).`,
                  )
                } else if (data.inactivity) {
                  console.warn(
                    'Chromeless Proxy disconnected due to inactivity (no commands sent for 30 seconds).',
                  )
                } else {
                  console.warn(
                    `Chromeless Proxy disconnected (we don't know why).`,
                    data,
                  )
                }

                await this.close()
              }
            })
          })
        })
      } catch (error) {
        console.error(error)

        reject(
          new Error('Unable to get presigned websocket URL and connect to it.'),
        )
//.........这里部分代码省略.........

  constructor($args: { mqttUrl: string; name: string }, $log: Ha4usLogger) {
    const url = new URL($args.mqttUrl)
    $log.info(
      `Connecting to ${url.protocol}//${url.username}@${url.host} as ${
      $args.name
      }`
    )
    const mqtt = Mqtt.connect(
      $args.mqttUrl,
      {
        clientId:
          $args.name +
          '_' +
          Math.random()
            .toString(16)
            .substr(2, 8),
        rejectUnauthorized: false,
        will: {
          topic: $args.name + '/connected',
          payload: '0',
          qos: 0,
          retain: true,
        },
      }
    )

    mqtt.once('connect', () => {
      this.connected = 1
      this.$log.info('Connected as %s', mqtt.options.clientId)
    })
    mqtt.on('error', function (err) {
      $log.error('Error mqtt', err)
    })

    mqtt.on('close', function () {
      $log.warn('closing mqtt')
    })

    super(mqtt)

    this.mqtt = mqtt
    this.$log = $log
    this.caching$ = new ReplaySubject(1)

    this.cache$ = this.caching$.pipe(
      distinct(), // only add if not already in there
      mergeMap(topic => this.observe(topic)), // observe the new topic
      scan((acc: Map<string, Ha4usMessage>, msg: Ha4usMessage) => {
        acc.set(msg.topic, msg)
        return acc
      }, new Map<string, Ha4usMessage>()),
      shareReplay()
    )

    this.cache$.subscribe(cache => {
      debug(`cache updated ( size: ${cache.size})`)
      this.cache = cache
    })

    this.domain = $args.name
  }