Commit dd87d694cfaf1ac9d32c085cc36f8fa79e425eca
1 parent
0afc61c9
remove the send_file_ref and get_file_ref from generic protocol to dispatch in g…
…et_file.anubis and send_file.anubis. This is due to fallback on FTP_** protocol used long time ago to replace the version with raw mode on message queue. There is a bug on that version which I can't find. I decide to fallback to a working version by taking the FTP* protocol used in package manager with MailFountain. This version is modified version to take in account the type File_ref
Showing
4 changed files
with
406 additions
and
48 deletions
Show diff stats
net_services/CXM_generic_protocol.anubis
| ... | ... | @@ -245,50 +245,3 @@ public define Bool |
| 245 | 245 | } |
| 246 | 246 | . |
| 247 | 247 | |
| 248 | -public define Bool | |
| 249 | - send_file_ref | |
| 250 | - ( | |
| 251 | - MessageQueue mQ, | |
| 252 | - Message msg, | |
| 253 | - )= | |
| 254 | - //println("NET_SERVICE send_file_ref received *"); | |
| 255 | - if *msg.what = _CXM_GET_FILE_REF then | |
| 256 | - if find_message(msg, "GET_FILE_REF") is { | |
| 257 | - failure then println("NET_SERVICE send_file_ref: can't find GET_FILE_REF message"); false, | |
| 258 | - success(file_ref_obj_msg) then | |
| 259 | - if (Maybe(File_ref))from_Message(file_ref_obj_msg) is | |
| 260 | - { | |
| 261 | - failure then println("NET_SERVICE send_file_ref: can't extract GET_FILE_REF message");false, | |
| 262 | - success(f_ref) then | |
| 263 | - //println("NET_SERVICE send_file_ref: GET_FILE_REF message File_ref extracted\n"+dump(file_ref_obj_msg)); | |
| 264 | - //create the target | |
| 265 | - with source_file = f_ref.path + "/" + f_ref.name, | |
| 266 | - with msg = message(0), | |
| 267 | - with size = file_size(source_file), | |
| 268 | - | |
| 269 | - println("source_file = "+source_file); | |
| 270 | - println("size = "+size); | |
| 271 | - forget(add_string(msg, "MODE", "NEW")); | |
| 272 | - forget(add_string(msg, "SIZE", to_String(size))); | |
| 273 | - | |
| 274 | - //println("NET_SERVICE send_file_ref: reply message \n"+dump(msg)); | |
| 275 | - if (Maybe(RStream))file(source_file, read) is | |
| 276 | - { | |
| 277 | - failure then println("NET_SERVICE send_file_ref: can't open source file"+source_file);false, //nothing to write | |
| 278 | - success(source) then | |
| 279 | - //println("NET_SERVICE send_file_ref send ack file size "+size); | |
| 280 | - send_ACK_ok(mQ, _CXM_GET_FILE_REF, msg); | |
| 281 | - //println("call flush_wait"); | |
| 282 | - mQ.flush_wait(unique); | |
| 283 | - if copy_file_to_Connection(file(source), mQ.get_connection(unique), size) is | |
| 284 | - { | |
| 285 | - failure then println("NET_SERVICE send_file_ref copy_file_to_Connection error");false, | |
| 286 | - success(_) then println("NET_SERVICE send_file_ref copy_file_to_Connection ok");true | |
| 287 | - } | |
| 288 | - } | |
| 289 | - } | |
| 290 | - } | |
| 291 | - else //not _CXM_GET_FILE_REF | |
| 292 | - println("[send_file_ref] it's not a _CXM_GET_FILE_REF"); | |
| 293 | - false | |
| 294 | -. | ... | ... |
| 1 | +/* | |
| 2 | + * Created by PyramIDE. | |
| 3 | + * User: フランスのトトロ aka (David RENÉ) | |
| 4 | + * Date: 26/05/2019 | |
| 5 | + * Time: 15:23 | |
| 6 | + * © David RENÉ | |
| 7 | + */ | |
| 8 | + | |
| 9 | + | |
| 10 | +read calexium_lib/CXM_message_constants.anubis | |
| 11 | +read calexium_lib/net_services/CXM_generic_protocol.anubis | |
| 12 | +read calexium_lib/net_services/CXM_generic_client.anubis //for get_ip | |
| 13 | +read system/message_queue.anubis | |
| 14 | +read system/message_transceiver.anubis | |
| 15 | +read system/muscle.anubis | |
| 16 | +read system/files.anubis | |
| 17 | +read tools/basis.anubis | |
| 18 | +read network/dns.anubis | |
| 19 | + | |
| 20 | +define Maybe(One) | |
| 21 | + receive_data | |
| 22 | + ( | |
| 23 | + MessageQueue mQ, | |
| 24 | + WStream fd, | |
| 25 | + Int so_far, | |
| 26 | + Int left_read | |
| 27 | + )= | |
| 28 | + //println("receive_data "); | |
| 29 | + if mQ.get_next_received_Message(30) is | |
| 30 | + { | |
| 31 | + timeout then failure, | |
| 32 | + closed then failure, | |
| 33 | + msg(_msg) then | |
| 34 | + with last_block = if find_bool(_msg, "End") is {failure then false, success(r) then r}, | |
| 35 | + if find_raw(_msg, "Data") is | |
| 36 | + { | |
| 37 | + failure then println("Can't find raw Data"); send_ACK_error(mQ, _CXM_FTP_DATA); failure, | |
| 38 | + success(data) then | |
| 39 | + if write(fd, data) is | |
| 40 | + { | |
| 41 | + failure then println("Can't write into file"); send_ACK_error(mQ, _CXM_FTP_DATA);failure, | |
| 42 | + success(len) then | |
| 43 | + send_ACK_ok(mQ, _CXM_FTP_DATA); | |
| 44 | + if last_block then | |
| 45 | + println("File received successfully"); | |
| 46 | + //TODO this is a big hack, we must check if all data are sent from the mQ | |
| 47 | + sleep(10000); | |
| 48 | + success(unique) | |
| 49 | + else | |
| 50 | +// println("Bytes received : " + (so_far + len)); | |
| 51 | + receive_data(mQ, fd, so_far + len, left_read - len) | |
| 52 | + } | |
| 53 | + } | |
| 54 | + } | |
| 55 | +. | |
| 56 | + | |
| 57 | +define Maybe(One) | |
| 58 | + start_get_file_transtert | |
| 59 | + ( | |
| 60 | + MessageQueue mQ, | |
| 61 | + String local_file, | |
| 62 | + Int size | |
| 63 | + )= | |
| 64 | + make_directories(local_file); | |
| 65 | + if file(local_file, new) is | |
| 66 | + { | |
| 67 | + failure then println("Can't create \""+local_file+"\" file"); failure, | |
| 68 | + success(fd) then //the local file is open | |
| 69 | + with start = message(_CXM_FTP_START_TRANSFERT), | |
| 70 | + mQ.add_Message_to_send(start); | |
| 71 | + receive_data(mQ, weaken(fd), 0, size) | |
| 72 | + } | |
| 73 | + . | |
| 74 | + | |
| 75 | +define Bool | |
| 76 | + get_file_ref | |
| 77 | + ( | |
| 78 | + MessageQueue mQ, | |
| 79 | + File_ref f_ref, | |
| 80 | + String tmp_path, | |
| 81 | + (LogLevel, String) -> One logger | |
| 82 | + )= | |
| 83 | + with target_file = tmp_path + "/" + f_ref.name, | |
| 84 | + with get_file_msg = message(_CXM_GET_FILE_REF), | |
| 85 | + forget(add_message(get_file_msg, "GET_FILE_REF", to_Message(f_ref))); | |
| 86 | + | |
| 87 | + mQ.add_Message_to_send(get_file_msg); | |
| 88 | + if wait_for_reply(mQ, _CXM_FTP_GET_FILE, 30) is | |
| 89 | + { | |
| 90 | + failure then println("get_file failure");failure, | |
| 91 | + timeout then println("get_file timeout");failure, | |
| 92 | + unknow_cmd then println("get_file unknow_cmd");failure, | |
| 93 | + error then println("get_file remote error msg");failure, | |
| 94 | + error(c,m) then println("get_file remote error [" + c + "] msg '" + m + "'");failure, | |
| 95 | + ok then println("get_file ok");failure, | |
| 96 | + ok_msg(msg)then | |
| 97 | + 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_}}, | |
| 98 | + start_get_file_transtert(mQ, target_file, size) | |
| 99 | + } | |
| 100 | +. | |
| 101 | + | |
| 102 | + | |
| 103 | +define Bool | |
| 104 | + request_for_service | |
| 105 | + ( | |
| 106 | + MessageQueue queue | |
| 107 | + )= | |
| 108 | + with test_msg = message(_CXM_REQUEST_FOR_SERVICE), | |
| 109 | + forget(add_int32(test_msg, "service", _CXM_FTP_SERVICE_ID)); | |
| 110 | + forget(add_int32(test_msg, "version", 1)); | |
| 111 | + queue.add_Message_to_send(test_msg); | |
| 112 | + if queue.get_next_received_Message(30) is | |
| 113 | + { | |
| 114 | + timeout then println("request_for_service receive timeout");false, | |
| 115 | + closed then println("request_for_service socket closed");false, | |
| 116 | + msg(msg) then | |
| 117 | + if find_int32(msg, "STATUS") is | |
| 118 | + { | |
| 119 | + failure then println("status not found");false, | |
| 120 | + success(v) then | |
| 121 | + if v = _CXM_OK then | |
| 122 | + true | |
| 123 | + else | |
| 124 | + false | |
| 125 | + } | |
| 126 | + }. | |
| 127 | + | |
| 128 | +public define Maybe(One) | |
| 129 | + ftp_get_file | |
| 130 | + ( | |
| 131 | + String server, | |
| 132 | + Word32 ip_port, | |
| 133 | + String remote_file, | |
| 134 | + String local_file, | |
| 135 | + Bool ftp, | |
| 136 | + ) = | |
| 137 | + if mb_get_ip(server) is | |
| 138 | + { | |
| 139 | + failure then println("server "+server+" DNS error");failure, | |
| 140 | + success(ip_adr) then | |
| 141 | + if connect( ip_adr, ip_port) is | |
| 142 | + { | |
| 143 | + error(_) then println("can't connect to ftp server"); failure, | |
| 144 | + ok(conn) then | |
| 145 | + with queue = create_MessageQueue("ftp_get_file", tcp(conn)), | |
| 146 | + message_transceiver(/*conn,*/ queue); | |
| 147 | + println("request for ftp service"); | |
| 148 | + if request_for_service(queue) then | |
| 149 | + with result = get_file(queue, remote_file, local_file, ftp), | |
| 150 | + queue.quit(unique);result | |
| 151 | + else | |
| 152 | + queue.quit(unique); | |
| 153 | + println("Service not found");failure | |
| 154 | + } | |
| 155 | + }. | |
| 156 | + | |
| 157 | + /* Exists for backward compatibility. | |
| 158 | + * The last argument set to false (no ftp directory on the remote, pickup on collection dir) | |
| 159 | + */ | |
| 160 | + | |
| 161 | +public define Maybe(One) | |
| 162 | + ftp_get_file | |
| 163 | + ( | |
| 164 | + String server, | |
| 165 | + Word32 ip_port, | |
| 166 | + String remote_file, | |
| 167 | + String local_file, | |
| 168 | + ) = | |
| 169 | + ftp_get_file(server, ip_port, remote_file, local_file, false). | |
| 170 | + | ... | ... |
| 1 | +/* | |
| 2 | + * Created by PyramIDE. | |
| 3 | + * User: フランスのトトロ aka (David RENÉ) | |
| 4 | + * Date: 26/05/2019 | |
| 5 | + * Time: 15:21 | |
| 6 | + * © David RENÉ | |
| 7 | + */ | |
| 8 | + | |
| 9 | + | |
| 10 | +read tools/basis.anubis | |
| 11 | +read tools/findstring.anubis | |
| 12 | +read system/muscle.anubis | |
| 13 | +read system/data_io.anubis | |
| 14 | +read system/convert.anubis | |
| 15 | +read system/string.anubis | |
| 16 | +read tools/basis.anubis | |
| 17 | +read system/message_queue.anubis | |
| 18 | +read system/files.anubis | |
| 19 | +read system/logger.anubis | |
| 20 | + | |
| 21 | +read calexium_lib/net_services_protocols/logger_service.anubis | |
| 22 | +read calexium_lib/CXM_message_constants.anubis | |
| 23 | +read calexium_lib/net_services/CXM_net_services.anubis | |
| 24 | +read calexium_lib/net_services/CXM_generic_protocol.anubis | |
| 25 | +read calexium_lib/types/generated/file_ref.anubis | |
| 26 | + read app_constants.anubis | |
| 27 | + | |
| 28 | +read types/app_types.anubis | |
| 29 | + read tools/app_loggers.anubis | |
| 30 | + | |
| 31 | +//TODO add return error handling in every functions | |
| 32 | + | |
| 33 | +public define Word32 updater_srv_version = 1. | |
| 34 | + | |
| 35 | +define Maybe(One) | |
| 36 | + send_file_state | |
| 37 | + ( | |
| 38 | + MessageQueue mQ, | |
| 39 | + RStream src_file, | |
| 40 | + Int left_read, | |
| 41 | + (LogLevel, String) -> One logger | |
| 42 | + )= | |
| 43 | + with read_size = min(65536, left_read), | |
| 44 | + if read(src_file, read_size, 10) is | |
| 45 | + { | |
| 46 | + error then logger(logError, "send_file_state read error");failure, | |
| 47 | + timeout then logger(logError, "send_file_state timeout");failure, | |
| 48 | + ok(buffer)then | |
| 49 | + with data_msg = message(_CXM_FTP_DATA), | |
| 50 | + if add_raw(data_msg, "Data", buffer) is | |
| 51 | + { | |
| 52 | + failure then logger(logError, "send_file_state add_raw failure ");failure, | |
| 53 | + success(_) then | |
| 54 | + with end_value = if left_read - read_size = 0 then | |
| 55 | + true | |
| 56 | + else | |
| 57 | + false, | |
| 58 | + | |
| 59 | + if add_bool(data_msg, "End", end_value) is | |
| 60 | + { | |
| 61 | + failure then failure, | |
| 62 | + success(_) then | |
| 63 | + //logDebug(main_log, "send_file_state send message"); | |
| 64 | + mQ.add_Message_to_send(data_msg); | |
| 65 | + if mQ.get_next_received_Message(30) is | |
| 66 | + { | |
| 67 | + timeout then logger(logError, "send_file_state timeout"); failure, | |
| 68 | + closed then logger(logWarning, "send_file_state closed"); success(unique), | |
| 69 | + msg(_msg) then | |
| 70 | + if *_msg.what = _CXM_ACK then | |
| 71 | + if find_int32(_msg, "CMD") is | |
| 72 | + { | |
| 73 | + failure then logger(logError, "send_file_state CMD not found");failure, | |
| 74 | + success(cmd) then | |
| 75 | + if find_int32(_msg, "STATUS") is | |
| 76 | + { | |
| 77 | + failure then logger(logError, "send_file_state STATUS not found");failure, | |
| 78 | + success(status) then | |
| 79 | + if cmd = _CXM_FTP_DATA & status = _CXM_OK then | |
| 80 | + if end_value then //it was the last block, then quit here | |
| 81 | + logger(logDebug, "FTP Server: File sent successfully"); | |
| 82 | + success(unique) | |
| 83 | + else | |
| 84 | + send_file_state(mQ, src_file, left_read - read_size, logger) | |
| 85 | + else | |
| 86 | + logger(logError, "send_file_state wrong command or status");failure | |
| 87 | + } | |
| 88 | + } | |
| 89 | + else | |
| 90 | + logger(logError, "send_file_state Not ACK message");failure | |
| 91 | + } | |
| 92 | + } | |
| 93 | + } | |
| 94 | + }. | |
| 95 | + | |
| 96 | +define String | |
| 97 | + extract_hash | |
| 98 | + ( | |
| 99 | + String full_path | |
| 100 | + ) = | |
| 101 | + with path = (List(String))extract_dir(full_path), | |
| 102 | + get_last = (List(String) dirs, String last) |-get_last-> | |
| 103 | + if dirs is | |
| 104 | + { | |
| 105 | + [] then last, | |
| 106 | + [h . t] then | |
| 107 | + if length(h) > 0 then get_last(t, h) | |
| 108 | + else get_last(t, last) | |
| 109 | + }, | |
| 110 | + get_last(path, ""). | |
| 111 | + | |
| 112 | +define Bool | |
| 113 | + send_file_start_transfert | |
| 114 | + ( | |
| 115 | + MessageQueue queue, | |
| 116 | + String file_name, | |
| 117 | + (LogLevel, String) -> One logger | |
| 118 | + )= | |
| 119 | + if queue.get_next_received_Message(30) is | |
| 120 | + { | |
| 121 | + timeout then logger(logWarning, "send_file_start_transfert timeout");false, | |
| 122 | + closed then logger(logWarning, "send_file_start_transfert connection closed");false, | |
| 123 | + msg(_msg) then | |
| 124 | + //check if the message is start the transfer | |
| 125 | + if *_msg.what = _CXM_FTP_START_TRANSFERT then | |
| 126 | + logger(logDebug, "_CXM_FTP_START_TRANSFERT received"); | |
| 127 | + with start_offset = if find_int32(_msg, "ResumeOffset") is | |
| 128 | + { | |
| 129 | + failure then 0, | |
| 130 | + success(value) then to_Int(value) | |
| 131 | + }, | |
| 132 | + if file(file_name, read) is | |
| 133 | + { | |
| 134 | + failure then logger(logError, "send_file_start_transfert can't open file '" + file_name + "'");false, | |
| 135 | + success(source) then | |
| 136 | + if send_file_state(queue, source, file_size(file_name) - start_offset, logger) is | |
| 137 | + { | |
| 138 | + failure then logger(logError, "send_file_start_transfert send_file_state failed"); false, | |
| 139 | + success(_) then logger(logInfo, "send_file_start_transfert file sent successfully"); true | |
| 140 | + } | |
| 141 | + } | |
| 142 | + else | |
| 143 | + false | |
| 144 | + }. | |
| 145 | + | |
| 146 | +public define Bool | |
| 147 | + send_file_ref | |
| 148 | + ( | |
| 149 | + MessageQueue queue, | |
| 150 | + Message msg, | |
| 151 | + (LogLevel, String) -> One logger | |
| 152 | + )= | |
| 153 | + if *msg.what = _CXM_GET_FILE_REF then | |
| 154 | + if find_message(msg, "GET_FILE_REF") is { | |
| 155 | + failure then logger(logError, "NET_SERVICE send_file_ref: can't find GET_FILE_REF message"); false, | |
| 156 | + success(file_ref_obj_msg) then | |
| 157 | + if (Maybe(File_ref))from_Message(file_ref_obj_msg) is | |
| 158 | + { | |
| 159 | + failure then logger(logError, "NET_SERVICE send_file_ref: can't extract GET_FILE_REF message");false, | |
| 160 | + success(f_ref) then | |
| 161 | + with source_file = f_ref.path + "/" + f_ref.name, | |
| 162 | + if file_exists(source_file) then | |
| 163 | + with result_msg = message(_CXM_ACK_RESULT_MSG), | |
| 164 | + with size = file_size(source_file), | |
| 165 | + forget(add_string(result_msg, "SIZE", to_String(size))); | |
| 166 | + | |
| 167 | + //forget(add_int32(result_msg, "FileSize", truncate_to_Word32(file_size(file_name)))); | |
| 168 | + logger(logDebug, "send _CXM_FTP_GET_FILE with filesize "+file_size(source_file)); | |
| 169 | + send_result(queue, _CXM_FTP_GET_FILE, success(result_msg)); | |
| 170 | + send_file_start_transfert(queue, source_file, logger) | |
| 171 | + else | |
| 172 | + logger(logError, "File '"+source_file+"' not found"); | |
| 173 | + send_ACK_error(queue, _CXM_FTP_GET_FILE); | |
| 174 | + false | |
| 175 | + } | |
| 176 | + } | |
| 177 | + else | |
| 178 | + false | |
| 179 | +. | |
| 180 | + | |
| 181 | + | |
| 182 | + | |
| 183 | + public define Bool | |
| 184 | + send_file_ref | |
| 185 | + ( | |
| 186 | + MessageQueue mQ, | |
| 187 | + Message msg, | |
| 188 | + (LogLevel, String) -> One logger | |
| 189 | + )= | |
| 190 | + //println("NET_SERVICE send_file_ref received *"); | |
| 191 | + if *msg.what = _CXM_GET_FILE_REF then | |
| 192 | + if find_message(msg, "GET_FILE_REF") is { | |
| 193 | + failure then println("NET_SERVICE send_file_ref: can't find GET_FILE_REF message"); false, | |
| 194 | + success(file_ref_obj_msg) then | |
| 195 | + if (Maybe(File_ref))from_Message(file_ref_obj_msg) is | |
| 196 | + { | |
| 197 | + failure then println("NET_SERVICE send_file_ref: can't extract GET_FILE_REF message");false, | |
| 198 | + success(f_ref) then | |
| 199 | + //println("NET_SERVICE send_file_ref: GET_FILE_REF message File_ref extracted\n"+dump(file_ref_obj_msg)); | |
| 200 | + //create the target | |
| 201 | + with source_file = f_ref.path + "/" + f_ref.name, | |
| 202 | + with msg = message(0), | |
| 203 | + with size = file_size(source_file), | |
| 204 | + | |
| 205 | + println("source_file = "+source_file); | |
| 206 | + println("size = "+size); | |
| 207 | + forget(add_string(msg, "MODE", "NEW")); | |
| 208 | + forget(add_string(msg, "SIZE", to_String(size))); | |
| 209 | + | |
| 210 | + //println("NET_SERVICE send_file_ref: reply message \n"+dump(msg)); | |
| 211 | + if (Maybe(RStream))file(source_file, read) is | |
| 212 | + { | |
| 213 | + failure then println("NET_SERVICE send_file_ref: can't open source file"+source_file);false, //nothing to write | |
| 214 | + success(source) then | |
| 215 | + | |
| 216 | + //println("NET_SERVICE send_file_ref send ack file size "+size); | |
| 217 | + send_ACK_ok(mQ, _CXM_GET_FILE_REF, msg); | |
| 218 | + //println("call flush_wait"); | |
| 219 | + mQ.flush_wait(unique); | |
| 220 | + if seek(source, 0) then | |
| 221 | + if copy_file_to_Connection(file(source), mQ.get_connection(unique), size) is | |
| 222 | + { | |
| 223 | + failure then println("NET_SERVICE send_file_ref copy_file_to_Connection error");false, | |
| 224 | + success(_) then println("NET_SERVICE send_file_ref copy_file_to_Connection ok");true | |
| 225 | + } | |
| 226 | + else | |
| 227 | + println("NET_SERVICE send_file_ref seek error");false, | |
| 228 | + | |
| 229 | + } | |
| 230 | + } | |
| 231 | + } | |
| 232 | + else //not _CXM_GET_FILE_REF | |
| 233 | + println("[send_file_ref] it's not a _CXM_GET_FILE_REF"); | |
| 234 | + false | |
| 235 | +. | ... | ... |