//.........这里部分代码省略.........

    let results: I.ResultData<Val>
    // KV txn. Query is a set of keys.
    let maxVersion: I.Version

    switch (query.type) {
      case I.QueryType.KV:
        results = new Map<I.Key, Val>()
        maxVersion = getKVResults(dbTxn, query.q, opts, results)
        break
      case I.QueryType.AllKV:
        results = new Map<I.Key, Val>()
        maxVersion = getAllResults(dbTxn, opts, results)
        break

      case I.QueryType.Range:
        bakedQuery = {type: I.QueryType.StaticRange, q: query.q.slice()}
      case I.QueryType.StaticRange: {
        const q = query.q as I.RangeQuery // | I.StaticRangeQuery
        maxVersion = V_ZERO
        results = q.map((range) => {
          const staticRange = (query.type === I.QueryType.Range) ? bakeRange(dbTxn, range) : (range as I.StaticRange)
          const docs = [] // A map might be nicer.
          for (const [k, bytes] of rangeContentIter(dbTxn, staticRange)) {
            const [lastMod, doc] = decode(bytes)
            maxVersion = versionLib.vMax(maxVersion, lastMod)
            docs.push([k, doc])
          }
          return docs
        })

        break
      }
      default: return Promise.reject(new err.UnsupportedTypeError(`${query.type} not supported by lmdb store`))
    }
      // = query.type === 'kv'
      //  ? getKVResults(dbTxn, query.q, opts, results)
      //  : getAllResults(dbTxn, opts, results)

    dbTxn.abort()

    // console.log('lmdb fetch', results, maxVersion, version)
    return Promise.resolve({
      bakedQuery,
      results,
      versions: [{from: maxVersion, to: version}]
    })
  }

  const subGroup = new SubGroup({initialVersion: [version], fetch, getOps: inner.getOps.bind(inner)})

  const store: I.Store<Val> = {
    storeInfo: {
      uid: `lmdb(${inner.storeInfo.uid})`, // TODO: Maybe just re-expose inner.storeinfo.uid? All kv wraps should be identical
      sources: [source],
      capabilities,
    },

    async mutate(type, _txn, versions, opts = {}) {
      // TODO: Refactor this out by implementing internalDidChange
      if (type !== I.ResultType.KV) throw new err.UnsupportedTypeError()
      const txn = _txn as I.KVTxn<Val>

      // await ready

      debug('mutate', txn)

const reconnector = <Val>(connect: (() => Promise<[TinyReader<N.SCMsg>, TinyWriter<N.CSMsg>]>)): [I.Store<Status>, Promise<I.Store<Val>>, Promise<void>] => {
  // This is a tiny store that the client can use to track & display whether
  // or not we're currently connected. Ite tempting to make a metastore be a
  // default feature.
  const status = singlemem('waiting')
  let innerStore: NetStore<Val> | null = null
  let shouldReconnect = true
  const uidChanged = resolvable()

  // const initialStoreP = 
  // initialStoreP.then(store => { innerStore = store; })

  const ready: Promise<I.Store<Val>> = new Promise((resolve, reject) => {
    const opts: ClientOpts<Val> = {
      preserveState: true,
      onClose() {
        // We don't clear innerStore yet - all requests will still go there
        // until we change the guard.
        setSingle(status, 'waiting')

        // This is pretty rough.
        ;(async () => {
          while (shouldReconnect) {
            console.log('... trying to reconnect ...')

            let r: TinyReader<N.SCMsg>, w: TinyWriter<N.CSMsg>
            let netConnected = false
            try {
              setSingle(status, 'connecting')
              ;[r, w] = await connect()
              netConnected = true
            } catch (e) {
              console.warn('Reconnection failed', e.message)
            }

            if (netConnected) try {
              console.log('createStore')
              await createStore(r!, w!, {
                ...opts,
                restoreFrom: innerStore!,
                syncReady(store) {
                  // Ok, we've reconnected & gotten our hello message.

                  // We have to initialize here to avoid an event loop frame
                  // where innerStore is set incorrectly.
                  innerStore = store
                  setSingle(status, 'connected')
                  console.warn('Reconnected')
                }
              })
              break
            } catch (e) {
              // TODO: Consider calling reject instead of resolve here - so the
              // error makes an exception by default.
              if (e instanceof err.StoreChangedError) {
                console.log('uid changed')
                w!.close()
                uidChanged.reject(e)
                break
              } else throw e
            }

            setSingle(status, 'waiting')
            await wait(5000)
            console.log('done waiting')
          }
        })()
      },
    }


    setSingle(status, 'connecting')
    connect()
    .then(([r, w]) => createStore(r, w, opts))
    .then(initialStore => {
      innerStore = initialStore
      setSingle(status, 'connected')

      // This is basically a proxy straight to innerStore.
      const s: I.Store<Val> = {
        storeInfo: initialStore.storeInfo,
        fetch(...args) { return innerStore!.fetch(...args) },
        getOps(...args) { return innerStore!.getOps(...args) },
        mutate(...args) { return innerStore!.mutate(...args) },
        subscribe(...args) { return innerStore!.subscribe(...args) },
        close() {
          shouldReconnect = false
          innerStore!.close()
          innerStore = null // Error if we get subsequent requests
          setSingle(status, 'closed')
          // ... And stop reconnecting.
        }
      }
      resolve(s)
    }, reject)
  })

  return [status, ready, uidChanged]
}

export default function createKVStore<Val>(
  _data?: Map<I.Key, Val>,
  storeOpts: KVMemOptions<Val> = {}
  // source: I.Source = genSource(),
  // initialVersion: I.Version = 0
): Promise<I.Store<Val>> {
  const inner = storeOpts.inner || opmem()
  const data: Map<I.Key, Val> = _data == null ? new Map() : _data
  const lastModVersion = new Map<I.Key, I.Version>()

  if (inner.storeInfo.sources.length !== 1) throw Error('Using an inner store with more than one source not currently supported')
  // const source = inner.storeInfo.sources[0]
  let lastVersion: I.Version = V_EMPTY

  let nextOpP = resolvablePromise()

  let closed = false

  let ready: Resolvable<void> = resolvablePromise()

  const fetch: I.FetchFn<Val> = async (query, opts = {}) => {
    if (ready) await ready
    // console.log('ready lv', lastVersion)

    // We can technically return anything if no minVersion is set.
    // Uncomment this for jerk mode where we do the minimum possible. Tests should still pass.
    // if (!opts.minVersion && !opts.atVersion && query.type === I.QueryType.KV) {
    //   return {results: new Map(), versions: [{from:V64(0), to:V64(0)}]}
    // }

    // console.log('fetch query', query)

    let results: Map<I.Key, Val> | I.RangeResult<Val>
    let lowerRange: Uint8Array = V64(0)//V_EMPTY
    let bakedQuery: I.Query | undefined
    const tag = (k: I.Key) => {
      const v = lastModVersion.get(k) || lastVersion
      lowerRange = vMax(lowerRange, v)
    }

    switch (query.type) {
      case I.QueryType.KV:
        results = resultMap.filter(data, query.q)
        for (const k of query.q) tag(k)
        break

      case I.QueryType.AllKV:
        results = new Map(data)
        lowerRange = lastVersion // Dodgy...
        break

      case I.QueryType.Range:
        bakedQuery = {type: I.QueryType.StaticRange, q: query.q.slice()}
      case I.QueryType.StaticRange: {
        const q = query.q as I.RangeQuery
        results = [] as I.RangeResult<Val>

        // We're going to modify this to the baked range info.
        // Baked range info has no limit and no key offsets.

        // This is expensive, but ... kvmem was never designed for large
        // data sets. Could refactor kvmem to use a btree or something
        // internally if it becomes a legitimate issue for anyone. Or just
        // make a fancier in memory store.
        const keys = Array.from(data.keys()).sort()

        for (let i = 0; i < q.length; i++) {
          const qc = q[i]
          let from: number, to: number
          if (query.type === I.QueryType.Range) {
            [from, to, (bakedQuery!.q as I.StaticRangeQuery)[i]] = findRangeAndBake(qc, keys)
          } else {
            [from, to] = findRangeStatic(qc, keys)
          }
          const vals = keys.slice(from, to).map(k => {
            // This is a bit sneaky.
            tag(k)
            return ([k, data.get(k)] as I.KVPair<Val>)
          })
          results.push(qc.reverse ? vals.reverse() : vals)
        }
        break
      }
      default:
        return Promise.reject(new err.UnsupportedTypeError())
    }

    // const results = qtype === 'allkv' ? new Map(data) : resultMap.filter(data, query)

    return Promise.resolve({
      // this is a bit inefficient.... ehhhh
      bakedQuery,
      results: opts.noDocs ? queryTypes[query.type].resultType.map(results, () => true) : results,
      versions: [{from:lowerRange, to:lastVersion}],
    })
  }

  // Initialized by the time ready is resolves.
  let subgroup: SubGroup<Val> | null = null

//.........这里部分代码省略.........

  ;(async () => {
    for await (const cu of sub) {
      // console.log('cu', cu, closed)
      if (closed) break

      // We should be able to handle this actually.
      if (cu.replace) throw new Error('Unexpected replace data in catchup')
      
      const toV = cu.toVersion[0]
      
      for (let i = 0; i < cu.txns.length; i++) {
        const txn = cu.txns[i].txn as I.KVTxn<Val>
        for (const [k, op] of txn) lastModVersion.set(k, toV!)
        resultMap.applyMut!(data, txn)
      }
      
      if (subgroup == null) subgroup = new SubGroup({
        initialVersion: cu.toVersion,
        fetch,
        getOps: inner.getOps.bind(inner)
      })

      if (cu.txns.length) subgroup.onOp(0, lastVersion, cu.txns)
      if (toV != null) lastVersion = toV

      // console.log('rotate', lastVersion, cu)
      nextOpP.resolve()
      nextOpP = resolvablePromise<void>()

      if (cu.caughtUp) ready.resolve()
    }
  })()