@arque/core - npm Package Compare versions

Comparing version 0.6.0 to 0.6.1

dist/libs/aggregate.d.ts

@@ -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>;

dist/libs/aggregate.js

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

package.json

 {
   "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"
 }

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

48449

increased by6.12%

Lines of code

790

increased by3.81%

No dependency changes