使用註解簡化 Mongoose 事務的使用

mongoose 提供的事務

MongoDB 4.0 開始提供了事務支持，mongoose 也提供了相應的實現，不過目前的寫法仍是比較繁瑣。
咱們看一下 mongoose 給出的 demo

const Customer = db.model('Customer', new Schema({ name: String }));

let session = null;
return Customer.createCollection().
  then(() => db.startSession()).
  then(_session => {
    session = _session;
    // Start a transaction
    session.startTransaction();
    // This `create()` is part of the transaction because of the `session`
    // option.
    return Customer.create([{ name: 'Test' }], { session: session });
  }).
  // Transactions execute in isolation, so unless you pass a `session`
  // to `findOne()` you won't see the document until the transaction
  // is committed.
  then(() => Customer.findOne({ name: 'Test' })).
  then(doc => assert.ok(!doc)).
  // This `findOne()` will return the doc, because passing the `session`
  // means this `findOne()` will run as part of the transaction.
  then(() => Customer.findOne({ name: 'Test' }).session(session)).
  then(doc => assert.ok(doc)).
  // Once the transaction is committed, the write operation becomes
  // visible outside of the transaction.
  then(() => session.commitTransaction()).
  then(() => Customer.findOne({ name: 'Test' })).
  then(doc => assert.ok(doc));

這個 demo 暴露了兩個問題：

1. 須要爲每個事務裏作提交和回滾的處理
2. 事務是用 session 來區分的，你須要一直傳遞 session

使用註解

因此 akajs 提供了一個事務的註解來簡化這個處理流程。

import * as mongoose from 'mongoose'
import {Schema} from 'mongoose'
import * as assert from 'assert'
import {Transactional, getSession} from './decorators/Transactional'

mongoose.connect('mongodb://localhost:27017,localhost:27018,localhost:27019/test?replicaSet=rs', {useNewUrlParser: true})
mongoose.set('debug', true)
let db = mongoose.connection
const Customer = db.model('Customer', new Schema({name: String}))

class ClassA {
  @Transactional()
  async main (key) {
    await new Customer({name: 'ClassA'}).save({session: getSession()})
    const doc1 = await Customer.findOne({name: 'ClassA'})
    assert.ok(!doc1)
    await new ClassB().step2()
    return key
  }
}

class ClassB {
  async step2 () {
    const doc2 = await Customer.findOne({name: 'ClassA'}).session(getSession())
    assert.ok(doc2)
    await Customer.remove({}).session(getSession())
  }
}

new ClassA().main('aaa').then((res) => {
  console.log('res', res)
  mongoose.disconnect(console.log)
}).catch(console.error)

解析：

• @Transactional() 註解會自動提交或回滾事務（發生異常時）。具體實現見Transactional.ts ，核心實現部分，使用 try catch 捕獲異常，實現自動回滾。

try {
  const value = await originalMethod.apply(this, [...args])
  // Since the mutations ran without an error, commit the transaction.
  await session.commitTransaction()
  // Return any value returned by `mutations` to make this function as transparent as possible.
  return value
} catch (error) {
  // Abort the transaction as an error has occurred in the mutations above.
  await session.abortTransaction()
  // Rethrow the error to be caught by the caller.
  throw error
} finally {
  // End the previous session.
  session.endSession()
}

• 爲了不嵌套調用時，你須要一直傳遞 session 的尷尬~，akajs 提供全局的 getSession() 方法，其實現原理是依賴 Async Hooks ，是 Node 的實驗性特性，

你對此介意的話，請不要在生產環境使用。

注意 mongodb 的事務必須在複製集上使用，在開發環境啓動 mongodb 複製集，推薦使用 run-rs

更進一步

固然，在每個須要 Session 的地方調用 getSession() 方法仍是稍顯累贅，咱們能夠經過 wrap mongoose 的各個方法，來實現自動注入 session。

例如把 mongoose 的 findOne 方法替換爲

let originFindOne = mongoose.findOne
mongoose.findOne = () => {
originFindOne().session(getSession())
}

可是工做量有些多，暫時沒時間作。