如何使用 RxJS observables 实现数据库连接/选择/关闭

Question

我正在使用 node-oracledb 连接到 Oracle 数据库。 API 提供了自己的 Promise，可以转换为 Promise<T>，因此“转换”为 Observable<T>。

使用 Observables，我想：

1. 打开数据库连接
2. 选择 N 条记录
3. 关闭数据库连接，即使 #2 抛出异常。

使用传统的、阻塞的、程序化的方式，它会是这样的：

try
{
    connection = Oracle.getConnection(...);
    resultSet = connection.execute("SELECT ... FROM ...");
}
catch (Exception)
{
    resultSet = EMPTY_RESULT;
}
finally
{
    if (connection)
        connection.close();
}

我尝试使用 Observables 编写此代码导致大量代码和回调。

受保护的方法 getConnection() 还是很简单的：

import * as Oracle from "oracledb";

protected getConnection() : Observable<IConnection>
{
    return OraUtil.from(Oracle.getConnection(this.parameters));
}

closeConnection() 方法也是如此。我在这里直接使用了 Promise，以避免更多的代码。

protected closeConnection(subscriber : Subscriber<IExecuteReturn>, connection : IConnection) : void
{
    connection.close()
        .then(() => subscriber.complete())
        .catch((error) => subscriber.error());
}

但是execute() 方法是麻烦开始的地方。

protected _execute(connection : IConnection, statement : string) : Observable<IExecuteReturn>
{
    return new Observable<IExecuteReturn>(
        (subscriber) => {
            OraUtil.from(connection.execute(statement)).subscribe(
                (result) => subscriber.next(result),
                (error) => {
                    subscriber.error(error);
                    this.closeConnection(subscriber, connection);
                },
                () => {
                    this.closeConnection(subscriber, connection);
                });
        });
}

public execute(statement : string) : Observable<IExecuteReturn>
{
    return this.getConnection().pipe(
        flatMap((connection) => this._execute(connection, statement))
    );
}

Answer 1

这就是我通常处理连接管理的方式。核心是using observable creator，它接受资源工厂作为第一个参数，设置函数作为第二个参数。

using(() => { unsubscribe() }, resource => observableOf(resource))

resource 是一个带有unsubscribe 方法的对象，它作为取消订阅的一部分被调用 - 因此您可以在其中隐藏任何逻辑并将任意对象的生命周期有效地绑定到可观察对象的生命周期。

我希望下面的代码有意义。

import * as Oracle from "oracledb";
import { mergeMap , ignoreElements} from 'rxjs/operators';
import { using } from 'rxjs/observable/using';
import { from as observableFrom } from 'rxjs/observable/from';
import { concat } from 'rxjs/observable/concat';
import { defer } from 'rxjs/observable/defer';
import { empty as observableEmpty } from 'rxjs/observable/empty';

class OracleConnection {
  constructor(parameters) {
    this.isClosed = false;
    this.connection = null;
    this.parameters = parameters;
  }

  setup() {
    return defer(() => Oracle.getConnection(this.parameters)
      .then(connection => { // do this in promise in case observable gets unsubscribed before connection is established
        this.connection = connection;
        if (this.isClosed) { // close in case connection got already closed before even got established
          this.terminate();
        }
        return connection;
      }));
  }

  close() {
    this.isClosed = true;
    if (this.connection !== null) {
      const connection = this.connection;
      this.connection = null;

      return observableFrom(connection.close())
        .pipe(ignoreElements()) // only propagate errors
    }

    return observableEmpty(); // connection already closed
  }
  
  terminate() {
    this.close().subscribe(/* handle error from connection close */);
  }

  unsubscribe() { // this will get called on observable unsubscribe
    if (!this.isClosed) {
      this.terminate();
    }
  }
}

class ConnectionManager {
  constructor(params) {
    this.params = params;
  }

  getConnection() {
    return using(() => new OracleConnection(this.params), oracleConnection => oracleConnection.setup())
  }
}

const manager = new ConnectionManager({ /* some params */ });

manager.getConnection()
  .pipe(
    mergeMap(connection => concat(
      connection.execute('SELECT 1'),
      connection.close() // explicitly close connection
    )),
    // alternatively
    // take(1) // to close connection automatically
  );

您可以做的很酷的事情是在失败的情况下轻松重试连接：

oracle.getConnection()
  .pipe(
    retry(3)
    ...
  );

Answer 2

一般来说，在使用 RxJs 6.x 时，类似“连接 + 执行 + 关闭”的顺序是按照这些思路实现的

let connection;

Oracle.getConnection(....)
.pipe(
  switchMap(conn => {
    connection = conn;
    return connection.execute(statement);
  }),
  finalize(() => connection.close())
)
.subscribe(
  result => resultSet = result,
  error => {
    console.error(error);
  }
)

语法细节可能不同，但关键思想是一旦连接 Observable 发出，然后你 switchMap 执行语句。订阅时会激活整个链。在订阅中，如果发生错误或执行返回的 Observable 完成时，您将关闭连接。