diff --git a/dist/index.d.ts b/dist/index.d.ts index 4965492..2257b5d 100644 --- a/dist/index.d.ts +++ b/dist/index.d.ts @@ -1,7 +1,5 @@ #!/usr/bin/env node declare global { - var SYNC_SUCCESS_EXIT_CODE: number; var CONFIG_DIR: string; - var SYNCING: boolean; } export {}; diff --git a/dist/lib/sync.js b/dist/lib/sync.js index 4ccceaa..e2ed967 100644 --- a/dist/lib/sync.js +++ b/dist/lib/sync.js @@ -36,16 +36,12 @@ var __importDefault = (this && this.__importDefault) || function (mod) { return (mod && mod.__esModule) ? mod : { "default": mod }; }; Object.defineProperty(exports, "__esModule", { value: true }); -const child_process_1 = require("child_process"); const files_1 = __importDefault(require("./watch/files")); const folders_1 = __importDefault(require("./watch/folders")); const get_last_edited_src_1 = __importDefault(require("../utils/get-last-edited-src")); const grab_folders_files_string_paths_1 = __importStar(require("../utils/grab-folders-files-string-paths")); const confFileProvidedJSON = process.argv[process.argv.length - 1]; -global.SYNC_SUCCESS_EXIT_CODE = 32; global.CONFIG_DIR = process.cwd(); -global.SYNCING = false; -const REWATCH_TIMEOUT = 1000; try { const configFileObject = JSON.parse(confFileProvidedJSON); const lastUpdated = (0, get_last_edited_src_1.default)({ @@ -88,17 +84,3 @@ catch (error) { console.log(error); process.exit(); } -process.on("exit", (code) => { - if (code == global.SYNC_SUCCESS_EXIT_CODE) { - const args = process.argv; - const cmd = args.shift(); - if (cmd) { - (0, child_process_1.spawn)(cmd, args, { - stdio: "inherit", - }); - } - } - else { - process.exit(0); - } -}); diff --git a/dist/lib/watch/files.js b/dist/lib/watch/files.js index 04fc850..b0ade6b 100644 --- a/dist/lib/watch/files.js +++ b/dist/lib/watch/files.js @@ -16,11 +16,19 @@ exports.default = watchFiles; const fs_1 = __importDefault(require("fs")); const delay_1 = __importDefault(require("../../utils/delay")); const sync_1 = __importDefault(require("../../utils/sync")); +const sync_scheduler_1 = __importDefault(require("../../utils/sync-scheduler")); function watchFiles(_a) { return __awaiter(this, arguments, void 0, function* ({ files, options, }) { - let timeout; const UPDATE_TIMEOUT = 1000; try { + const INTERVAL = (options === null || options === void 0 ? void 0 : options.interval) ? options.interval : UPDATE_TIMEOUT; + const scheduler = new sync_scheduler_1.default((filePath, firstRun) => (0, sync_1.default)({ + options, + dirPath: filePath, + dirs: files, + isFiles: true, + firstRun, + }), INTERVAL); for (let i = 0; i < files.length; i++) { const file = files[i]; const filePath = typeof file == "string" ? file : (file === null || file === void 0 ? void 0 : file.path) ? file.path : null; @@ -61,23 +69,7 @@ function watchFiles(_a) { fs_1.default.watchFile(filePath, { interval: interval || 200, }, (curr, prev) => { - if (global.SYNCING) - return; - const INTERVAL = (options === null || options === void 0 ? void 0 : options.interval) - ? options.interval - : UPDATE_TIMEOUT; - clearTimeout(timeout); - timeout = setTimeout(() => { - global.SYNCING = true; - (0, sync_1.default)({ - options, - dirPath: filePath, - dirs: files, - isFiles: true, - }).finally(() => { - process.exit(global.SYNC_SUCCESS_EXIT_CODE); - }); - }, INTERVAL); + scheduler.schedule(filePath); }); } } @@ -85,17 +77,7 @@ function watchFiles(_a) { const lastUpdatedFilePath = typeof lastUpdatedFile == "string" ? lastUpdatedFile : lastUpdatedFile.path; - global.SYNCING = true; - yield (0, sync_1.default)({ - dirPath: lastUpdatedFilePath, - dirs: files, - options, - isFiles: true, - firstRun: true, - }); - setTimeout(() => { - global.SYNCING = false; - }, UPDATE_TIMEOUT); + scheduler.enqueue(lastUpdatedFilePath, true); } catch (error) { console.log("ERROR:", error.message); diff --git a/dist/lib/watch/folders.js b/dist/lib/watch/folders.js index f228506..9d8b614 100644 --- a/dist/lib/watch/folders.js +++ b/dist/lib/watch/folders.js @@ -14,20 +14,19 @@ var __importDefault = (this && this.__importDefault) || function (mod) { Object.defineProperty(exports, "__esModule", { value: true }); exports.default = watchFolders; const fs_1 = __importDefault(require("fs")); -const delay_1 = __importDefault(require("../../utils/delay")); const sync_1 = __importDefault(require("../../utils/sync")); +const sync_scheduler_1 = __importDefault(require("../../utils/sync-scheduler")); function watchFolders(_a) { return __awaiter(this, arguments, void 0, function* ({ folders, options, }) { - let timeout; const UPDATE_TIMEOUT = 1000; try { const dirs = folders; - console.log("global.SYNCING", global.SYNCING); console.log(`Now handling ${dirs.length} Directories`); /** * # Watch Directories */ const INTERVAL = (options === null || options === void 0 ? void 0 : options.interval) ? options.interval : UPDATE_TIMEOUT; + const scheduler = new sync_scheduler_1.default((dirPath, firstRun) => (0, sync_1.default)({ dirPath, dirs, options, firstRun }), INTERVAL); for (let i = 0; i < dirs.length; i++) { const dir = dirs[i]; if (!dir) { @@ -35,7 +34,6 @@ function watchFolders(_a) { continue; } const dirPath = typeof dir == "string" ? dir : dir.path; - console.log("global.SYNCING", global.SYNCING); if ((typeof dir == "string" && !fs_1.default.existsSync(dirPath)) || (typeof dir == "object" && dir.path && @@ -64,21 +62,9 @@ function watchFolders(_a) { } } if (typeof dir == "string") { - yield (0, delay_1.default)(); fs_1.default.watch(dirPath, { recursive: true }, (evt, fileName) => { console.log("Folder Changed", evt, fileName); - if (global.SYNCING) { - console.log("Existing Sync found. Returning ..."); - return; - } - clearTimeout(timeout); - timeout = setTimeout(() => { - console.log("Folder Syncing in progress ..."); - global.SYNCING = true; - (0, sync_1.default)({ dirPath, dirs, options }).finally(() => { - process.exit(global.SYNC_SUCCESS_EXIT_CODE); - }); - }, INTERVAL); + scheduler.schedule(dirPath); }); } } @@ -89,16 +75,7 @@ function watchFolders(_a) { const lastUpdatedDirPath = typeof lastUpdatedDir == "string" ? lastUpdatedDir : lastUpdatedDir.path; - global.SYNCING = true; - yield (0, sync_1.default)({ - dirPath: lastUpdatedDirPath, - dirs, - options, - firstRun: true, - }); - setTimeout(() => { - global.SYNCING = false; - }, UPDATE_TIMEOUT); + scheduler.enqueue(lastUpdatedDirPath, true); } catch (error) { console.log("ERROR:", error.message); diff --git a/dist/utils/sync-scheduler.d.ts b/dist/utils/sync-scheduler.d.ts new file mode 100644 index 0000000..86be4ca --- /dev/null +++ b/dist/utils/sync-scheduler.d.ts @@ -0,0 +1,13 @@ +type SyncTask = (dirPath: string, firstRun?: boolean) => Promise; +export default class SyncScheduler { + private readonly task; + private readonly debounceMs; + private pending; + private timers; + private flushing; + constructor(task: SyncTask, debounceMs: number); + schedule(dirPath: string): void; + enqueue(dirPath: string, firstRun?: boolean): void; + private flush; +} +export {}; diff --git a/dist/utils/sync-scheduler.js b/dist/utils/sync-scheduler.js new file mode 100644 index 0000000..8148bbd --- /dev/null +++ b/dist/utils/sync-scheduler.js @@ -0,0 +1,68 @@ +"use strict"; +var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) { + function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); } + return new (P || (P = Promise))(function (resolve, reject) { + function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } } + function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } } + function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); } + step((generator = generator.apply(thisArg, _arguments || [])).next()); + }); +}; +Object.defineProperty(exports, "__esModule", { value: true }); +class SyncScheduler { + constructor(task, debounceMs) { + this.task = task; + this.debounceMs = debounceMs; + this.pending = new Map(); + this.timers = new Map(); + this.flushing = false; + } + schedule(dirPath) { + const existing = this.timers.get(dirPath); + if (existing) + clearTimeout(existing); + const timer = setTimeout(() => { + this.timers.delete(dirPath); + this.pending.set(dirPath, false); + void this.flush(); + }, this.debounceMs); + this.timers.set(dirPath, timer); + } + enqueue(dirPath, firstRun) { + const existing = this.timers.get(dirPath); + if (existing) { + clearTimeout(existing); + this.timers.delete(dirPath); + } + this.pending.set(dirPath, !!firstRun); + void this.flush(); + } + flush() { + return __awaiter(this, void 0, void 0, function* () { + if (this.flushing) + return; + this.flushing = true; + try { + while (this.pending.size > 0) { + const batch = [...this.pending.entries()]; + this.pending.clear(); + for (const [dirPath, firstRun] of batch) { + try { + yield this.task(dirPath, firstRun); + } + catch (error) { + console.log("ERROR:", error.message); + } + } + } + } + finally { + this.flushing = false; + if (this.pending.size > 0) { + void this.flush(); + } + } + }); + } +} +exports.default = SyncScheduler; diff --git a/dist/utils/sync.js b/dist/utils/sync.js index 27fd38b..2783563 100644 --- a/dist/utils/sync.js +++ b/dist/utils/sync.js @@ -15,10 +15,66 @@ Object.defineProperty(exports, "__esModule", { value: true }); exports.default = sync; const fs_1 = __importDefault(require("fs")); const path_1 = __importDefault(require("path")); +const os_1 = __importDefault(require("os")); const util_1 = __importDefault(require("util")); +const crypto_1 = __importDefault(require("crypto")); const child_process_1 = require("child_process"); const grab_dir_names_1 = __importDefault(require("./grab-dir-names")); +const grab_folders_files_string_paths_1 = require("./grab-folders-files-string-paths"); +const delay_1 = __importDefault(require("./delay")); const execPromise = util_1.default.promisify(child_process_1.exec); +const LOCK_STALE_MS = 5 * 60 * 1000; +const LOCK_RETRY_MS = 100; +function lockPathFor(dirPath) { + const hash = crypto_1.default + .createHash("sha1") + .update(path_1.default.resolve(dirPath)) + .digest("hex") + .slice(0, 16); + return path_1.default.join(os_1.default.tmpdir(), `turbosync-${hash}.lock`); +} +function acquireLock(lockPath) { + return __awaiter(this, void 0, void 0, function* () { + while (true) { + try { + const fd = fs_1.default.openSync(lockPath, "wx"); + fs_1.default.writeFileSync(fd, String(process.pid)); + fs_1.default.closeSync(fd); + return; + } + catch (error) { + if (error.code !== "EEXIST") + throw error; + try { + const { mtimeMs } = fs_1.default.statSync(lockPath); + if (Date.now() - mtimeMs > LOCK_STALE_MS) { + fs_1.default.unlinkSync(lockPath); + continue; + } + } + catch (_a) { + continue; + } + yield (0, delay_1.default)(LOCK_RETRY_MS); + } + } + }); +} +function acquireLocks(lockPaths) { + return __awaiter(this, void 0, void 0, function* () { + for (const lockPath of lockPaths) { + yield acquireLock(lockPath); + } + }); +} +function releaseLocks(lockPaths) { + for (const lockPath of lockPaths) { + try { + fs_1.default.unlinkSync(lockPath); + } + catch (_a) { } + } +} function sync(_a) { return __awaiter(this, arguments, void 0, function* ({ options, dirs, dirPath, isFiles, firstRun, }) { var _b, _c; @@ -78,9 +134,19 @@ function sync(_a) { } allCommandsArr.push(cmdArray); } - yield Promise.all(allCommandsArr.map((cmdArr) => { - return execPromise(cmdArr.join(" ")); - })); + const lockPaths = [dirPath, ...dstDirs.map((dr) => (0, grab_folders_files_string_paths_1.fldFileToStr)(dr))] + .filter((pth) => Boolean(pth)) + .map((pth) => lockPathFor(pth)) + .sort(); + yield acquireLocks(lockPaths); + try { + yield Promise.all(allCommandsArr.map((cmdArr) => { + return execPromise(cmdArr.join(" ")); + })); + } + finally { + releaseLocks(lockPaths); + } console.log(`${dirPath} Folder Sync Complete. Exiting ...`); }); } diff --git a/index.ts b/index.ts index 1a7a37f..d219d26 100644 --- a/index.ts +++ b/index.ts @@ -7,9 +7,7 @@ import handleEnvVars from "./utils/env"; import { TurboSyncConfigArray } from "./types"; declare global { - var SYNC_SUCCESS_EXIT_CODE: number; var CONFIG_DIR: string; - var SYNCING: boolean; } const confFileProvidedPath = process.argv[process.argv.length - 1]; diff --git a/lib/sync.ts b/lib/sync.ts index 824d241..f23926a 100644 --- a/lib/sync.ts +++ b/lib/sync.ts @@ -1,4 +1,3 @@ -import { spawn } from "child_process"; import watchFiles from "./watch/files"; import watchFolders from "./watch/folders"; import { TurboSyncConfigObject } from "../types"; @@ -9,11 +8,7 @@ import fldFileToStrArr, { const confFileProvidedJSON = process.argv[process.argv.length - 1]; -global.SYNC_SUCCESS_EXIT_CODE = 32; global.CONFIG_DIR = process.cwd(); -global.SYNCING = false; - -const REWATCH_TIMEOUT = 1000; try { const configFileObject: TurboSyncConfigObject = @@ -70,17 +65,3 @@ try { console.log(error); process.exit(); } - -process.on("exit", (code) => { - if (code == global.SYNC_SUCCESS_EXIT_CODE) { - const args = process.argv; - const cmd = args.shift(); - if (cmd) { - spawn(cmd, args, { - stdio: "inherit", - }); - } - } else { - process.exit(0); - } -}); diff --git a/lib/watch/files.ts b/lib/watch/files.ts index eeb4109..672c71a 100644 --- a/lib/watch/files.ts +++ b/lib/watch/files.ts @@ -3,15 +3,29 @@ import delay from "../../utils/delay"; import { SyncFilesFnParams } from "../../types"; import sync from "../../utils/sync"; +import SyncScheduler from "../../utils/sync-scheduler"; export default async function watchFiles({ files, options, }: SyncFilesFnParams) { - let timeout: any; const UPDATE_TIMEOUT = 1000; try { + const INTERVAL = options?.interval ? options.interval : UPDATE_TIMEOUT; + + const scheduler = new SyncScheduler( + (filePath, firstRun) => + sync({ + options, + dirPath: filePath, + dirs: files, + isFiles: true, + firstRun, + }), + INTERVAL + ); + for (let i = 0; i < files.length; i++) { const file = files[i]; const filePath = @@ -61,26 +75,7 @@ export default async function watchFiles({ interval: interval || 200, }, (curr, prev) => { - if (global.SYNCING) return; - - const INTERVAL = options?.interval - ? options.interval - : UPDATE_TIMEOUT; - - clearTimeout(timeout); - - timeout = setTimeout(() => { - global.SYNCING = true; - - sync({ - options, - dirPath: filePath, - dirs: files, - isFiles: true, - }).finally(() => { - process.exit(global.SYNC_SUCCESS_EXIT_CODE); - }); - }, INTERVAL); + scheduler.schedule(filePath); } ); } @@ -92,19 +87,7 @@ export default async function watchFiles({ ? lastUpdatedFile : lastUpdatedFile.path; - global.SYNCING = true; - - await sync({ - dirPath: lastUpdatedFilePath, - dirs: files, - options, - isFiles: true, - firstRun: true, - }); - - setTimeout(() => { - global.SYNCING = false; - }, UPDATE_TIMEOUT); + scheduler.enqueue(lastUpdatedFilePath, true); } catch (error: any) { console.log("ERROR:", error.message); process.exit(0); diff --git a/lib/watch/folders.ts b/lib/watch/folders.ts index 4d9fa63..b7146dd 100644 --- a/lib/watch/folders.ts +++ b/lib/watch/folders.ts @@ -1,19 +1,17 @@ import fs from "fs"; -import delay from "../../utils/delay"; import { SyncFoldersFnParams } from "../../types"; import sync from "../../utils/sync"; +import SyncScheduler from "../../utils/sync-scheduler"; export default async function watchFolders({ folders, options, }: SyncFoldersFnParams) { - let timeout: any; const UPDATE_TIMEOUT = 1000; try { const dirs = folders; - console.log("global.SYNCING", global.SYNCING); console.log(`Now handling ${dirs.length} Directories`); /** @@ -21,6 +19,11 @@ export default async function watchFolders({ */ const INTERVAL = options?.interval ? options.interval : UPDATE_TIMEOUT; + const scheduler = new SyncScheduler( + (dirPath, firstRun) => sync({ dirPath, dirs, options, firstRun }), + INTERVAL + ); + for (let i = 0; i < dirs.length; i++) { const dir = dirs[i]; @@ -31,8 +34,6 @@ export default async function watchFolders({ const dirPath = typeof dir == "string" ? dir : dir.path; - console.log("global.SYNCING", global.SYNCING); - if ( (typeof dir == "string" && !fs.existsSync(dirPath)) || (typeof dir == "object" && @@ -70,27 +71,10 @@ export default async function watchFolders({ } if (typeof dir == "string") { - await delay(); - fs.watch(dirPath, { recursive: true }, (evt, fileName) => { console.log("Folder Changed", evt, fileName); - if (global.SYNCING) { - console.log("Existing Sync found. Returning ..."); - return; - } - - clearTimeout(timeout); - - timeout = setTimeout(() => { - console.log("Folder Syncing in progress ..."); - - global.SYNCING = true; - - sync({ dirPath, dirs, options }).finally(() => { - process.exit(global.SYNC_SUCCESS_EXIT_CODE); - }); - }, INTERVAL); + scheduler.schedule(dirPath); }); } } @@ -104,18 +88,7 @@ export default async function watchFolders({ ? lastUpdatedDir : lastUpdatedDir.path; - global.SYNCING = true; - - await sync({ - dirPath: lastUpdatedDirPath, - dirs, - options, - firstRun: true, - }); - - setTimeout(() => { - global.SYNCING = false; - }, UPDATE_TIMEOUT); + scheduler.enqueue(lastUpdatedDirPath, true); } catch (error: any) { console.log("ERROR:", error.message); process.exit(0); diff --git a/package.json b/package.json index c5d750e..e5a101f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@moduletrace/turbosync", - "version": "1.2.5", + "version": "1.2.6", "module": "dist/index.js", "scripts": { "start": "node dist/index.js", diff --git a/utils/sync-scheduler.ts b/utils/sync-scheduler.ts new file mode 100644 index 0000000..3349e1d --- /dev/null +++ b/utils/sync-scheduler.ts @@ -0,0 +1,63 @@ +type SyncTask = (dirPath: string, firstRun?: boolean) => Promise; + +export default class SyncScheduler { + private pending = new Map(); + private timers = new Map(); + private flushing = false; + + constructor( + private readonly task: SyncTask, + private readonly debounceMs: number + ) {} + + schedule(dirPath: string) { + const existing = this.timers.get(dirPath); + if (existing) clearTimeout(existing); + + const timer = setTimeout(() => { + this.timers.delete(dirPath); + this.pending.set(dirPath, false); + void this.flush(); + }, this.debounceMs); + + this.timers.set(dirPath, timer); + } + + enqueue(dirPath: string, firstRun?: boolean) { + const existing = this.timers.get(dirPath); + if (existing) { + clearTimeout(existing); + this.timers.delete(dirPath); + } + + this.pending.set(dirPath, !!firstRun); + void this.flush(); + } + + private async flush() { + if (this.flushing) return; + + this.flushing = true; + + try { + while (this.pending.size > 0) { + const batch = [...this.pending.entries()]; + this.pending.clear(); + + for (const [dirPath, firstRun] of batch) { + try { + await this.task(dirPath, firstRun); + } catch (error: any) { + console.log("ERROR:", error.message); + } + } + } + } finally { + this.flushing = false; + + if (this.pending.size > 0) { + void this.flush(); + } + } + } +} diff --git a/utils/sync.ts b/utils/sync.ts index 0207911..04ee3be 100644 --- a/utils/sync.ts +++ b/utils/sync.ts @@ -1,12 +1,68 @@ import fs from "fs"; import path from "path"; +import os from "os"; import util from "util"; +import crypto from "crypto"; import { exec } from "child_process"; import { SyncFoldersSyncFnParams } from "../types"; import grabDirNames from "./grab-dir-names"; +import { fldFileToStr } from "./grab-folders-files-string-paths"; +import delay from "./delay"; const execPromise = util.promisify(exec); +const LOCK_STALE_MS = 5 * 60 * 1000; +const LOCK_RETRY_MS = 100; + +function lockPathFor(dirPath: string) { + const hash = crypto + .createHash("sha1") + .update(path.resolve(dirPath)) + .digest("hex") + .slice(0, 16); + + return path.join(os.tmpdir(), `turbosync-${hash}.lock`); +} + +async function acquireLock(lockPath: string) { + while (true) { + try { + const fd = fs.openSync(lockPath, "wx"); + fs.writeFileSync(fd, String(process.pid)); + fs.closeSync(fd); + return; + } catch (error: any) { + if (error.code !== "EEXIST") throw error; + + try { + const { mtimeMs } = fs.statSync(lockPath); + if (Date.now() - mtimeMs > LOCK_STALE_MS) { + fs.unlinkSync(lockPath); + continue; + } + } catch { + continue; + } + + await delay(LOCK_RETRY_MS); + } + } +} + +async function acquireLocks(lockPaths: string[]) { + for (const lockPath of lockPaths) { + await acquireLock(lockPath); + } +} + +function releaseLocks(lockPaths: string[]) { + for (const lockPath of lockPaths) { + try { + fs.unlinkSync(lockPath); + } catch {} + } +} + export default async function sync({ options, dirs, @@ -94,11 +150,22 @@ export default async function sync({ allCommandsArr.push(cmdArray); } - await Promise.all( - allCommandsArr.map((cmdArr) => { - return execPromise(cmdArr.join(" ")); - }) - ); + const lockPaths = [dirPath, ...dstDirs.map((dr) => fldFileToStr(dr))] + .filter((pth): pth is string => Boolean(pth)) + .map((pth) => lockPathFor(pth)) + .sort(); + + await acquireLocks(lockPaths); + + try { + await Promise.all( + allCommandsArr.map((cmdArr) => { + return execPromise(cmdArr.join(" ")); + }) + ); + } finally { + releaseLocks(lockPaths); + } console.log(`${dirPath} Folder Sync Complete. Exiting ...`); }