MentDB WeakMQL Recipes
← innov-ai.com All profiles Portal
Profiles / Programming
ED

Event developer

Event-driven scripts: file watchers that push each file event to the stack, validation, archiving, errors and safe restarts.

10 recipes on this page · open in the interactive view

Recipes
  1. Run a script when a file arrives in a folder
  2. Inspect the file watchers of the server
  3. The watcher validates, a heavy loader runs later in the stack
  4. The three events of a file: create, modify, delete
  5. Process a CSV on arrival and archive it
  6. A bad file ends in the error table of the stack
  7. Restart a watcher safely
  8. Wait for the READY flag before loading the files
  9. Route the dropped files by type and content
  10. Refuse a file received twice under two names
executed on MentDB

Run a script when a file arrives in a folder

Partners drop order files in tmp/tuto_as_in. A file watcher sends every event of that folder to a script through the stack; the script counts the lines of each new file and logs it. The demo drops a file, waits for the run, reads the log and stops the watcher.

Pipeline
01 · folderDrop folderfile mkdir→02 · watchFolder to scriptfile_watcher start→03 · dropA partner file arrivesfile create→04 · proofLog line of the runlog search→05 · stopStop watchingfile_watcher kill
Script
admin
#1. The drop folder and the script, it receives [kind] and [file];
file mkdir "tmp/tuto_as_in";
script create exe "tuto.as.on_drop" true 1
	(param
		(var "[kind]" {true} "Event kind sent by the watcher" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name sent by the watcher" is_null:true is_empty:true "order_1001.csv")
	;)
	"React to a file dropped in tmp/tuto_as_in"
{
	if (equal [kind] "ENTRY_CREATE") {
		-> "[lines]" (file count_lines (concat "tmp/tuto_as_in/" [file]));
		log write (concat [file] " received, " [lines] " lines") "OK" "file" [file];
		sequence increment "tuto_as_drops";
	};
} "1";
sequence add "tuto_as_drops" 0;
#2. Watch the folder;
-> "[t0]" (date now);
file_watcher start "tuto_as_inbox" "admin" "tmp/tuto_as_in" "tuto.as.on_drop.exe";
#3. A partner drops a file;
file create "tmp/tuto_as_in/order_1001.csv" (concat "sku;qty" [_n_] "A1;3" [_n_] "B7;5" [_n_] "C2;1");
-> "[i]" 0;
while (and (< [i] 45) (== (sequence get current "tuto_as_drops") 0)) {wait "1000"; ++ "[i]";};
#4. The proof in the logs;
json load "tuto_as_l" (string substring (log search "ok" "tuto.as.on_drop.exe" "file" "order_1001.csv" null [t0] (date dateaddt (date now) "MIN" 1) ASC 1 5) 39);
-> "[out]" (concat "run about " [i] " s after the drop: " (json select "tuto_as_l" "/data/[0]/msg") " (pid " (atom get_last (json select "tuto_as_l" "/data/[0]/pid") " ") ")");
#5. Stop watching, cleanup;
file_watcher kill "tuto_as_inbox";
-> "[out]" (concat [out] [_n_] "watcher still there: " (file_watcher exist "tuto_as_inbox"));
wait "1000";
stack delete_wait_script "tuto.as.on_drop.exe";
stack delete_closed_script "tuto.as.on_drop.exe";
file delete "tmp/tuto_as_in/order_1001.csv";
file delete "tmp/tuto_as_in";
json unload "tuto_as_l";
sequence remove "tuto_as_drops";
script delete "tuto.as.on_drop.exe";
[out]
Result
mentdb
run about 29 s after the drop: order_1001.csv received, 4 lines (pid 3r)
watcher still there: 0
Go further
  • The watcher also sends ENTRY_MODIFY and ENTRY_DELETE: test [kind] first.
  • The runs are normal stack runs: stack search "error" "tuto.as.on_drop.exe" ... shows the failed ones.
executed on MentDB

Inspect the file watchers of the server

Two watchers serve two drop folders. The developer checks that each one exists, lists only the watchers of the project with their folder, script and state, stops one and checks again.

Pipeline
01 · startTwo watchersfile_watcher start→02 · existIs it runningfile_watcher exist→03 · showFolder, script, statefile_watcher show→04 · killStop onefile_watcher kill
Script
admin
#1. Two folders, one script;
file mkdir "tmp/tuto_as_invoices";
file mkdir "tmp/tuto_as_credits";
script create exe "tuto.as.on_doc" false 1
	(param
		(var "[kind]" {true} "Event kind" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name" is_null:true is_empty:true "inv_1.pdf")
	;)
	"Register an accounting document"
{
	[file];
} "The file";
file_watcher start "tuto_as_w_invoices" "admin" "tmp/tuto_as_invoices" "tuto.as.on_doc.exe";
file_watcher start "tuto_as_w_credits" "admin" "tmp/tuto_as_credits" "tuto.as.on_doc.exe";
#2. Only the watchers of the project;
-> "[ex]" (concat "exist: " (file_watcher exist "tuto_as_w_invoices") " " (file_watcher exist "tuto_as_w_credits"));
json load "tuto_as_fw" (file_watcher show);
-> "[list]" "";
for (-> "[i]" 0) (< [i] (json count "tuto_as_fw" "/")) (++ "[i]") {
	if (string starts_with (json select "tuto_as_fw" (concat "/[" [i] "]/Key")) "tuto_as_w_") {
		-> "[list]" (concat [list] (json select "tuto_as_fw" (concat "/[" [i] "]/Key")) ": " (json select "tuto_as_fw" (concat "/[" [i] "]/Dir")) " -> " (json select "tuto_as_fw" (concat "/[" [i] "]/Script")) ", " (json select "tuto_as_fw" (concat "/[" [i] "]/State")) [_n_]);
	};
};
#3. Stop the credits watcher;
file_watcher kill "tuto_as_w_credits";
-> "[after]" (concat "after kill: " (file_watcher exist "tuto_as_w_invoices") " " (file_watcher exist "tuto_as_w_credits"));
#4. Cleanup;
file_watcher kill "tuto_as_w_invoices";
json unload "tuto_as_fw";
file delete "tmp/tuto_as_invoices";
file delete "tmp/tuto_as_credits";
script delete "tuto.as.on_doc.exe";
concat [ex] [_n_] [list] [after]
Result
mentdb
exist: 1 1
tuto_as_w_invoices: tmp/tuto_as_invoices -> tuto.as.on_doc.exe, RUNNING...
tuto_as_w_credits: tmp/tuto_as_credits -> tuto.as.on_doc.exe, RUNNING...
after kill: 1 0
Go further
  • Err in file_watcher show gives the last error of a watcher in state ERROR.
  • Start the watchers from a boot script so they come back after a restart.
executed on MentDB

The watcher validates, a heavy loader runs later in the stack

The watcher script must stay light: it only checks that a new file is a CSV and pushes a loader run with a high priority. The loader sums the quantities of the file. The demo follows the file from the event to the loaded total and reads the parent of the loader run.

Pipeline
01 · optionsLoader priorityscript set stack→02 · watchWatch the folderfile_watcher start→03 · pushWatcher pushes the loaderstack→04 · loadRead the filefile load→05 · parentLoader knows the eventstack get
Script
admin
#1. The loader sums the quantities of the file;
file mkdir "tmp/tuto_as_orders";
script create exe "tuto.as.load_orders" false 1
	(param
		(var "[path]" {true} "File to load" is_null:false is_empty:false "tmp/tuto_as_orders/o.csv")
	;)
	"Load an order file"
{
	-> "[txt]" (string trim (file load [path]));
	-> "[total]" 0;
	for (-> "[k]" 2) (<= [k] (atom size [txt] [_n_])) (++ "[k]") {
		-> "[total]" (+ [total] (atom get (atom get [txt] [k] [_n_]) 2 ";"));
	};
	sequence update "tuto_as_orders_qty" [total];
} "1";
script set stack "tuto.as.load_orders.exe" 5 2 true;
#2. The watcher script only validates and queues;
script create exe "tuto.as.on_order" false 1
	(param
		(var "[kind]" {true} "Event kind" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name" is_null:true is_empty:true "o.csv")
	;)
	"Validate a dropped order file and queue its loading"
{
	if (and (equal [kind] "ENTRY_CREATE") (string ends_with [file] ".csv")) {
		stack (date now) "tuto.as.load_orders.exe" "[path]" (concat "tmp/tuto_as_orders/" [file]);
	};
} "1";
sequence add "tuto_as_orders_qty" 0;
#3. Watch, then drop a file;
-> "[t0]" (date now);
file_watcher start "tuto_as_orders" "admin" "tmp/tuto_as_orders" "tuto.as.on_order.exe";
file create "tmp/tuto_as_orders/orders_0425.csv" (concat "sku;qty" [_n_] "A1;12" [_n_] "B7;30" [_n_] "C2;8");
-> "[i]" 0;
while (and (< [i] 75) (== (sequence get current "tuto_as_orders_qty") 0)) {wait "1000"; ++ "[i]";};
file_watcher kill "tuto_as_orders";
#4. The loader run and its parent, the watcher run;
json load "tuto_as_c" (string substring (stack search "closed" "tuto.as.load_orders.exe" "create" [t0] (date dateaddt (date now) "HOUR" 1) ASC 1 5) 39);
json load "tuto_as_p" (stack get (atom get_last (json select "tuto_as_c" "/data/[0]/pid") " "));
json load "tuto_as_v" (stack var_show (json select "tuto_as_p" "/pid"));
json load "tuto_as_par" (stack get (json select "tuto_as_p" "/parent_id"));
-> "[out]" (concat "total quantity loaded: " (sequence get current "tuto_as_orders_qty") ", about " [i] " s after the drop" [_n_] "loader pid " (json select "tuto_as_p" "/pid") ", priority " (json select "tuto_as_p" "/priority") ", " (json select "tuto_as_v" "/[0]/param") "=" (json select "tuto_as_v" "/[0]/value") [_n_] "parent pid " (json select "tuto_as_p" "/parent_id") " = " (json select "tuto_as_par" "/script"));
#5. Cleanup;
wait "500";
stack delete_wait_script "tuto.as.on_order.exe";
stack delete_closed_script "tuto.as.on_order.exe";
stack delete_closed_script "tuto.as.load_orders.exe";
file delete "tmp/tuto_as_orders/orders_0425.csv";
file delete "tmp/tuto_as_orders";
json unload "tuto_as_c";
json unload "tuto_as_p";
json unload "tuto_as_v";
json unload "tuto_as_par";
sequence remove "tuto_as_orders_qty";
script delete "tuto.as.on_order.exe";
script delete "tuto.as.load_orders.exe";
[out]
Result
mentdb
total quantity loaded: 50, about 59 s after the drop
loader pid 3t, priority 5, [path]=tmp/tuto_as_orders/orders_0425.csv
parent pid 3s = tuto.as.on_order.exe
Go further
  • restricted show_pid on the watcher pid gives the logs of the event and of the loader.
  • Push with a future date to load the files only at night.
executed on MentDB

The three events of a file: create, modify, delete

To write a correct watcher script, the developer must know which events arrive. The demo creates, appends to and deletes a file in a watched folder; the script logs each [kind] and [file] it receives.

Pipeline
01 · watchWatch the folderfile_watcher start→02 · createNew filefile create→03 · modifyAppend a linefile append→04 · deleteRemove the filefile delete→05 · readEvents receivedlog search
Script
admin
#1. The script logs what it receives;
file mkdir "tmp/tuto_as_events";
script create exe "tuto.as.on_event" true 1
	(param
		(var "[kind]" {true} "Event kind" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name" is_null:true is_empty:true "a.txt")
	;)
	"Log the event received"
{
	log write (concat [kind] " " [file]) "OK" "kind" [kind];
} "1";
script set stack "tuto.as.on_event.exe" 0 5 false;
#2. Create, append, delete, with a pause between them;
-> "[t0]" (date now);
file_watcher start "tuto_as_events" "admin" "tmp/tuto_as_events" "tuto.as.on_event.exe";
file create "tmp/tuto_as_events/stock.txt" "A1;3";
wait "3000";
file append "tmp/tuto_as_events/stock.txt" (concat [_n_] "B7;5");
wait "3000";
file delete "tmp/tuto_as_events/stock.txt";
wait "3000";
file_watcher kill "tuto_as_events";
#3. Wait for the stack to run all the events;
-> "[i]" 0;
-> "[n]" 0;
-> "[q]" 1;
while (and (< [i] 45) (> [q] 0)) {
	wait "1000";
	++ "[i]";
	json load "tuto_as_q" (string substring (stack search "wait" "tuto.as.on_event.exe" "create" [t0] (date dateaddt (date now) "HOUR" 1) ASC 1 20) 39);
	-> "[q]" (json count "tuto_as_q" "/data");
};
json load "tuto_as_l" (string substring (log search "ok" "tuto.as.on_event.exe" "kind" null null [t0] (date dateaddt (date now) "MIN" 1) ASC 1 20) 39);
-> "[ev]" "";
for (-> "[k]" 0) (< [k] (json count "tuto_as_l" "/data")) (++ "[k]") {
	-> "[ev]" (concat [ev] [_n_] (json select "tuto_as_l" (concat "/data/[" [k] "]/msg")));
};
#4. Cleanup;
-> "[n]" (stack delete_closed_script "tuto.as.on_event.exe");
file delete "tmp/tuto_as_events";
json unload "tuto_as_q";
json unload "tuto_as_l";
script delete "tuto.as.on_event.exe";
concat [n] " runs for 3 file operations:" [ev]
Result
mentdb
3 runs for 3 file operations:
ENTRY_CREATE stock.txt
ENTRY_MODIFY stock.txt
ENTRY_DELETE stock.txt
Go further
  • A large file copied slowly can send several ENTRY_MODIFY: process the file on ENTRY_CREATE only, or test its size.
  • Write the files under a temporary name, then rename them, to send a single complete event.
executed on MentDB

Process a CSV on arrival and archive it

Each stock file dropped in the inbox must be summed into a tuto_ node and then moved to an archive folder. The watcher script reads the file, writes the result node, copies the file to the archive and deletes it from the inbox.

Pipeline
01 · readRead the filefile load→02 · storeResult in a nodenode iobject→03 · archiveCopy to the archivefile copy_file→04 · inboxInbox emptiedfile dir_list→05 · resultRead the nodenode show
Script
admin
#1. Inbox and archive;
file mkdir "tmp/tuto_as_stock_in";
file mkdir "tmp/tuto_as_stock_done";
script create exe "tuto.as.on_stock" false 1
	(param
		(var "[kind]" {true} "Event kind" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name" is_null:true is_empty:true "stock.csv")
	;)
	"Sum a stock file, then archive it"
{
	if (equal [kind] "ENTRY_CREATE") {
		-> "[src]" (concat "tmp/tuto_as_stock_in/" [file]);
		-> "[txt]" (string trim (file load [src]));
		-> "[n]" (- (atom size [txt] [_n_]) 1);
		-> "[qty]" 0;
		for (-> "[k]" 2) (<= [k] (+ [n] 1)) (++ "[k]") {
			-> "[qty]" (+ [qty] (atom get (atom get [txt] [k] [_n_]) 2 ";"));
		};
		node create "tuto_as_stock_result";
		node iobject "tuto_as_stock_result" / "file" [file] STR;
		node iobject "tuto_as_stock_result" / "lines" [n] INT;
		node iobject "tuto_as_stock_result" / "qty" [qty] INT;
		file copy_file [src] (concat "tmp/tuto_as_stock_done/" [file]);
		file delete [src];
	};
} "1";
#2. Watch and drop a stock file;
file_watcher start "tuto_as_stock" "admin" "tmp/tuto_as_stock_in" "tuto.as.on_stock.exe";
file create "tmp/tuto_as_stock_in/stock_week44.csv" (concat "sku;qty" [_n_] "A1;120" [_n_] "B7;45" [_n_] "C2;300" [_n_] "D9;15");
-> "[i]" 0;
while (and (< [i] 45) (== (node exist "tuto_as_stock_result") 0)) {wait "1000"; ++ "[i]";};
wait "1000";
file_watcher kill "tuto_as_stock";
#3. Result, inbox and archive;
-> "[out]" (concat "processed about " [i] " s after the drop" [_n_] "result: " (node show "tuto_as_stock_result") [_n_] "inbox: " (file dir_list "tmp/tuto_as_stock_in") ", archive: " (file dir_list "tmp/tuto_as_stock_done"));
#4. Cleanup;
wait "500";
stack delete_wait_script "tuto.as.on_stock.exe";
stack delete_closed_script "tuto.as.on_stock.exe";
node delete "tuto_as_stock_result";
file delete "tmp/tuto_as_stock_done/stock_week44.csv";
file delete "tmp/tuto_as_stock_done";
file delete "tmp/tuto_as_stock_in";
script delete "tuto.as.on_stock.exe";
[out]
Result
mentdb
processed about 29 s after the drop
result: {
  "file": "stock_week44.csv",
  "qty": 480,
  "lines": 4,
  "key": "tuto_as_stock_result"
}
inbox: [], archive: [
  "stock_week44.csv"
]
Go further
  • Name the archive with the date: concat "tmp/tuto_as_stock_done/" (date sysdate) "_" [file].
  • On a failure, move the file to an error folder in a try block instead of losing it.
executed on MentDB

A bad file ends in the error table of the stack

A watcher script refuses files without the expected header. The demo drops a good and a bad file, then reads the error message of the bad one in the stack and cleans the errors of that script.

Pipeline
01 · checkHeader teststring starts_with→02 · refuseRaise an errorexception→03 · errorsFailed runsstack search→04 · messageWhy it failedstack get→05 · cleanErrors of the scriptstack delete_error_script
Script
admin
#1. The script refuses a file without the header sku and qty;
file mkdir "tmp/tuto_as_check";
script create exe "tuto.as.on_check" false 1
	(param
		(var "[kind]" {true} "Event kind" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name" is_null:true is_empty:true "a.csv")
	;)
	"Check the header of a dropped file"
{
	if (equal [kind] "ENTRY_CREATE") {
		-> "[txt]" (file load (concat "tmp/tuto_as_check/" [file]));
		if (not (string starts_with [txt] "sku;qty")) {exception (1) (concat [file] ": bad header " (atom get [txt] 1 [_n_]));};
		sequence increment "tuto_as_good_files";
	};
} "1";
script set stack "tuto.as.on_check.exe" 0 5 false;
sequence add "tuto_as_good_files" 0;
#2. One good file, one bad file;
-> "[t0]" (date now);
file_watcher start "tuto_as_check" "admin" "tmp/tuto_as_check" "tuto.as.on_check.exe";
file create "tmp/tuto_as_check/ok_0425.csv" (concat "sku;qty" [_n_] "A1;3");
file create "tmp/tuto_as_check/ko_0425.csv" (concat "ref,quantity" [_n_] "A1,3");
wait "4000";
file_watcher kill "tuto_as_check";
#3. Wait for the runs, then read the error;
-> "[i]" 0;
-> "[n]" 0;
while (and (< [i] 45) (< [n] 1)) {
	wait "1000";
	++ "[i]";
	json load "tuto_as_e" (string substring (stack search "error" "tuto.as.on_check.exe" "create" [t0] (date dateaddt (date now) "HOUR" 1) ASC 1 20) 39);
	-> "[n]" (json count "tuto_as_e" "/data");
};
wait "1000";
json load "tuto_as_p" (stack get (atom get_last (json select "tuto_as_e" "/data/[0]/pid") " "));
-> "[out]" (concat "good files: " (sequence get current "tuto_as_good_files") ", runs in error: " [n] [_n_] "error: " (string regexp_extract (json select "tuto_as_p" "/lasterrormsg") "1: (.*)$" 1));
#4. Cleanup;
-> "[out]" (concat [out] [_n_] "errors removed: " (stack delete_error_script "tuto.as.on_check.exe"));
stack delete_wait_script "tuto.as.on_check.exe";
stack delete_closed_script "tuto.as.on_check.exe";
file delete "tmp/tuto_as_check/ok_0425.csv";
file delete "tmp/tuto_as_check/ko_0425.csv";
file delete "tmp/tuto_as_check";
json unload "tuto_as_e";
json unload "tuto_as_p";
sequence remove "tuto_as_good_files";
script delete "tuto.as.on_check.exe";
[out]
Result
mentdb
good files: 1, runs in error: 1
error: ko_0425.csv: bad header ref,quantity
errors removed: 1
Go further
  • Make the script replayable with script set stack to replay a file once the partner fixed it.
  • Write the error with log write and status KO to find it with log search "ko" ....
executed on MentDB

Restart a watcher safely

A deploy script must move a watcher to a new folder without failing when it already runs. The script shows the error of a second start, kills the watcher only if it exists and starts it again on the new folder.

Pipeline
01 · startFirst startfile_watcher start→02 · twiceSecond start refusedtry→03 · existKill only if it existsfile_watcher exist→04 · moveStart on the new folderfile_watcher kill→05 · checkNew folder in the listfile_watcher show
Script
admin
#1. Old and new folder, one script;
file mkdir "tmp/tuto_as_v1";
file mkdir "tmp/tuto_as_v2";
script create exe "tuto.as.on_any" false 1
	(param
		(var "[kind]" {true} "Event kind" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name" is_null:true is_empty:true "a.csv")
	;)
	"React to a file"
{
	[file];
} "The file";
file_watcher start "tuto_as_feed" "admin" "tmp/tuto_as_v1" "tuto.as.on_any.exe";
#2. A second start with the same key is refused;
try {
	file_watcher start "tuto_as_feed" "admin" "tmp/tuto_as_v2" "tuto.as.on_any.exe";
	-> "[twice]" "started twice";
} {
	-> "[twice]" (string regexp_extract [err] "Sorry[^\]]*" 0);
} "[err]";
#3. Move it: kill if it exists, start on the new folder;
if (file_watcher exist "tuto_as_feed") {file_watcher kill "tuto_as_feed";};
file_watcher start "tuto_as_feed" "admin" "tmp/tuto_as_v2" "tuto.as.on_any.exe";
json load "tuto_as_fw" (file_watcher show);
-> "[dir]" "";
for (-> "[i]" 0) (< [i] (json count "tuto_as_fw" "/")) (++ "[i]") {
	if (equal (json select "tuto_as_fw" (concat "/[" [i] "]/Key")) "tuto_as_feed") {-> "[dir]" (concat (json select "tuto_as_fw" (concat "/[" [i] "]/Dir")) " " (json select "tuto_as_fw" (concat "/[" [i] "]/State")));};
};
#4. Cleanup;
file_watcher kill "tuto_as_feed";
json unload "tuto_as_fw";
file delete "tmp/tuto_as_v1";
file delete "tmp/tuto_as_v2";
script delete "tuto.as.on_any.exe";
concat "second start: " [twice] [_n_] "now watching: " [dir] [_n_] "after kill: " (file_watcher exist "tuto_as_feed")
Result
mentdb
second start: Sorry, watch service key 'tuto_as_feed' already exist.
now watching: tmp/tuto_as_v2 RUNNING...
after kill: 0
Go further
  • Put this block in a boot script: watchers live in memory and stop with the server.
  • The watcher runs its script with the user given at start: use a technical user with only the needed groups.
executed on MentDB

Wait for the READY flag before loading the files

A partner drops its data files one by one, then a READY file when the delivery is complete. The watcher queues an ingest for each data file; a delay condition on the ingest postpones it until the READY file exists.

Pipeline
01 · delayCondition on the flag filescript set delay→02 · watchWatch the delivery folderfile_watcher start→03 · queueOne ingest per data filestack→04 · flagREADY arrivesfile create→05 · checkExist the flagfile exist
Script
admin
#1. The ingest waits for tmp/tuto_as_delivery/READY, checked again 10 s later;
file mkdir "tmp/tuto_as_delivery";
script create exe "tuto.as.ingest" false 1
	(param
		(var "[path]" {true} "Data file" is_null:false is_empty:false "tmp/tuto_as_delivery/part1.csv")
	;)
	"Ingest one data file of a complete delivery"
{
	sequence increment "tuto_as_ingested";
} "1";
script set stack "tuto.as.ingest.exe" 0 5 false;
script set delay "tuto.as.ingest.exe" 10 sec {
	file exist "tmp/tuto_as_delivery/READY";
};
#2. The watcher queues an ingest for each new csv;
script create exe "tuto.as.on_delivery" false 1
	(param
		(var "[kind]" {true} "Event kind" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name" is_null:true is_empty:true "part1.csv")
	;)
	"Queue the ingest of each delivered csv"
{
	if (and (equal [kind] "ENTRY_CREATE") (string ends_with [file] ".csv")) {
		stack (date now) "tuto.as.ingest.exe" "[path]" (concat "tmp/tuto_as_delivery/" [file]);
	};
} "1";
sequence add "tuto_as_ingested" 0;
-> "[t0]" (date now);
file_watcher start "tuto_as_delivery" "admin" "tmp/tuto_as_delivery" "tuto.as.on_delivery.exe";
#3. Two parts arrive, no READY yet;
file create "tmp/tuto_as_delivery/part1.csv" "sku;qty";
file create "tmp/tuto_as_delivery/part2.csv" "sku;qty";
-> "[i]" 0;
-> "[n]" 0;
while (and (< [i] 45) (< [n] 2)) {
	wait "1000";
	++ "[i]";
	json load "tuto_as_q" (string substring (stack search "wait" "tuto.as.ingest.exe" "create" [t0] (date dateaddt (date now) "HOUR" 1) ASC 1 20) 39);
	-> "[n]" (json count "tuto_as_q" "/data");
};
-> "[before]" (concat "ingests queued: " [n] ", ingested without READY: " (sequence get current "tuto_as_ingested"));
#4. The READY flag arrives;
file create "tmp/tuto_as_delivery/READY" "2 files";
-> "[j]" 0;
while (and (< [j] 50) (< (sequence get current "tuto_as_ingested") 2)) {wait "1000"; ++ "[j]";};
file_watcher kill "tuto_as_delivery";
-> "[out]" (concat [before] [_n_] "READY exists: " (file exist "tmp/tuto_as_delivery/READY") ", ingested about " [j] " s later: " (sequence get current "tuto_as_ingested"));
#5. Cleanup;
stack delete_wait_script "tuto.as.on_delivery.exe";
stack delete_closed_script "tuto.as.on_delivery.exe";
stack delete_wait_script "tuto.as.ingest.exe";
stack delete_closed_script "tuto.as.ingest.exe";
file delete "tmp/tuto_as_delivery/part1.csv";
file delete "tmp/tuto_as_delivery/part2.csv";
file delete "tmp/tuto_as_delivery/READY";
file delete "tmp/tuto_as_delivery";
json unload "tuto_as_q";
sequence remove "tuto_as_ingested";
script delete "tuto.as.on_delivery.exe";
script delete "tuto.as.ingest.exe";
[out]
Result
mentdb
ingests queued: 2, ingested without READY: 0
READY exists: 1, ingested about 30 s later: 2
Go further
  • script generate delay "tuto.as.ingest.exe" prints the delay to copy into a deploy script.
  • Delete the READY file at the end of the ingest so the next delivery waits again.
executed on MentDB

Route the dropped files by type and content

Partners drop JSON and CSV files in one inbox. The watcher script routes each file by extension, checks that the content really is JSON or CSV, moves it to its outbox or to rejected, and logs the decision.

Pipeline
01 · foldersInbox and outboxesfile mkdir→02 · routeRoute by extension and contenttype is_json→03 · moveCopy to the outbox, delete the originalfile copy_file→04 · watchWatch the inboxfile_watcher start→05 · waitWait for an empty inboxfile dir_list
Script
admin
#1. One inbox, three outboxes;
file mkdir "tmp/tuto_p4m_in";
file mkdir "tmp/tuto_p4m_out";
file mkdir "tmp/tuto_p4m_out/json";
file mkdir "tmp/tuto_p4m_out/csv";
file mkdir "tmp/tuto_p4m_out/rejected";
#2. The router: route by extension, check the content, then move;
script create exe "tuto.p4m.router" false 1
	(param
		(var "[kind]" {true} "Event kind sent by the watcher" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name sent by the watcher" is_null:true is_empty:true "order.json")
	;)
	"Route the files dropped in tmp/tuto_p4m_in"
{
	if (not (equal [kind] "ENTRY_CREATE")) {"ignored"} {
		-> "[src]" (concat "tmp/tuto_p4m_in/" [file]);
		-> "[data]" (file load [src]);
		-> "[box]" (switch (string lower (atom get_last [file] "."))
			("json") {if (type is_json [data]) {"json"} {"rejected"}}
			("csv") {if (string contains [data] ";") {"csv"} {"rejected"}}
			{"rejected"}
		;);
		file copy_file [src] (concat "tmp/tuto_p4m_out/" [box] "/" [file]);
		file delete [src];
		log write (concat [file] " routed to " [box]) "OK" "route" [box];
		[box];
	};
} "The outbox";
#3. Watch, then three partners drop their files;
file_watcher start "tuto_p4m_router" "admin" "tmp/tuto_p4m_in" "tuto.p4m.router.exe";
file create "tmp/tuto_p4m_in/order_77.json" "{\"order\": 77, \"total\": 129.9}";
file create "tmp/tuto_p4m_in/stock.CSV" (concat "sku;qty" [_n_] "A1;12");
file create "tmp/tuto_p4m_in/broken.json" "{\"order\": 78,";
#4. Wait for the inbox to be empty;
-> "[i]" 0;
while (and (< [i] 50) (> (string regexp_count (file dir_list "tmp/tuto_p4m_in") "\"") 0)) {wait 1000; ++ "[i]";};
-> "[out]" (concat "inbox empty after ~" [i] " s" [_n_]);
for (-> "[b]" 1) (<= [b] 3) (++ "[b]") {
	-> "[box]" (atom get "json,csv,rejected" [b] ",");
	-> "[out]" (concat [out] [box] ": " (string regexp_replace (file dir_list (concat "tmp/tuto_p4m_out/" [box])) "\s+" "") [_n_]);
};
#5. Stop and clean up;
file_watcher kill "tuto_p4m_router";
wait 1000;
stack delete_closed_script "tuto.p4m.router.exe";
stack delete_wait_script "tuto.p4m.router.exe";
file delete "tmp/tuto_p4m_out";
file delete "tmp/tuto_p4m_in";
script delete "tuto.p4m.router.exe";
[out]
Result
mentdb
inbox empty after ~25 s
json: ["order_77.json"]
csv: ["stock.CSV"]
rejected: ["broken.json"]
Go further
  • The extension test uses string lower: stock.CSV is routed like stock.csv.
  • Deleting the original sends an ENTRY_DELETE event: the script must ignore every kind except ENTRY_CREATE.
executed on MentDB

Refuse a file received twice under two names

A partner sometimes sends the same invoice again under a new name. The watcher script hashes the content, remembers each hash in a node and deletes the copy, whichever of the two the stack runs first. The check and the insert are under a synchronized lock because the stack runs the files in parallel.

Pipeline
01 · memoryA node of known hashesnode create→02 · hashHash the contentstring md5→03 · lockCheck and insert under a locksynchronized→04 · watchWatch and drop three filesfile_watcher start→05 · proofThe decisions in the loglog search
Script
admin
#1. The inbox and a node that remembers the content hash of every file received;
file mkdir "tmp/tuto_p4m_drop";
node create "tuto_p4m_hashes";
sequence add "tuto_p4m_seen" 0;
#2. On arrival, with the activity log on: hash the content, refuse a file already received under another name;
script create exe "tuto.p4m.dedupe" true 1
	(param
		(var "[kind]" {true} "Event kind sent by the watcher" is_null:true is_empty:true "ENTRY_CREATE")
		(var "[file]" {true} "File name sent by the watcher" is_null:true is_empty:true "invoice_A.xml")
	;)
	"Skip the files already received"
{
	if (equal [kind] "ENTRY_CREATE") {
		-> "[path]" (concat "tmp/tuto_p4m_drop/" [file]);
		-> "[md5]" (string md5 (file load [path]));
		#The stack runs the files in parallel: check and insert under a lock;
		synchronized ("tuto_p4m_hashes") {
			-> "[first]" (node select "tuto_p4m_hashes" (concat "/h" [md5]));
			if (is null [first]) {
				node iobject "tuto_p4m_hashes" / (concat "h" [md5]) [file] STR;
			};
		};
		if (is null [first]) {
			log write (concat [file] " accepted") "OK" "dedupe" "new";
		} {
			log write (concat [file] " is a copy of " [first] ", deleted") "OK" "dedupe" "duplicate";
			file delete [path];
		};
		sequence increment "tuto_p4m_seen";
	};
} "1";
#3. Watch, then the partner sends the same invoice twice under two names;
-> "[t0]" (date now);
file_watcher start "tuto_p4m_dedupe" "admin" "tmp/tuto_p4m_drop" "tuto.p4m.dedupe.exe";
file create "tmp/tuto_p4m_drop/invoice_A.xml" "<invoice id=\"F-2026-118\" total=\"540.00\"/>";
file create "tmp/tuto_p4m_drop/invoice_B.xml" "<invoice id=\"F-2026-119\" total=\"75.50\"/>";
file create "tmp/tuto_p4m_drop/invoice_A_resent.xml" "<invoice id=\"F-2026-118\" total=\"540.00\"/>";
-> "[i]" 0;
while (and (< [i] 50) (< (sequence get current "tuto_p4m_seen") 3)) {wait 1000; ++ "[i]";};
#4. The decisions in the log, and what is left in the inbox;
json load "rx_p4m_l" (string substring (log search "ok" "tuto.p4m.dedupe.exe" "dedupe" null null [t0] (date dateaddt (date now) "MIN" 1) ASC 1 10) 39);
-> "[out]" (concat "3 files handled after ~" [i] " s" [_n_]);
json parse_array "rx_p4m_l" "/data" "[row]" {
	-> "[out]" (concat [out] (json load select "rx_p4m_r" [row] /c_val) ": " (json load select "rx_p4m_r" [row] /msg) [_n_]);
};
-> "[out]" (concat [out] "inbox: " (string regexp_replace (file dir_list "tmp/tuto_p4m_drop") "\s+" "") ", known hashes: " (- (node count "tuto_p4m_hashes" /) 1));
#5. Clean up;
file_watcher kill "tuto_p4m_dedupe";
wait 1000;
stack delete_closed_script "tuto.p4m.dedupe.exe";
stack delete_wait_script "tuto.p4m.dedupe.exe";
file delete "tmp/tuto_p4m_drop";
node delete "tuto_p4m_hashes";
sequence remove "tuto_p4m_seen";
json unload "rx_p4m_r";
json unload "rx_p4m_l";
script delete "tuto.p4m.dedupe.exe";
[out]
Result
mentdb
3 files handled after ~17 s
new: invoice_B.xml accepted
new: invoice_A_resent.xml accepted
duplicate: invoice_A.xml is a copy of invoice_A_resent.xml, deleted
inbox: ["invoice_A_resent.xml","invoice_B.xml"], known hashes: 2
Go further
  • Without synchronized, the three runs read the node at the same time and all of them accept the file: try it.
  • log write only writes when the activity log of the script is on (second parameter of script create).

Also in Programming

MB
MQL beginner
Variables, operators, if, loops and your first try/exception: learn MQL step by step.
10 recipes
MD
MQL developer
Atom lists, data types, constants, translations, session variables and error patterns.
10 recipes
BW
Backend & web service developer
Typed scripts, execute / call / include, generated calls and URLs, parameters and logs.
10 recipes
DT
Date & time developer
Calendars, business days, time zones, ISO and epoch, durations, quarters and week numbers.
10 recipes
SE
String & encoding developer
Cleaning, case styles, regex, URL/HTML/Base64 encodings, number bases and encryption.
10 recipes
DV
Data validation developer
IBAN, BIC, SIRET, Luhn, email, IP, UUID, JSON and date checks gathered in a validation report.
10 recipes
JD
JSON & document store developer
In-memory JSON documents, persistent nodes and sequences: build, walk, sort, clean up.
10 recipes
SD
Script developer
Full script lifecycle in the MQL editor: typed parameters, read back, generate, patch, deploy, copy, grant, delete.
10 recipes
EM
Execution modes
execute, call, include, sub_include, parallel, stack: which environment, which checks, which logs.
11 recipes
WS
Web service developer
A small REST resource with get, post, put and delete scripts: validation, JSON answers, codes, grants, URLs and API docs.
10 recipes
AD
Async developer
Asynchronous processing with the stack: push runs now or later, priorities and parallel limits, errors and replays, flows and cleanup.
10 recipes
SD
Scheduler developer
Cron jobs and delayed runs: add, resume, pause, update and remove jobs, prove they ran, and postpone stack runs with a delay condition.
10 recipes
CC
Configuration & connections
Parameters (private, described, locked, deployed), environment variables, sessions and connections used by your scripts.
12 recipes
VD
Versioning & deployment
Group your scripts, commit, compare, roll back, export, import and package a release.
10 recipes
DE
Debugging & error handling
Coded errors with try/exception, logs with correlation ids, timing, metrics and defensive checks.
10 recipes
SD
SQL developer in MQL
Connections, DDL and DML with sql encode, transactions, cursors, parse, and exports to JSON, CSV, Excel and PDF on a real local database. Run the recipes in order: the first one creates the tuto_db_h2 database.
10 recipes
CD
CRUD & data API developer
Generate SCRUD scripts from a table, patch and install them, write typed read and write APIs, check data quality and manage connections per environment. Uses the tuto_db_h2 database of the SQL developer recipes; the last recipe removes it.
10 recipes
WA
Web application developer
Create a web app, its host name, pages, ajax forms, menu and grants, then remove it all.
10 recipes
AI
Algorithms in MQL
Sorting, searching, recursion, data structures and number algorithms: MQL is a real programming language.
10 recipes
CF
Control flow
if, if force, case, switch, mode, for, while, repeat, break, continue, parallel and synchronized on real cases.
10 recipes
TM
The MQL language in depth
Data types and null, booleans, comments, variables and scope, eval and mql blocks, coded errors and try.
10 recipes
SU
Screen & UI developer (sajax)
Build the screens of MentDB web apps in MQL: Bootstrap grids, forms and every control, ajax buttons, alerts, refreshes and charts.
10 recipes