@arque/core
Advanced tools
Comparing version 0.6.0 to 0.6.1
@@ -24,2 +24,3 @@ /// <reference types="node" /> | ||
private _state; | ||
private readonly logger; | ||
private mutex; | ||
@@ -43,3 +44,3 @@ private commandHandlers; | ||
process(command: ExtractCommand<TCommandHandler>, ctx?: Buffer, opts?: { | ||
noInitialReload?: true; | ||
noReload?: true; | ||
maxRetries?: number; | ||
@@ -46,0 +47,0 @@ }): Promise<void>; |
@@ -9,2 +9,3 @@ "use strict"; | ||
const assert_1 = tslib_1.__importDefault(require("assert")); | ||
const debug_1 = require("debug"); | ||
const event_id_1 = require("./event-id"); | ||
@@ -20,2 +21,9 @@ const store_adapter_1 = require("./adapters/store-adapter"); | ||
this._state = _state; | ||
this.logger = { | ||
info: (0, debug_1.debug)('info:Aggregate'), | ||
error: (0, debug_1.debug)('error:Aggregate'), | ||
warn: (0, debug_1.debug)('warn:Aggregate'), | ||
verbose: (0, debug_1.debug)('verbose:Aggregate'), | ||
debug: (0, debug_1.debug)('debug:Aggregate'), | ||
}; | ||
this._lastEvent = null; | ||
@@ -90,2 +98,4 @@ this.mutex = new async_mutex_1.Mutex(); | ||
async _reload() { | ||
let hrtime; | ||
hrtime = process.hrtime(); | ||
const snapshot = await this.store.findLatestSnapshot({ | ||
@@ -97,2 +107,3 @@ aggregate: { | ||
}); | ||
this.logger.debug(`findLatestSnapshot: elapsed=${Math.floor(process.hrtime(hrtime)[1] / 1e6)}ms`); | ||
if (snapshot) { | ||
@@ -102,3 +113,4 @@ this._state = this.opts.deserializeState(snapshot.state); | ||
} | ||
await this.digest(await this.store.listEvents({ | ||
hrtime = process.hrtime(); | ||
const events = await this.store.listEvents({ | ||
aggregate: { | ||
@@ -108,3 +120,5 @@ id: this.id, | ||
}, | ||
})); | ||
}); | ||
this.logger.debug(`listEvents: elapsed=${Math.floor(process.hrtime(hrtime)[1] / 1e6)}ms`); | ||
await this.digest(events); | ||
} | ||
@@ -121,3 +135,6 @@ async reload() { | ||
async dispatch(params, ctx) { | ||
let hrtime; | ||
hrtime = process.hrtime(); | ||
await this.store.saveEvents(params); | ||
this.logger.debug(`saveEvents: elapsed=${Math.floor(process.hrtime(hrtime)[1] / 1e6)}ms`); | ||
const events = params.events.map((item, index) => (Object.assign(Object.assign({}, item), { timestamp: params.timestamp, aggregate: { | ||
@@ -127,2 +144,3 @@ id: this.id, | ||
}, meta: Object.assign(Object.assign({}, item.meta), { __ctx: ctx }) }))); | ||
hrtime = process.hrtime(); | ||
await this.stream.sendEvents([ | ||
@@ -134,4 +152,6 @@ { | ||
]); | ||
this.logger.debug(`sendEvents: elapsed=${Math.floor(process.hrtime(hrtime)[1] / 1e6)}ms`); | ||
await this.digest(events); | ||
if (this.shoudTakeSnapshot()) { | ||
hrtime = process.hrtime(); | ||
await this.store.saveSnapshot({ | ||
@@ -145,2 +165,3 @@ aggregate: { | ||
}); | ||
this.logger.debug(`saveSnapshot: elapsed=${Math.floor(process.hrtime(hrtime)[1] / 1e6)}ms`); | ||
} | ||
@@ -152,9 +173,12 @@ } | ||
const release = await this.mutex.acquire(); | ||
let firstRun = true; | ||
let first = true; | ||
if ((opts === null || opts === void 0 ? void 0 : opts.noReload) !== true) { | ||
await this._reload(); | ||
} | ||
try { | ||
await (0, exponential_backoff_1.backOff)(async () => { | ||
if (!firstRun || !(opts === null || opts === void 0 ? void 0 : opts.noInitialReload)) { | ||
if (!first) { | ||
await this._reload(); | ||
} | ||
firstRun = false; | ||
first = false; | ||
const timestamp = new Date(); | ||
@@ -187,8 +211,12 @@ const event = await handler.handle({ | ||
jitter: 'full', | ||
maxDelay: 1600, | ||
numOfAttempts: (_a = opts === null || opts === void 0 ? void 0 : opts.maxRetries) !== null && _a !== void 0 ? _a : 10, | ||
maxDelay: 800, | ||
numOfAttempts: (_a = opts === null || opts === void 0 ? void 0 : opts.maxRetries) !== null && _a !== void 0 ? _a : 5, | ||
startingDelay: 100, | ||
timeMultiple: 2, | ||
retry(err) { | ||
return err instanceof store_adapter_1.AggregateVersionConflictError; | ||
retry: (err) => { | ||
if (err instanceof store_adapter_1.AggregateVersionConflictError) { | ||
this.logger.warn(`retrying: error="${err.message}"`); | ||
return true; | ||
} | ||
return false; | ||
}, | ||
@@ -195,0 +223,0 @@ }); |
{ | ||
"name": "@arque/core", | ||
"version": "0.6.0", | ||
"version": "0.6.1", | ||
"description": "`arque` core library", | ||
@@ -34,3 +34,3 @@ "author": "ScaleForge", | ||
}, | ||
"gitHead": "ac65ce278706a8ff357f716b258acd71b00312e0" | ||
"gitHead": "a911eff0ab1507813bd849f3f845a3643580b748" | ||
} |
Sorry, the diff of this file is not supported yet
License Policy Violation
LicenseThis package is not allowed per your license policy. Review the package's license to ensure compliance.
Found 1 instance in 1 package
License Policy Violation
LicenseThis package is not allowed per your license policy. Review the package's license to ensure compliance.
Found 1 instance in 1 package
48449
790