runInXaTransaction<T extends Object> method

Future<Result<T>> runInXaTransaction<T extends Object>(String connectionId, Xid xid, Future<Result<T>> action(XaTransactionHandle xa), { bool onePhase = false, })

Implementation

Future<Result<T>> runInXaTransaction<T extends Object>(
  String connectionId,
  Xid xid,
  Future<Result<T>> Function(XaTransactionHandle xa) action, {
  bool onePhase = false,
}) =>
    OdbcErrorBoundary.run('runInXaTransaction', () async {
      final start = await xaStart(connectionId, xid);
      if (start.isError()) return Failure(start.exceptionOrNull()!);
      final xa = start.getOrNull()!;
      final userResult =
          await OdbcErrorBoundary.run('xaAction', () => action(xa));
      if (userResult.isError()) {
        return Failure(
          await _xaAbort(
            xa,
            normalizeOdbcError(
              userResult.exceptionOrNull()!,
              operation: 'xaAction',
            ),
          ),
        );
      }
      if (xa.state == XaState.committed) return userResult;
      if (xa.state == XaState.rolledBack ||
          xa.commitAttempted ||
          xa.outcomeUnknown ||
          xa.state == XaState.failedAfterPrepare ||
          xa.state == XaState.failed) {
        return Failure(
          xa.lastError ??
              const ValidationError(
                message: 'The XA action did not leave an active transaction',
              ),
        );
      }
      Future<OdbcError?> phase(
        String operation,
        Future<bool> Function() invoke,
      ) async {
        try {
          if (await invoke()) return null;
          final original = xa.lastError ??
              QueryError(
                message: 'The XA transaction phase failed',
                details: OdbcErrorDetails(
                  operation: operation,
                  code: OdbcErrorCode.transaction,
                  transactionId: xid.toString(),
                  outcomeUnknown: operation == 'xaCommitPrepared' ||
                      operation == 'xaCommitOnePhase',
                ),
              );
          return original.withDetails(
            original.details.copyWith(
              operation: operation,
              code: OdbcErrorCode.transaction,
              transactionId: xid.toString(),
              outcomeUnknown: xa.outcomeUnknown,
            ),
          );
        } on Object catch (error, stack) {
          final primary = xa.lastError ??
              normalizeOdbcError(
                error,
                operation: operation,
                stackTrace: stack,
              );
          return primary.withDetails(
            primary.details.copyWith(
              transactionId: xid.toString(),
              outcomeUnknown: xa.outcomeUnknown,
            ),
          );
        }
      }

      if (onePhase) {
        final error = await phase('xaCommitOnePhase', xa.commitOnePhase);
        if (error != null) return Failure(error);
        return userResult;
      }
      if (xa.state == XaState.active) {
        final end = await phase('xaEnd', xa.end);
        if (end != null) return Failure(await _xaAbort(xa, end));
      }
      if (xa.state == XaState.idle) {
        final prepare = await phase('xaPrepare', xa.prepare);
        if (prepare != null) return Failure(await _xaAbort(xa, prepare));
      }
      final commit = await phase('xaCommitPrepared', xa.commitPrepared);
      if (commit != null) return Failure(commit);
      return userResult;
    });