diff --git a/net_services/CXM_generic_client.anubis b/net_services/CXM_generic_client.anubis index 4ded637..42502c9 100644 --- a/net_services/CXM_generic_client.anubis +++ b/net_services/CXM_generic_client.anubis @@ -8,6 +8,7 @@ */ transmit system/message_transceiver.anubis +transmit system/logger.anubis transmit calexium_lib/net_services_protocols/logger_service.anubis transmit calexium_lib/CXM_message_constants.anubis @@ -66,24 +67,24 @@ public define Maybe(NetServiceAnswer) */ generic_send_message ( - MessageQueue queue, - Message msg_to_send, - Int timeout, - String logger_name + MessageQueue queue, + Message msg_to_send, + Int timeout, + (LogLevel, String) -> One logger )= queue.add_Message_to_send(msg_to_send); if queue.get_next_received_Message(timeout) is { - timeout then logError(logger_name, "["+queue.get_name(unique)+"]: receive timeout");failure, - closed then logError(logger_name, "["+queue.get_name(unique)+"]: socket closed");failure, + timeout then logger(logError, "["+queue.get_name(unique)+"]: receive timeout");failure, + closed then logger(logError, "["+queue.get_name(unique)+"]: socket closed");failure, msg(msg) then if find_int32(msg, "CMD") is { - failure then logError(logger_name, "["+queue.get_name(unique)+"]: CMD field not found"); failure, + failure then logger(logError, "["+queue.get_name(unique)+"]: CMD field not found"); failure, success(cmd) then if find_int32(msg, "STATUS") is { - failure then logError(logger_name, "["+queue.get_name(unique)+"]: STATUS field not found"); failure, + failure then logger(logError, "["+queue.get_name(unique)+"]: STATUS field not found"); failure, success(v) then if v = _CXM_OK then success(netservice_ok(cmd, find_message(msg, "RESULT"))) @@ -102,7 +103,7 @@ public define Maybe($T) Int timeout, Message msg_to_send, (Message, MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = if generic_send_message(queue, msg_to_send, timeout, logger) is { @@ -111,12 +112,12 @@ public define Maybe($T) if net_result is { netservice_error(cmd, err_code, err_str) then - logger("["+queue.get_name(unique)+"]: message status ERROR [0x" + to_hexa(err_code) + ", '" + err_str + "']"); + logger(logError, "["+queue.get_name(unique)+"]: message status ERROR [0x" + to_hexa(err_code) + ", '" + err_str + "']"); failure, netservice_ok(cmd, mb_msg) then if mb_msg is { - failure then logger("["+queue.get_name(unique)+"]: can't find RESULT message."); failure, + failure then logger(logError,"["+queue.get_name(unique)+"]: can't find RESULT message."); failure, success(result) then handler(result, queue, timestamp) } } @@ -125,10 +126,10 @@ public define Maybe($T) public define Maybe(One) no_result_handler ( - MessageQueue queue, - Int timeout, - Message msg_to_send, - (String) -> One logger + MessageQueue queue, + Int timeout, + Message msg_to_send, + (LogLevel, String) -> One logger ) = if generic_send_message(queue, msg_to_send, timeout, logger) is { @@ -137,13 +138,13 @@ public define Maybe(One) if net_result is { netservice_error(cmd, err_code, err_str) then - logger("["+queue.get_name(unique)+"]: message status ERROR [0x" + to_hexa(err_code) + ", '" + err_str + "']"); + logger(logError, "["+queue.get_name(unique)+"]: message status ERROR [0x" + to_hexa(err_code) + ", '" + err_str + "']"); failure, netservice_ok(cmd, mb_msg) then if mb_msg is { failure then unique, - success(result) then logger("An unattended RESULT msg was found. Ignoring it...") + success(result) then logger(logError, "An unattended RESULT msg was found. Ignoring it...") }; success(unique) } @@ -155,7 +156,7 @@ public define (MessageQueue, String) -> Maybe($T) Int timeout, Message msg_to_send, (Message, MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = (MessageQueue queue, String timestamp) |-> simple_handler(queue, timestamp, timeout, msg_to_send, handler, logger). @@ -168,7 +169,7 @@ define Maybe($T) Word32 service_version, String domain, (MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger )= with test_msg = message(_CXM_REQUEST_FOR_SERVICE), forget(add_int32(test_msg, "SERVICE", service_id)); @@ -177,71 +178,71 @@ define Maybe($T) queue.add_Message_to_send(test_msg); if queue.get_next_received_Message(10) is { - timeout then logger("["+queue.get_name(unique)+"]: requesting service receive timeout");failure, - closed then logger("["+queue.get_name(unique)+"]: requesting service socket closed");failure, + timeout then logger(logError, "["+queue.get_name(unique)+"]: requesting service receive timeout");failure, + closed then logger(logError, "["+queue.get_name(unique)+"]: requesting service socket closed");failure, msg(msg) then if find_int32(msg, "STATUS") is { - failure then logger("["+queue.get_name(unique)+"]: requesting service STATUS not found");failure, + failure then logger(logError, "["+queue.get_name(unique)+"]: requesting service STATUS not found");failure, success(v) then if v = _CXM_OK then if find_message(msg, "RESULT") is { - failure then logger("["+queue.get_name(unique)+"]: requesting service RESULT not found");failure, + failure then logger(logError, "["+queue.get_name(unique)+"]: requesting service RESULT not found");failure, success(result) then with timestamp = if find_string(result, "TIMESTAMP") is { - failure then logger("["+queue.get_name(unique)+"]: requesting service TIMESTAMP not found"); "", + failure then logger(logError, "["+queue.get_name(unique)+"]: requesting service TIMESTAMP not found"); "", success(timestamp) then timestamp }, handler(queue, timestamp) } else - logger("["+queue.get_name(unique)+"]: the requested service is not available on server.");failure + logger(logError, "["+queue.get_name(unique)+"]: the requested service is not available on server.");failure } }. -define Maybe($T) - generic_request_for_service - ( - MessageQueue queue, - Word32 service_id, - Word32 service_version, - String domain, - (MessageQueue, String) -> Maybe($T) handler, - String logger - )= - with test_msg = message(_CXM_REQUEST_FOR_SERVICE), - forget(add_int32(test_msg, "SERVICE", service_id)); - forget(add_int32(test_msg, "VERSION", service_version)); - forget(add_string(test_msg, "DOMAIN", domain)); - queue.add_Message_to_send(test_msg); - if queue.get_next_received_Message(10) is - { - timeout then logError(logger, "["+queue.get_name(unique)+"]: requesting service receive timeout");failure, - closed then logError(logger, "["+queue.get_name(unique)+"]: requesting service socket closed");failure, - msg(msg) then - if find_int32(msg, "STATUS") is - { - failure then logError(logger, "["+queue.get_name(unique)+"]: requesting service STATUS not found");failure, - success(v) then - if v = _CXM_OK then - if find_message(msg, "RESULT") is - { - failure then logError(logger, "["+queue.get_name(unique)+"]: requesting service RESULT not found");failure, - success(result) then - with timestamp = if find_string(result, "TIMESTAMP") is - { - failure then logError(logger, "["+queue.get_name(unique)+"]: requesting service TIMESTAMP not found"); "", - success(timestamp) then timestamp - }, - handler(queue, timestamp) - } - else - logError(logger, "["+queue.get_name(unique)+"]: the requested service is not available on server.");failure - } - }. - +// define Maybe($T) +// generic_request_for_service +// ( +// MessageQueue queue, +// Word32 service_id, +// Word32 service_version, +// String domain, +// (MessageQueue, String) -> Maybe($T) handler, +// (LogLevel, String) -> One logger +// )= +// with test_msg = message(_CXM_REQUEST_FOR_SERVICE), +// forget(add_int32(test_msg, "SERVICE", service_id)); +// forget(add_int32(test_msg, "VERSION", service_version)); +// forget(add_string(test_msg, "DOMAIN", domain)); +// queue.add_Message_to_send(test_msg); +// if queue.get_next_received_Message(10) is +// { +// timeout then logger(logError, "["+queue.get_name(unique)+"]: requesting service receive timeout");failure, +// closed then logger(logError, "["+queue.get_name(unique)+"]: requesting service socket closed");failure, +// msg(msg) then +// if find_int32(msg, "STATUS") is +// { +// failure then logger(logError, "["+queue.get_name(unique)+"]: requesting service STATUS not found");failure, +// success(v) then +// if v = _CXM_OK then +// if find_message(msg, "RESULT") is +// { +// failure then logger(logError, "["+queue.get_name(unique)+"]: requesting service RESULT not found");failure, +// success(result) then +// with timestamp = if find_string(result, "TIMESTAMP") is +// { +// failure then logger(logError, "["+queue.get_name(unique)+"]: requesting service TIMESTAMP not found"); "", +// success(timestamp) then timestamp +// }, +// handler(queue, timestamp) +// } +// else +// logger(logError, "["+queue.get_name(unique)+"]: the requested service is not available on server.");failure +// } +// }. +// public define Maybe(MessageQueue) get_message_queue_to_net_service ( @@ -252,12 +253,12 @@ public define Maybe(MessageQueue) Word32 service_version, String domain, (MessageQueue, String) -> Maybe(One) handler, //1st function to apply if need (i.e authentication to remote service) - String logger + (LogLevel, String) -> One logger ) = if connect( server, port) is { - error(_) then logError(logger, queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure, + error(_) then logger(logError, queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure, ok(conn) then // println("[" + virtual_machine_id + "] netservices create queue"); with queue = create_MessageQueue(queue_name, tcp(conn)), @@ -267,7 +268,7 @@ public define Maybe(MessageQueue) { failure then //we can't apply first function correctly, so we ask to Message Queue to quit and return failure - logError(logger, "can't apply first function correctly"); + logger(logError, "can't apply first function correctly"); queue.quit(unique); failure, success(_) then success(queue) @@ -285,12 +286,12 @@ public define Maybe($T) Word32 service_version, String domain, (MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = if connect( server, port) is { - error(_) then logger(queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure, + error(_) then logger(logError, queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure, ok(conn) then // println("[" + virtual_machine_id + "] netservices create queue"); with queue = create_MessageQueue(queue_name, tcp(conn)), @@ -313,7 +314,7 @@ public define Maybe($T) Word32 service_id, Word32 service_version, (MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = generic_connect_to_net_service(queue_name, server, port, service_id, service_version, "", handler, logger). @@ -322,37 +323,38 @@ define Word32 get_ip ( String url_or_ip, - (String) -> One logger + (LogLevel, String) -> One logger ) = - //logInfo(debug_log,"Try to resolve URL [" + url_or_ip+ "]."); + logger(logTrace, "Try to resolve URL [" + url_or_ip+ "]."); if ip_address(url_or_ip) is success(ip) then - //logInfo(debug_log,"Try to resolve URL OK ip"+ip); + logger(logTrace, "Try to resolve URL OK ip"+ip); ip else if dns(url_or_ip) is ok(ip_adr) then - //logInfo(debug_log,"Try to resolve URL OK dns"+ip_adr); + logger(logTrace, "Try to resolve URL OK dns"+ip_adr); ip_adr else - logger("Can't resolve URL [" + url_or_ip+ "]. Using localhost."); + logger(logError, "Can't resolve URL [" + url_or_ip+ "]. Using localhost (127.0.0.1) ."); ip_address((127,0,0,1)). public define Maybe(Word32) mb_get_ip ( - String url_or_ip + String url_or_ip, + (LogLevel, String) -> One logger ) = - //logInfo(debug_log,"Try to resolve URL [" + url_or_ip+ "]."); + logger(logTrace, "Try to resolve URL [" + url_or_ip+ "]."); if ip_address(url_or_ip) is success(ip) then - //logInfo(debug_log,"Try to resolve URL OK ip"+ip); + logger(logTrace, "Try to resolve URL OK ip"+ip); success(ip) else if dns(url_or_ip) is ok(ip_adr) then - //logInfo(debug_log,"Try to resolve URL OK dns"+ip_adr); + logger(logTrace, "Try to resolve URL OK dns"+ip_adr); success(ip_adr) else - println("Can't resolve URL [" + url_or_ip+ "]."); + logger(logError, "Can't resolve URL [" + url_or_ip+ "]."); failure. -public define Word32 + public define Word32 get_ip ( String url_or_ip @@ -377,7 +379,7 @@ public define Maybe($T) Word32 service_version, String domain, (MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = generic_connect_to_net_service(queue_name, get_ip(server, logger), port, service_id, service_version, domain, handler, logger). @@ -390,7 +392,7 @@ public define Maybe($T) Word32 service_id, Word32 service_version, (MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = generic_connect_to_net_service(queue_name, server, port, service_id, service_version, "", handler, logger). @@ -407,12 +409,12 @@ public define Maybe($T) Word32 service_version, String domain, (MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = if open_SSL_connection( server_name, server_ip, port, accept_policy) is { - error(_) then logger(queue_name + ": Can't connect to domain manager ["+ip_addr_to_string(server_ip)+":"+port+"]");failure, + error(_) then logger(logError, queue_name + ": Can't connect to domain manager ["+ip_addr_to_string(server_ip)+":"+port+"]");failure, ok(conn) then with queue = create_MessageQueue(queue_name, ssl(conn)), message_transceiver(/*ssl(conn),*/ queue); @@ -434,7 +436,7 @@ public define Maybe($T) Word32 service_id, Word32 service_version, (MessageQueue, String) -> Maybe($T) handler, - (String) -> One logger + (LogLevel, String) -> One logger ) = generic_connect_to_net_service_SSL(queue_name, server_name, server_ip, port, accept_policy, service_id, service_version, "", handler, logger). diff --git a/net_services/CXM_net_services.anubis b/net_services/CXM_net_services.anubis index 451b016..4a0d63c 100644 --- a/net_services/CXM_net_services.anubis +++ b/net_services/CXM_net_services.anubis @@ -13,6 +13,7 @@ read system/muscle.anubis read system/data_io.anubis read system/message_queue.anubis read system/message_transceiver.anubis +read system/logger.anubis read CXM_generic_protocol.anubis read calexium_lib/CXM_message_constants.anubis @@ -22,24 +23,27 @@ public type NetService: Word32 id, String name, List(String) domains, - (MessageQueue, String, String) -> One handler // Parameters are MessageQueue, peer IP and timestamp string - ). + (MessageQueue, String, String, (LogLevel, String) -> One ) -> One handler // Parameters are MessageQueue, peer IP and timestamp string, logger + ) +. -define One - print_services +define String + dump_services ( List(NetService) net_services ) = - - map_forget((NetService net_s)|-> - if net_s is net_service(version, id, name, domains, _) then - println(" id : 0x"+ to_hexa(id)); - println(" version : " + to_String(version)); - println(" name : "+ name ); - println(" domains : "); - map_forget((String domain) |-> println(" : "+domain), domains); - println("----------------------------------------") - ,net_services). + join("\n", map((NetService net_s) + |-> + if net_s is net_service(version, id, name, domains, _) then + " id : 0x"+ to_hexa(id)+"\n"+ + " version : " + to_String(version)+"\n"+ + " name : "+ name +"\n"+ + " domains : "+"\n"+ + join("\n",map((String domain) |-> " : "+domain, domains))+"\n"+ + "----------------------------------------\n" + ,net_services) + ) +. /** Try to find the service_id in services_list. If the service is found in that list * the corresponding NetService object is return @@ -47,14 +51,14 @@ define One define Maybe(NetService) find_service ( - List(NetService) services_list, - Word32 service_id, - Word32 service_version, - String domain + List(NetService) services_list, //List of all available NetService + Word32 service_id, //requested service ID + Word32 service_version, //requested service Version + String domain //requested domain for above resquested service ID/version )= if services_list is { - [] then failure, + [] then failure, [h . t] then if h.id = service_id & h.version >=+ service_version then if domain = "" then @@ -65,7 +69,8 @@ define Maybe(NetService) find_service(t, service_id, service_version, domain) else find_service(t, service_id, service_version, domain) - }. + } +. /** Check if the muscle message msg has the correct fields for requesting a net_services * if we found "service" and "version" fields on the message, we try to find if the service @@ -111,7 +116,8 @@ define Maybe(NetService) success(domain) then find_service(net_services, service_id, service_version, domain) } } - }. + } +. define String get_time_stamp @@ -127,23 +133,26 @@ define String define One service_negociation ( - MessageQueue queue, - Message msg, - List(NetService) net_services, - String peer + MessageQueue queue, + Message msg, + List(NetService) net_services, + String peer, + (LogLevel, String) -> One logger )= - //println("Service NEGOCIATION [" + to_hexa(*msg.what) + "] received"); + logger(logTrace, "Service NEGOCIATION [" + to_hexa(*msg.what) + "] received"); if * msg.what = _CXM_REQUEST_FOR_SERVICE then if has_service(queue, msg, net_services) is { failure then + logger(logError, "Unknown service"); send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE, _CXM_UNKNOW_SERVICE, "Unknown service") + success(net_service) then with result = message(0), timestamp = get_time_stamp, forget(add_string(result, "TIMESTAMP", timestamp)); send_ACK_ok(queue, _CXM_REQUEST_FOR_SERVICE, result); - net_service.handler(queue, peer, timestamp) + net_service.handler(queue, peer, timestamp, logger) } else send_ACK_error(queue, *msg.what, _CXM_UNKNOW_CMD, "Unknown command [" + (*msg.what) + "]") @@ -155,59 +164,64 @@ define One public define One message_receiver ( - MessageQueue queue, - List(NetService) net_services, - String peer + MessageQueue queue, + List(NetService) net_services, + String peer, + (LogLevel, String) -> One logger ) = if queue.quit_requested(unique) then unique else - //println("PRE SERVICE message_receiver "+"["+virtual_machine_id + "]"); + logger(logTrace,"PRE SERVICE message_receiver ["+virtual_machine_id + "]"); if queue.get_next_received_Message(1) is { timeout then //println("PRE timeout"); - message_receiver(queue, net_services, peer), - closed then //println("PRE closed"); + message_receiver(queue, net_services, peer, logger), + closed then + logger(logTrace,"PRE SERVICE message_receiver ["+virtual_machine_id + "] closed"); unique, msg(msg) then unique; //println("PRE negociation"); - service_negociation(queue, msg, net_services, peer); - message_receiver(queue, net_services, peer) + service_negociation(queue, msg, net_services, peer, logger); + message_receiver(queue, net_services, peer, logger) }. define Server -> (RWStream) -> One net_services_handler ( - List(NetService) net_services, + List(NetService) net_services, + (LogLevel, String) -> One logger ) = (Server server) |-> (RWStream conn) |-> if remote_IP_address_and_port(conn) is (num_peer,_) then //convert IP address of the client to string with peer = ip_addr_to_string(num_peer), - //println("NET SERVICES Accepting connection with "+peer); + logger(logInfo,"NET SERVICES Accepting connection with "+peer); //now managing the list of SERVICES with queue = create_MessageQueue("CXM Net Services", tcp(conn)), - message_transceiver(/*conn,*/ queue); - message_receiver(queue, net_services, peer). + message_transceiver(queue); + message_receiver(queue, net_services, peer, logger). public define Maybe(Server) start_net_services ( - List(NetService) net_services, - Word32 network_port, + List(NetService) net_services, + Word32 network_port, + (LogLevel, String) -> One logger )= if start_server(0, network_port, - net_services_handler(net_services), + net_services_handler(net_services, logger), (One u) |-> unique) is { - cannot_create_the_socket then println("Cannot create the listening socket."); failure, - cannot_bind_to_port then println("Cannot bind to port " + network_port ); failure, - cannot_listen_on_port then println("Cannot listen on port " + network_port); failure, + cannot_create_the_socket then logger(logError, "Cannot create the listening socket."); failure, + cannot_bind_to_port then logger(logError, "Cannot bind to port " + network_port); failure, + cannot_listen_on_port then logger(logError, "Cannot listen on port " + network_port); failure, ok(server) then - println("Net services started on port " + network_port); - println("------ Available services ------"); - print_services(net_services); + logger(logInfo, "Net services started on port " + network_port); + logger(logInfo, "------ Available services ------"); + logger(logInfo, dump_services(net_services)); success(server) - }. + } +. diff --git a/net_services_protocols/ftp_client.anubis b/net_services_protocols/ftp_client.anubis index bb4422d..7ac7c05 100644 --- a/net_services_protocols/ftp_client.anubis +++ b/net_services_protocols/ftp_client.anubis @@ -145,18 +145,19 @@ public define Maybe(One) String remote_file, String local_file, Bool ftp, + (LogLevel, String) -> One logger ) = - if mb_get_ip(server) is + if mb_get_ip(server, logger) is { - failure then println("server "+server+" DNS error");failure, + failure then logger(logError, "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, + error(_) then logger(logError, "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"); + logger(logInfo, "request for ftp service"); if request_for_service(queue) then with result = get_file(queue, remote_file, local_file, ftp), queue.quit(unique);result @@ -177,6 +178,7 @@ public define Maybe(One) Word32 ip_port, String remote_file, String local_file, + (LogLevel, String) -> One logger ) = - ftp_get_file(server, ip_port, remote_file, local_file, false). + ftp_get_file(server, ip_port, remote_file, local_file, false, logger). -- libgit2 0.21.4