## 0.24.4

### Fixed

-   Fixed so many inefficient and buggy modules inherited from the past.

### Improved

-   Tasks are now executed in an efficient asynchronous library.
-   On-demand chunk fetching is now more efficient and keeps the interval between requests.
    -   This will reduce the load on the server and the network.
    -   And, safe for the Cloudant.
This commit is contained in:
vorotamoroz committed 2024-12-12 11:10:50 +00:00
1 parent 3bb4aba395
commit 21861d8c51
2 files changed
+21 -28

No files matched your search

+3 -4
View File
@@ -13,15 +13,16 @@ import {
VER, VER,
type EntryBody, type EntryBody,
type EntryDoc, type EntryDoc,
type EntryLeaf,
type LoadedEntry, type LoadedEntry,
type MetaEntry, type MetaEntry,
} from "../../lib/src/common/types"; } from "../../lib/src/common/types";
import { QueueProcessor } from "octagonal-wheels/concurrency/processor"; import { QueueProcessor } from "octagonal-wheels/concurrency/processor";
import { getPath, isChunk, isValidPath, scheduleTask } from "../../common/utils"; import { getPath, isChunk, isValidPath, scheduleTask } from "../../common/utils";
import { sendValue } from "octagonal-wheels/messagepassing/signal";
import { isAnyNote } from "../../lib/src/common/utils"; import { isAnyNote } from "../../lib/src/common/utils";
import { EVENT_FILE_SAVED, eventHub } from "../../common/events"; import { EVENT_FILE_SAVED, eventHub } from "../../common/events";
import type { LiveSyncAbstractReplicator } from "../../lib/src/replication/LiveSyncAbstractReplicator"; import type { LiveSyncAbstractReplicator } from "../../lib/src/replication/LiveSyncAbstractReplicator";
import { globalSlipBoard } from "../../lib/src/bureau/bureau";
export class ModuleReplicator extends AbstractModule implements ICoreModule { export class ModuleReplicator extends AbstractModule implements ICoreModule {
$everyOnloadAfterLoadSettings(): Promise<boolean> { $everyOnloadAfterLoadSettings(): Promise<boolean> {
@@ -242,9 +243,7 @@ Or if you are sure know what had been happened, we can unlock the database from
const change = docs[0]; const change = docs[0];
if (!change) return; if (!change) return;
if (isChunk(change._id)) { if (isChunk(change._id)) {
// SendSignal? globalSlipBoard.submit("read-chunk", change._id, change as EntryLeaf);
// this.parseIncomingChunk(change);
sendValue(`leaf-${change._id}`, change);
return; return;
} }
if (await this.core.$anyModuleParsedReplicationResultItem(change)) return; if (await this.core.$anyModuleParsedReplicationResultItem(change)) return;
+18 -24
View File
@@ -1,5 +1,4 @@
import { unique } from "octagonal-wheels/collection"; import { unique } from "octagonal-wheels/collection";
import { QueueProcessor } from "octagonal-wheels/concurrency/processor";
import { throttle } from "octagonal-wheels/function"; import { throttle } from "octagonal-wheels/function";
import { eventHub } from "../../common/events.ts"; import { eventHub } from "../../common/events.ts";
import { BASE_IS_NEW, compareFileFreshness, EVEN, getPath, isValidPath, TARGET_IS_NEW } from "../../common/utils.ts"; import { BASE_IS_NEW, compareFileFreshness, EVEN, getPath, isValidPath, TARGET_IS_NEW } from "../../common/utils.ts";
@@ -19,7 +18,7 @@ import { isAnyNote } from "../../lib/src/common/utils.ts";
import { stripAllPrefixes } from "../../lib/src/string_and_binary/path.ts"; import { stripAllPrefixes } from "../../lib/src/string_and_binary/path.ts";
import { AbstractModule } from "../AbstractModule.ts"; import { AbstractModule } from "../AbstractModule.ts";
import type { ICoreModule } from "../ModuleTypes.ts"; import type { ICoreModule } from "../ModuleTypes.ts";
import { withConcurrency } from "octagonal-wheels/iterable/map";
export class ModuleInitializerFile extends AbstractModule implements ICoreModule { export class ModuleInitializerFile extends AbstractModule implements ICoreModule {
async $$performFullScan(showingNotice?: boolean): Promise<void> { async $$performFullScan(showingNotice?: boolean): Promise<void> {
this._log("Opening the key-value database", LOG_LEVEL_VERBOSE); this._log("Opening the key-value database", LOG_LEVEL_VERBOSE);
@@ -152,35 +151,30 @@ export class ModuleInitializerFile extends AbstractModule implements ICoreModule
if (!this.localDatabase.isReady) throw Error("Database is not ready!"); if (!this.localDatabase.isReady) throw Error("Database is not ready!");
let success = 0; let success = 0;
let failed = 0; let failed = 0;
const step = 10; let total = 0;
const processor = new QueueProcessor( for await (const result of withConcurrency(
objects,
async (e) => { async (e) => {
try { try {
await callback(e[0]); await callback(e);
success++; return true;
// return
} catch (ex) { } catch (ex) {
this._log(`Error while ${procedureName}`, LOG_LEVEL_NOTICE); this._log(`Error while ${procedureName}`, LOG_LEVEL_NOTICE);
this._log(ex, LOG_LEVEL_VERBOSE); this._log(ex, LOG_LEVEL_VERBOSE);
failed++; return false;
} }
if ((success + failed) % step == 0) {
const msg = `${procedureName}: DONE:${success}, FAILED:${failed}, LAST:${processor._queue.length}`;
updateLog(procedureName, msg);
}
return;
}, },
{ 10
batchSize: 1, )) {
concurrentLimit: 10, if (result) {
delay: 0, success++;
suspended: true, } else {
maintainDelay: false, failed++;
interval: 0, }
}, total++;
objects const msg = `${procedureName}: DONE:${success}, FAILED:${failed}, LAST:${objects.length - total}`;
); updateLog(procedureName, msg);
await processor.waitForAllDoneAndTerminate(); }
const msg = `${procedureName} All done: DONE:${success}, FAILED:${failed}`; const msg = `${procedureName} All done: DONE:${success}, FAILED:${failed}`;
updateLog(procedureName, msg); updateLog(procedureName, msg);
}; };