if (not (group exist "0001_folder1_folder2_folder3")) { group add "0001_folder1_folder2_folder3"; }; script merge "folder1.folder2.folder3.step_1_source_connect.exe" true 5 (param ) "Connect to the source" { # CONFIGURATION ; -> "[CONF_NAME_OF_THE_FLOW]" "name_of_the_flow"; # INITIALIZATION ; -> "[FLOW_PID]" [PID]; # HANDLE ; try { # Flow initialization; stack flow_init [FLOW_PID] [CONF_NAME_OF_THE_FLOW] "{}"; stack flow_step [FLOW_PID] 1 "source_connect..."; # Local connection: don't use step 1; log write "Source: Local connection" OK null null; #Step 1 is valid; stack flow_step [FLOW_PID] 1 "source_connect_ok"; include "folder1.folder2.folder3.step_2_extract.exe"; } { #Step 1 is not valid; stack flow_step [FLOW_PID] 1 "source_connect_KO"; # Give the error to the stack and stop the process; exception (1) ([global_err]); } "[global_err]"; } "Return nothing"; if (not (group is granted script "folder1.folder2.folder3.step_1_source_connect.exe" "0001_folder1_folder2_folder3")) { group grant script "folder1.folder2.folder3.step_1_source_connect.exe" "0001_folder1_folder2_folder3"; }; script merge "folder1.folder2.folder3.step_2_extract.exe" true 1 (param ) "Extract files from the source" { # CONFIGURATION ; -> "[CONF_SOURCE_CM]" "demo_cm"; -> "[CONF_DIR_SOURCE]" "tmp/source"; -> "[CONF_SOURCE_FILTER]" ".txt"; # HANDLE ; try { # Mark the flow as step 2; stack flow_step [FLOW_PID] 2 "extract..."; # Create the local directory PID; file mkdir (concat "home/" [FLOW_PID]); # Get files into PID directory; log write (concat "Get files from '" [CONF_DIR_SOURCE]"'.") OK null null; json load "valid_files" "[]"; json load "files_sources" (cifs ls [CONF_DIR_SOURCE] {cm get [CONF_SOURCE_CM];}); json parse_array "files_sources" "/data" "row" { if (string ends_with (json select "row" "/[0]") [CONF_SOURCE_FILTER]) { json iarray "valid_files" / (json select "row" "/[0]") STR; }; }; json parse_array "valid_files" "/" "[filename]" { cifs get (concat [CONF_DIR_SOURCE] "/" [filename]) (concat "home/" [FLOW_PID] "/" [filename]) {cm get [CONF_SOURCE_CM];}; log write (concat "Get file 'home/" [FLOW_PID] "/" [filename] "'.") OK null null; }; # Delete source file; json parse_array "valid_files" "/" "[filename]" { cifs rm (concat [CONF_DIR_SOURCE] "/" [filename]) {cm get [CONF_SOURCE_CM];}; log write (concat "Remove source file '" [CONF_DIR_SOURCE] "/" [filename] "'.") OK null null; }; } { #Step 2 is not valid; stack flow_step [FLOW_PID] 2 "extract_KO"; # Give the error to the stack and stop the process; exception (1) ([global_err]); } "[global_err]"; log write (concat (json count "valid_files" /) " file(s) to transform.") OK null null; if (> (json count "valid_files" /) 0) { #Step 2 is valid; stack flow_step [FLOW_PID] 2 (concat "extract_ok_" (json count "valid_files" /) "_files"); json parse_array "valid_files" "/" "[filename]" { stack (date now) "folder1.folder2.folder3.step_3_transform.exe" "[FLOW_PID]" [FLOW_PID] "[filename]" [filename]; }; } { #Step 2 is valid; stack flow_step [FLOW_PID] 2 "extract_ok_zero_file"; # No file into the directory; file delete (concat "home/" [FLOW_PID]); }; } "Return nothing"; if (not (group is granted script "folder1.folder2.folder3.step_2_extract.exe" "0001_folder1_folder2_folder3")) { group grant script "folder1.folder2.folder3.step_2_extract.exe" "0001_folder1_folder2_folder3"; }; script merge "folder1.folder2.folder3.step_3_transform.exe" true 1 (param (var "[FLOW_PID]" {true} "The flow PID" is_null:true is_empty:true "1") (var "[filename]" {true} "The filename" is_null:true is_empty:true "file.xxx") ) "Transform the file" { # CONFIGURATION ; -> "[CONF_DESTINATION_NEW_FILENAME]" (concat "new_filename_" (date systimestamp_min) ".txt"); -> "[CONF_DESTINATION_TXT_SEPARATOR]" "|"; # HANDLE ; try { # Mark the flow as step 3; stack flow_step [FLOW_PID] 3 "transform..."; file writer_open "w1" (concat "home/" [FLOW_PID] "/" [CONF_DESTINATION_NEW_FILENAME]) true TEXT "utf-8"; file writer_add_line "w1" (concat "HEADER"); file writer_flush "w1"; log write (concat "Build the file 'home/" [FLOW_PID] "/" [CONF_DESTINATION_NEW_FILENAME] "' from 'home/" [FLOW_PID] "/" [filename] "'...") OK null null; file reader_open "r1_source" (concat "home/" [FLOW_PID] "/" [filename]) TEXT "utf-8"; #Parse the file; while (is not null (-> "[line_source]" (file reader_get_line "r1_source"));) { -> "[T_A]" (atom get [line_source] 1 "|"); -> "[T_B]" (atom get [line_source] 2 "|"); -> "[T_C]" (atom get [line_source] 3 "|"); -> "[line]" ""; concat_var "[line]" [_n_]; concat_var "[line]" (string lpad [T_A] "0" 10) [CONF_DESTINATION_TXT_SEPARATOR]; concat_var "[line]" (string lpad [T_B] "0" 10) [CONF_DESTINATION_TXT_SEPARATOR]; concat_var "[line]" (string lpad [T_C] "0" 10); file writer_add_line "w1" [line]; file writer_flush "w1"; }; #Close the reader; file reader_close "r1_source"; file writer_close "w1"; log write (concat "Builded.") OK null null; #Step 3 is valid; stack flow_step [FLOW_PID] 3 "transform_ok"; stack (date now) "folder1.folder2.folder3.step_4_destination_connect.exe" "[FLOW_PID]" [FLOW_PID] "[filename]" [CONF_DESTINATION_NEW_FILENAME]; } { try {file writer_close "w1";} {} "[err]"; try {file reader_close "r1_source";} {} "[sub_err]"; #Step 3 is not valid; stack flow_step [FLOW_PID] 3 "transform_KO"; # Give the error to the stack and stop the process; exception (1) ([global_err]); } "[global_err]"; } "Return nothing"; if (not (group is granted script "folder1.folder2.folder3.step_3_transform.exe" "0001_folder1_folder2_folder3")) { group grant script "folder1.folder2.folder3.step_3_transform.exe" "0001_folder1_folder2_folder3"; }; script merge "folder1.folder2.folder3.step_4_destination_connect.exe" true 5 (param (var "[FLOW_PID]" {true} "The flow PID" is_null:true is_empty:true "1") (var "[filename]" {true} "The filename" is_null:true is_empty:true "file.xxx") ) "Connect to the destination" { # CONFIGURATION ; -> "[CONF_DESTINATION_CM]" "demo_cm"; # HANDLE ; try { # Flow initialization; stack flow_step [FLOW_PID] 4 "destination_connect..."; ftp connect "destination_sess" {cm get [CONF_DESTINATION_CM];}; ftp active "destination_sess"; ftp passive "destination_sess"; ftp active compression "destination_sess" true; ftp set type "destination_sess" "BINARY"; #Step 4 is valid; stack flow_step [FLOW_PID] 4 "destination_connect_ok"; include "folder1.folder2.folder3.step_5_load.exe"; } { try {ftp disconnect "destination_sess";} {} "[err]"; #Step 4 is not valid; stack flow_step [FLOW_PID] 4 "destination_connect_KO"; # Give the error to the stack and stop the process; exception (1) ([global_err]); } "[global_err]"; } "Return nothing"; if (not (group is granted script "folder1.folder2.folder3.step_4_destination_connect.exe" "0001_folder1_folder2_folder3")) { group grant script "folder1.folder2.folder3.step_4_destination_connect.exe" "0001_folder1_folder2_folder3"; }; script merge "folder1.folder2.folder3.step_5_load.exe" true 1 (param ) "Load file to the destination" { # CONFIGURATION ; -> "[CONF_DIR_DESTINATION]" "/remote/dir"; # HANDLE ; try { ftp cd "destination_sess" [CONF_DIR_DESTINATION]; # Mark the flow as step 5; stack flow_step [FLOW_PID] 5 "load..."; log write (concat "Send '" (concat "home/" [FLOW_PID] "/" [filename]) "' to '" (concat [CONF_DIR_DESTINATION] "/" [filename]) "' ...") OK null null; # Send file into the destination directory; ftp put "destination_sess" (concat "home/" [FLOW_PID] "/" [filename]) "RESUME"; log write "Sended." OK null null; ftp disconnect "destination_sess"; #Step 5 is valid; stack flow_step [FLOW_PID] 5 "load_ok"; } { try {ftp disconnect "destination_sess";} {} "[err]"; #Step 5 is not valid; stack flow_step [FLOW_PID] 5 "load_KO"; # Give the error to the stack and stop the process; exception (1) ([global_err]); } "[global_err]"; } "Return nothing"; if (not (group is granted script "folder1.folder2.folder3.step_5_load.exe" "0001_folder1_folder2_folder3")) { group grant script "folder1.folder2.folder3.step_5_load.exe" "0001_folder1_folder2_folder3"; };