Oak-Team
/
oak-domain
13
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 33 Wiki Activity

重构了一下异步上下文的saveOpRecord行为国

This commit is contained in:
Xu Chang 2023-12-08 11:56:42 +08:00
parent 1d61db4592
commit 1fd17a6932
7 changed files with 139 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
lib/store/AsyncRowStore.d.ts vendored
View File

@ -3,11 +3,11 @@ import { EntityDict, RowStore, OperateOption, OperationResult, SelectOption, Con
import { IncomingHttpHeaders } from "http";
export declare abstract class AsyncContext<ED extends EntityDict> implements Context {
rowStore: AsyncRowStore<ED, this>;
clusterInfo?: ClusterInfo;
private uuid?;
opRecords: OpRecord<ED>[];
private scene?;
private headers?;
headers?: IncomingHttpHeaders;
clusterInfo?: ClusterInfo;
private message?;
events: {
commit: Array<() => Promise<void>>;
@ -18,12 +18,12 @@ export declare abstract class AsyncContext<ED extends EntityDict> implements Con
*/
abstract refineOpRecords(): Promise<void>;
constructor(store: AsyncRowStore<ED, AsyncContext<ED>>, headers?: IncomingHttpHeaders, clusterInfo?: ClusterInfo);
setHeaders(headers: IncomingHttpHeaders): void;
getHeader(key: string): string | string[] | undefined;
getScene(): string | undefined;
setScene(scene?: string): void;
private resetEvents;
on(event: 'commit' | 'rollback', callback: () => Promise<void>): void;
saveOpRecord<T extends keyof ED>(entity: T, operation: ED[T]['Operation']): void;
/**
* 一个context中不应该有并发的事务，这里将事务串行化，使用的时候千万要注意不要自己等自己
* @param options

39
lib/store/AsyncRowStore.js
View File

@ -2,14 +2,15 @@
Object.defineProperty(exports, "__esModule", { value: true });
exports.AsyncContext = void 0;
const tslib_1 = require("tslib");
const action_1 = require("../actions/action");
const assert_1 = tslib_1.__importDefault(require("assert"));
class AsyncContext {
rowStore;
clusterInfo;
uuid;
opRecords;
scene;
headers;
clusterInfo;
message;
events;
constructor(store, headers, clusterInfo) {
@ -24,9 +25,6 @@ class AsyncContext {
this.headers = headers;
}
}
setHeaders(headers) {
this.headers = headers;
}
getHeader(key) {
if (this.headers) {
return this.headers[key];
@ -47,6 +45,39 @@ class AsyncContext {
on(event, callback) {
this.uuid && this.events[event].push(callback);
}
saveOpRecord(entity, operation) {
const { action, data, filter, id } = operation;
switch (action) {
case 'create': {
this.opRecords.push({
id,
a: 'c',
e: entity,
d: data
});
break;
}
case 'remove': {
this.opRecords.push({
id,
a: 'r',
e: entity,
f: filter,
});
break;
}
default: {
(0, assert_1.default)(!action_1.readOnlyActions.includes(action));
this.opRecords.push({
id,
a: 'u',
e: entity,
d: data,
f: filter,
});
}
}
}
/**
* 一个context中不应该有并发的事务，这里将事务串行化，使用的时候千万要注意不要自己等自己
* @param options

41
lib/store/CascadeStore.js
View File

@ -1133,11 +1133,12 @@ class CascadeStore extends RowStore_1.RowStore {
await createInner(operation);
}
if (!option.dontCollect) {
context.opRecords.push({
context.saveOpRecord(entity, operation);
/* context.opRecords.push({
a: 'c',
e: entity,
d: data,
});
d: data as ED[T]['OpSchema'] | ED[T]['OpSchema'][],
}); */
}
if (!option.dontCreateOper && !['oper', 'operEntity', 'modiEntity', 'modi'].includes(entity)) {
// 按照框架要求生成Oper和OperEntity这两个内置的对象
@ -1319,7 +1320,17 @@ class CascadeStore extends RowStore_1.RowStore {
};
if (action === 'remove') {
if (!option.dontCollect) {
context.opRecords.push({
context.saveOpRecord(entity, {
id: operId,
action,
data: {},
filter: {
id: {
$in: ids,
}
}
});
/* context.opRecords.push({
a: 'r',
e: entity,
f: {
@ -1327,7 +1338,7 @@ class CascadeStore extends RowStore_1.RowStore {
$in: ids,
}
},
});
}); */
}
}
else {
@ -1338,16 +1349,26 @@ class CascadeStore extends RowStore_1.RowStore {
$$updateAt$$: now,
});
if (!option.dontCollect) {
context.opRecords.push({
a: 'u',
e: entity,
d: data,
f: {
context.saveOpRecord(entity, {
id: operId,
action,
data: data,
filter: {
id: {
$in: ids,
}
},
});
/* context.opRecords.push({
a: 'u',
e: entity,
d: data as ED[T]['Update']['data'],
f: {
id: {
$in: ids,
}
},
}); */
}
}
else if (action !== 'update') {

3
lib/types/Entity.d.ts vendored
View File

@ -143,17 +143,20 @@ type RemoveOperationData = {
export type RemoveOperation = Operation<'remove', RemoveOperationData, Filter, Sorter>;
export type CUDOperation = CreateOperation | UpdateOperation | RemoveOperation;
export type CreateOpResult<ED extends EntityDict, T extends keyof ED> = {
id?: string;
a: 'c';
e: T;
d: ED[T]['OpSchema'] | ED[T]['OpSchema'][];
};
export type UpdateOpResult<ED extends EntityDict, T extends keyof ED> = {
id?: string;
a: 'u';
e: T;
d: UpdateOperationData;
f?: Filter;
};
export type RemoveOpResult<ED extends EntityDict, T extends keyof ED> = {
id?: string;
a: 'r';
e: T;
f?: Filter;

43
src/store/AsyncRowStore.ts
View File

@ -1,15 +1,16 @@
import { EntityDict, RowStore, OperateOption, OperationResult, SelectOption, Context, TxnOption, OpRecord, AggregationResult, ClusterInfo } from "../types";
import { readOnlyActions } from '../actions/action';
import assert from "assert";
import { IncomingHttpHeaders } from "http";
export abstract class AsyncContext<ED extends EntityDict> implements Context {
rowStore: AsyncRowStore<ED, this>;
clusterInfo?: ClusterInfo;
private uuid?: string;
opRecords: OpRecord<ED>[];
private scene?: string;
private headers?: IncomingHttpHeaders;
headers?: IncomingHttpHeaders;
clusterInfo?: ClusterInfo;
private message?: string;
events: {
commit: Array<() => Promise<void>>;
@ -34,10 +35,6 @@ export abstract class AsyncContext<ED extends EntityDict> implements Context {
}
}
setHeaders(headers: IncomingHttpHeaders) {
this.headers = headers;
}
getHeader(key: string): string | string[] | undefined {
if (this.headers) {
return this.headers[key];
@ -61,6 +58,40 @@ export abstract class AsyncContext<ED extends EntityDict> implements Context {
this.uuid && this.events[event].push(callback);
}
saveOpRecord<T extends keyof ED>(entity: T, operation: ED[T]['Operation']) {
const { action, data, filter, id } = operation;
switch (action) {
case 'create': {
this.opRecords.push({
id,
a: 'c',
e: entity,
d: data as ED[T]['OpSchema']
});
break;
}
case 'remove': {
this.opRecords.push({
id,
a: 'r',
e: entity,
f: filter,
});
break;
}
default: {
assert(!readOnlyActions.includes(action));
this.opRecords.push({
id,
a: 'u',
e: entity,
d: data as ED[T]['Update']['data'],
f: filter,
});
}
}
}
/**
* 一个context中不应该有并发的事务，这里将事务串行化，使用的时候千万要注意不要自己等自己
* @param options

33
src/store/CascadeStore.ts
View File

@ -1382,11 +1382,12 @@ export abstract class CascadeStore<ED extends EntityDict & BaseEntityDict> exten
}
if (!option.dontCollect) {
context.opRecords.push({
context.saveOpRecord(entity, operation);
/* context.opRecords.push({
a: 'c',
e: entity,
d: data as ED[T]['OpSchema'] | ED[T]['OpSchema'][],
});
}); */
}
if (!option.dontCreateOper && !['oper', 'operEntity', 'modiEntity', 'modi'].includes(entity as string)) {
// 按照框架要求生成Oper和OperEntity这两个内置的对象
@ -1581,7 +1582,17 @@ export abstract class CascadeStore<ED extends EntityDict & BaseEntityDict> exten
};
if (action === 'remove') {
if (!option.dontCollect) {
context.opRecords.push({
context.saveOpRecord(entity, {
id: operId,
action,
data: {},
filter: {
id: {
$in: ids,
}
}
});
/* context.opRecords.push({
a: 'r',
e: entity,
f: {
@ -1589,7 +1600,7 @@ export abstract class CascadeStore<ED extends EntityDict & BaseEntityDict> exten
$in: ids,
}
},
});
}); */
}
}
else {
@ -1600,7 +1611,17 @@ export abstract class CascadeStore<ED extends EntityDict & BaseEntityDict> exten
$$updateAt$$: now,
});
if (!option.dontCollect) {
context.opRecords.push({
context.saveOpRecord(entity, {
id: operId,
action,
data: data as ED[T]['Update']['data'],
filter: {
id: {
$in: ids,
}
},
});
/* context.opRecords.push({
a: 'u',
e: entity,
d: data as ED[T]['Update']['data'],
@ -1609,7 +1630,7 @@ export abstract class CascadeStore<ED extends EntityDict & BaseEntityDict> exten
$in: ids,
}
},
});
}); */
}
}
else if (action !== 'update') {

3
src/types/Entity.ts
View File

@ -187,12 +187,14 @@ export type RemoveOperation = Operation<'remove', RemoveOperationData, Filter, S
export type CUDOperation = CreateOperation | UpdateOperation | RemoveOperation;
export type CreateOpResult<ED extends EntityDict, T extends keyof ED> = {
id?: string;
a: 'c';
e: T;
d: ED[T]['OpSchema'] | ED[T]['OpSchema'][];
};
export type UpdateOpResult<ED extends EntityDict, T extends keyof ED> = {
id?: string;
a: 'u',
e: T;
d: UpdateOperationData;
@ -200,6 +202,7 @@ export type UpdateOpResult<ED extends EntityDict, T extends keyof ED> = {
};
export type RemoveOpResult<ED extends EntityDict, T extends keyof ED> = {
id?: string;
a: 'r',
e: T;
f?: Filter;