node.js - ฉันจะทำการเชื่อมต่อฐานข้อมูล/เลือก/ปิดโดยใช้ RxJS observables ได้อย่างไร

ฉันกำลังใช้ node-oracledb เพื่อเชื่อมต่อกับฐานข้อมูล Oracle API มี promises ของตัวเอง ซึ่งสามารถถูกแปลงเป็น 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() ฉันใช้พรอมิสโดยตรงที่นี่ เพื่อหลีกเลี่ยงโค้ดที่มากขึ้น

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))
    );
}

คำตอบ #1

นี่คือวิธีที่ฉันจัดการการเชื่อมต่อโดยทั่วไป แกนหลักคือ using สร้าง observable ที่รับ factory ของทรัพยากรเป็นอาร์กิวเมนต์แรกและฟังก์ชันการตั้งค่าเป็นอาร์กิวเมนต์ที่สอง

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

resource คือวัตถุที่มีเมธอด unsubscribe ซึ่งจะถูกเรียกในขั้นตอนของการยกเลิกการสมัคร - ดังนั้นคุณสามารถซ่อนตรรกะใด ๆ ไว้ที่นั่นและผูกวงจรชีวิตของวัตถุใด ๆ กับวงจรชีวิตของ observable ได้อย่างมีประสิทธิภาพ

ฉันหวังว่าโค้ดด้านล่างจะสมเหตุสมผล

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)
    ...
  );

คำตอบ #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 ที่ส่งคืนโดยการดำเนินการเสร็จสิ้น