diff --git a/net_services/CXM_generic_protocol.anubis b/net_services/CXM_generic_protocol.anubis index 37fb8b5..a9bf32b 100644 --- a/net_services/CXM_generic_protocol.anubis +++ b/net_services/CXM_generic_protocol.anubis @@ -245,50 +245,3 @@ public define Bool } . -public define Bool - send_file_ref - ( - MessageQueue mQ, - Message msg, - )= - //println("NET_SERVICE send_file_ref received *"); - if *msg.what = _CXM_GET_FILE_REF then - if find_message(msg, "GET_FILE_REF") is { - failure then println("NET_SERVICE send_file_ref: can't find GET_FILE_REF message"); false, - success(file_ref_obj_msg) then - if (Maybe(File_ref))from_Message(file_ref_obj_msg) is - { - failure then println("NET_SERVICE send_file_ref: can't extract GET_FILE_REF message");false, - success(f_ref) then - //println("NET_SERVICE send_file_ref: GET_FILE_REF message File_ref extracted\n"+dump(file_ref_obj_msg)); - //create the target - with source_file = f_ref.path + "/" + f_ref.name, - with msg = message(0), - with size = file_size(source_file), - - println("source_file = "+source_file); - println("size = "+size); - forget(add_string(msg, "MODE", "NEW")); - forget(add_string(msg, "SIZE", to_String(size))); - - //println("NET_SERVICE send_file_ref: reply message \n"+dump(msg)); - if (Maybe(RStream))file(source_file, read) is - { - failure then println("NET_SERVICE send_file_ref: can't open source file"+source_file);false, //nothing to write - success(source) then - //println("NET_SERVICE send_file_ref send ack file size "+size); - send_ACK_ok(mQ, _CXM_GET_FILE_REF, msg); - //println("call flush_wait"); - mQ.flush_wait(unique); - if copy_file_to_Connection(file(source), mQ.get_connection(unique), size) is - { - failure then println("NET_SERVICE send_file_ref copy_file_to_Connection error");false, - success(_) then println("NET_SERVICE send_file_ref copy_file_to_Connection ok");true - } - } - } - } - else //not _CXM_GET_FILE_REF - println("[send_file_ref] it's not a _CXM_GET_FILE_REF"); - false -. diff --git a/net_services/get_file.anubis b/net_services/get_file.anubis new file mode 100644 index 0000000..0e2ff80 --- /dev/null +++ b/net_services/get_file.anubis @@ -0,0 +1,170 @@ +/* + * Created by PyramIDE. + * User: フランスのトトロ aka (David RENÉ) + * Date: 26/05/2019 + * Time: 15:23 + * © David RENÉ + */ + + +read calexium_lib/CXM_message_constants.anubis +read calexium_lib/net_services/CXM_generic_protocol.anubis +read calexium_lib/net_services/CXM_generic_client.anubis //for get_ip +read system/message_queue.anubis +read system/message_transceiver.anubis +read system/muscle.anubis +read system/files.anubis +read tools/basis.anubis +read network/dns.anubis + +define Maybe(One) + receive_data + ( + MessageQueue mQ, + WStream fd, + Int so_far, + Int left_read + )= + //println("receive_data "); + if mQ.get_next_received_Message(30) is + { + timeout then failure, + closed then failure, + msg(_msg) then + with last_block = if find_bool(_msg, "End") is {failure then false, success(r) then r}, + if find_raw(_msg, "Data") is + { + failure then println("Can't find raw Data"); send_ACK_error(mQ, _CXM_FTP_DATA); failure, + success(data) then + if write(fd, data) is + { + failure then println("Can't write into file"); send_ACK_error(mQ, _CXM_FTP_DATA);failure, + success(len) then + send_ACK_ok(mQ, _CXM_FTP_DATA); + if last_block then + println("File received successfully"); + //TODO this is a big hack, we must check if all data are sent from the mQ + sleep(10000); + success(unique) + else +// println("Bytes received : " + (so_far + len)); + receive_data(mQ, fd, so_far + len, left_read - len) + } + } + } +. + +define Maybe(One) + start_get_file_transtert + ( + MessageQueue mQ, + String local_file, + Int size + )= + make_directories(local_file); + if file(local_file, new) is + { + failure then println("Can't create \""+local_file+"\" file"); failure, + success(fd) then //the local file is open + with start = message(_CXM_FTP_START_TRANSFERT), + mQ.add_Message_to_send(start); + receive_data(mQ, weaken(fd), 0, size) + } + . + +define Bool + get_file_ref + ( + MessageQueue mQ, + File_ref f_ref, + String tmp_path, + (LogLevel, String) -> One logger + )= + with target_file = tmp_path + "/" + f_ref.name, + with get_file_msg = message(_CXM_GET_FILE_REF), + forget(add_message(get_file_msg, "GET_FILE_REF", to_Message(f_ref))); + + mQ.add_Message_to_send(get_file_msg); + if wait_for_reply(mQ, _CXM_FTP_GET_FILE, 30) is + { + failure then println("get_file failure");failure, + timeout then println("get_file timeout");failure, + unknow_cmd then println("get_file unknow_cmd");failure, + error then println("get_file remote error msg");failure, + error(c,m) then println("get_file remote error [" + c + "] msg '" + m + "'");failure, + ok then println("get_file ok");failure, + ok_msg(msg)then + with size = if find_string(rmsg, "SIZE") is {failure then f_ref.size, success(size_str) then if decimal_scan(size_str) is { failure then should_not_happen(0), success(_size_) then _size_}}, + start_get_file_transtert(mQ, target_file, size) + } +. + + +define Bool + request_for_service + ( + MessageQueue queue + )= + with test_msg = message(_CXM_REQUEST_FOR_SERVICE), + forget(add_int32(test_msg, "service", _CXM_FTP_SERVICE_ID)); + forget(add_int32(test_msg, "version", 1)); + queue.add_Message_to_send(test_msg); + if queue.get_next_received_Message(30) is + { + timeout then println("request_for_service receive timeout");false, + closed then println("request_for_service socket closed");false, + msg(msg) then + if find_int32(msg, "STATUS") is + { + failure then println("status not found");false, + success(v) then + if v = _CXM_OK then + true + else + false + } + }. + +public define Maybe(One) + ftp_get_file + ( + String server, + Word32 ip_port, + String remote_file, + String local_file, + Bool ftp, + ) = + if mb_get_ip(server) is + { + failure then println("server "+server+" DNS error");failure, + success(ip_adr) then + if connect( ip_adr, ip_port) is + { + error(_) then println("can't connect to ftp server"); failure, + ok(conn) then + with queue = create_MessageQueue("ftp_get_file", tcp(conn)), + message_transceiver(/*conn,*/ queue); + println("request for ftp service"); + if request_for_service(queue) then + with result = get_file(queue, remote_file, local_file, ftp), + queue.quit(unique);result + else + queue.quit(unique); + println("Service not found");failure + } + }. + + /* Exists for backward compatibility. + * The last argument set to false (no ftp directory on the remote, pickup on collection dir) + */ + +public define Maybe(One) + ftp_get_file + ( + String server, + Word32 ip_port, + String remote_file, + String local_file, + ) = + ftp_get_file(server, ip_port, remote_file, local_file, false). + diff --git a/net_services/send_file.anubis b/net_services/send_file.anubis new file mode 100644 index 0000000..fd27c64 --- /dev/null +++ b/net_services/send_file.anubis @@ -0,0 +1,235 @@ +/* + * Created by PyramIDE. + * User: フランスのトトロ aka (David RENÉ) + * Date: 26/05/2019 + * Time: 15:21 + * © David RENÉ + */ + + +read tools/basis.anubis +read tools/findstring.anubis +read system/muscle.anubis +read system/data_io.anubis +read system/convert.anubis +read system/string.anubis +read tools/basis.anubis +read system/message_queue.anubis +read system/files.anubis +read system/logger.anubis + +read calexium_lib/net_services_protocols/logger_service.anubis +read calexium_lib/CXM_message_constants.anubis +read calexium_lib/net_services/CXM_net_services.anubis +read calexium_lib/net_services/CXM_generic_protocol.anubis +read calexium_lib/types/generated/file_ref.anubis + read app_constants.anubis + +read types/app_types.anubis + read tools/app_loggers.anubis + +//TODO add return error handling in every functions + +public define Word32 updater_srv_version = 1. + +define Maybe(One) + send_file_state + ( + MessageQueue mQ, + RStream src_file, + Int left_read, + (LogLevel, String) -> One logger + )= + with read_size = min(65536, left_read), + if read(src_file, read_size, 10) is + { + error then logger(logError, "send_file_state read error");failure, + timeout then logger(logError, "send_file_state timeout");failure, + ok(buffer)then + with data_msg = message(_CXM_FTP_DATA), + if add_raw(data_msg, "Data", buffer) is + { + failure then logger(logError, "send_file_state add_raw failure ");failure, + success(_) then + with end_value = if left_read - read_size = 0 then + true + else + false, + + if add_bool(data_msg, "End", end_value) is + { + failure then failure, + success(_) then + //logDebug(main_log, "send_file_state send message"); + mQ.add_Message_to_send(data_msg); + if mQ.get_next_received_Message(30) is + { + timeout then logger(logError, "send_file_state timeout"); failure, + closed then logger(logWarning, "send_file_state closed"); success(unique), + msg(_msg) then + if *_msg.what = _CXM_ACK then + if find_int32(_msg, "CMD") is + { + failure then logger(logError, "send_file_state CMD not found");failure, + success(cmd) then + if find_int32(_msg, "STATUS") is + { + failure then logger(logError, "send_file_state STATUS not found");failure, + success(status) then + if cmd = _CXM_FTP_DATA & status = _CXM_OK then + if end_value then //it was the last block, then quit here + logger(logDebug, "FTP Server: File sent successfully"); + success(unique) + else + send_file_state(mQ, src_file, left_read - read_size, logger) + else + logger(logError, "send_file_state wrong command or status");failure + } + } + else + logger(logError, "send_file_state Not ACK message");failure + } + } + } + }. + +define String + extract_hash + ( + String full_path + ) = + with path = (List(String))extract_dir(full_path), + get_last = (List(String) dirs, String last) |-get_last-> + if dirs is + { + [] then last, + [h . t] then + if length(h) > 0 then get_last(t, h) + else get_last(t, last) + }, + get_last(path, ""). + +define Bool + send_file_start_transfert + ( + MessageQueue queue, + String file_name, + (LogLevel, String) -> One logger + )= + if queue.get_next_received_Message(30) is + { + timeout then logger(logWarning, "send_file_start_transfert timeout");false, + closed then logger(logWarning, "send_file_start_transfert connection closed");false, + msg(_msg) then + //check if the message is start the transfer + if *_msg.what = _CXM_FTP_START_TRANSFERT then + logger(logDebug, "_CXM_FTP_START_TRANSFERT received"); + with start_offset = if find_int32(_msg, "ResumeOffset") is + { + failure then 0, + success(value) then to_Int(value) + }, + if file(file_name, read) is + { + failure then logger(logError, "send_file_start_transfert can't open file '" + file_name + "'");false, + success(source) then + if send_file_state(queue, source, file_size(file_name) - start_offset, logger) is + { + failure then logger(logError, "send_file_start_transfert send_file_state failed"); false, + success(_) then logger(logInfo, "send_file_start_transfert file sent successfully"); true + } + } + else + false + }. + +public define Bool + send_file_ref + ( + MessageQueue queue, + Message msg, + (LogLevel, String) -> One logger + )= + if *msg.what = _CXM_GET_FILE_REF then + if find_message(msg, "GET_FILE_REF") is { + failure then logger(logError, "NET_SERVICE send_file_ref: can't find GET_FILE_REF message"); false, + success(file_ref_obj_msg) then + if (Maybe(File_ref))from_Message(file_ref_obj_msg) is + { + failure then logger(logError, "NET_SERVICE send_file_ref: can't extract GET_FILE_REF message");false, + success(f_ref) then + with source_file = f_ref.path + "/" + f_ref.name, + if file_exists(source_file) then + with result_msg = message(_CXM_ACK_RESULT_MSG), + with size = file_size(source_file), + forget(add_string(result_msg, "SIZE", to_String(size))); + + //forget(add_int32(result_msg, "FileSize", truncate_to_Word32(file_size(file_name)))); + logger(logDebug, "send _CXM_FTP_GET_FILE with filesize "+file_size(source_file)); + send_result(queue, _CXM_FTP_GET_FILE, success(result_msg)); + send_file_start_transfert(queue, source_file, logger) + else + logger(logError, "File '"+source_file+"' not found"); + send_ACK_error(queue, _CXM_FTP_GET_FILE); + false + } + } + else + false +. + + + + public define Bool + send_file_ref + ( + MessageQueue mQ, + Message msg, + (LogLevel, String) -> One logger + )= + //println("NET_SERVICE send_file_ref received *"); + if *msg.what = _CXM_GET_FILE_REF then + if find_message(msg, "GET_FILE_REF") is { + failure then println("NET_SERVICE send_file_ref: can't find GET_FILE_REF message"); false, + success(file_ref_obj_msg) then + if (Maybe(File_ref))from_Message(file_ref_obj_msg) is + { + failure then println("NET_SERVICE send_file_ref: can't extract GET_FILE_REF message");false, + success(f_ref) then + //println("NET_SERVICE send_file_ref: GET_FILE_REF message File_ref extracted\n"+dump(file_ref_obj_msg)); + //create the target + with source_file = f_ref.path + "/" + f_ref.name, + with msg = message(0), + with size = file_size(source_file), + + println("source_file = "+source_file); + println("size = "+size); + forget(add_string(msg, "MODE", "NEW")); + forget(add_string(msg, "SIZE", to_String(size))); + + //println("NET_SERVICE send_file_ref: reply message \n"+dump(msg)); + if (Maybe(RStream))file(source_file, read) is + { + failure then println("NET_SERVICE send_file_ref: can't open source file"+source_file);false, //nothing to write + success(source) then + + //println("NET_SERVICE send_file_ref send ack file size "+size); + send_ACK_ok(mQ, _CXM_GET_FILE_REF, msg); + //println("call flush_wait"); + mQ.flush_wait(unique); + if seek(source, 0) then + if copy_file_to_Connection(file(source), mQ.get_connection(unique), size) is + { + failure then println("NET_SERVICE send_file_ref copy_file_to_Connection error");false, + success(_) then println("NET_SERVICE send_file_ref copy_file_to_Connection ok");true + } + else + println("NET_SERVICE send_file_ref seek error");false, + + } + } + } + else //not _CXM_GET_FILE_REF + println("[send_file_ref] it's not a _CXM_GET_FILE_REF"); + false +. diff --git a/net_services_protocols/ftp_client.anubis b/net_services_protocols/ftp_client.anubis index 69aa469..bb4422d 100644 --- a/net_services_protocols/ftp_client.anubis +++ b/net_services_protocols/ftp_client.anubis @@ -53,7 +53,7 @@ define Maybe(One) } } } - . +. define Maybe(One) start_get_file_transtert -- libgit2 0.21.4