Anubis / Xlib

Commit 8dac565c6ccdf5f38e484397cae9654de185dc50

Authored by David RENÉ
1 parent e1285ef5

According to the main Anubis library, move the Connection into MessageQueue

Showing 3 changed files with 221 additions and 221 deletions   Show diff stats
Inline Side-by-side
net_services/CXM_generic_client.anubis
  Show/Hide comments   View file @8dac565
@@ -260,8 +260,8 @@ public define Maybe(MessageQueue) @@ -260,8 +260,8 @@ public define Maybe(MessageQueue)
260 error(_) then logError(logger, queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure, 260 error(_) then logError(logger, queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure,
261 ok(conn) then 261 ok(conn) then
262 // println("[" + virtual_machine_id + "] netservices create queue"); 262 // println("[" + virtual_machine_id + "] netservices create queue");
263 - with queue = create_MessageQueue(queue_name),  
264 - message_transceiver(tcp(conn), queue); 263 + with queue = create_MessageQueue(queue_name, tcp(conn)),
  264 + message_transceiver(/*tcp(conn),*/ queue);
265 // println("[" + virtual_machine_id + "] netservices generic_request_for_service()"); 265 // println("[" + virtual_machine_id + "] netservices generic_request_for_service()");
266 if generic_request_for_service(queue, service_id, service_version, domain, handler, logger) is 266 if generic_request_for_service(queue, service_id, service_version, domain, handler, logger) is
267 { 267 {
@@ -293,8 +293,8 @@ public define Maybe($T) @@ -293,8 +293,8 @@ public define Maybe($T)
293 error(_) then logger(queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure, 293 error(_) then logger(queue_name + ": Can't connect to service ["+ip_addr_to_string(server)+":"+port+"]");failure,
294 ok(conn) then 294 ok(conn) then
295 // println("[" + virtual_machine_id + "] netservices create queue"); 295 // println("[" + virtual_machine_id + "] netservices create queue");
296 - with queue = create_MessageQueue(queue_name),  
297 - message_transceiver(tcp(conn), queue); 296 + with queue = create_MessageQueue(queue_name, tcp(conn)),
  297 + message_transceiver(queue);
298 // println("[" + virtual_machine_id + "] netservices generic_request_for_service()"); 298 // println("[" + virtual_machine_id + "] netservices generic_request_for_service()");
299 with result = generic_request_for_service(queue, service_id, service_version, domain, handler, logger), 299 with result = generic_request_for_service(queue, service_id, service_version, domain, handler, logger),
300 // println("[" + virtual_machine_id + "] netservices client quit"); 300 // println("[" + virtual_machine_id + "] netservices client quit");
@@ -414,8 +414,8 @@ public define Maybe($T) @@ -414,8 +414,8 @@ public define Maybe($T)
414 { 414 {
415 error(_) then logger(queue_name + ": Can't connect to domain manager ["+ip_addr_to_string(server_ip)+":"+port+"]");failure, 415 error(_) then logger(queue_name + ": Can't connect to domain manager ["+ip_addr_to_string(server_ip)+":"+port+"]");failure,
416 ok(conn) then 416 ok(conn) then
417 - with queue = create_MessageQueue(queue_name),  
418 - message_transceiver(ssl(conn), queue); 417 + with queue = create_MessageQueue(queue_name, ssl(conn)),
  418 + message_transceiver(/*ssl(conn),*/ queue);
419 with result = generic_request_for_service(queue, service_id, service_version, domain, handler, logger), 419 with result = generic_request_for_service(queue, service_id, service_version, domain, handler, logger),
420 queue.quit(unique); 420 queue.quit(unique);
421 //logInfo(debug_log,"domain_manager client quit"); 421 //logInfo(debug_log,"domain_manager client quit");
net_services/CXM_net_services.anubis
  Show/Hide comments   View file @8dac565
1 -/*  
2 - *  
3 - * User: David RENE  
4 - * Date: 25/04/2007  
5 - * Time: 11:01  
6 - * (c) Calexium  
7 - *  
8 - */  
9 -read tools/basis.anubis  
10 -read system/convert.anubis  
11 -read system/string.anubis  
12 -read system/muscle.anubis  
13 -read system/data_io.anubis  
14 -read system/message_queue.anubis  
15 -read system/message_transceiver.anubis  
16 -read CXM_generic_protocol.anubis  
17 -read calexium_lib/CXM_message_constants.anubis  
18 -  
19 -public type NetService:  
20 - net_service(  
21 - Word32 version,  
22 - Word32 id,  
23 - String name,  
24 - List(String) domains,  
25 - (MessageQueue, String, String) -> One handler // Parameters are MessageQueue, peer IP and timestamp string  
26 - ).  
27 -  
28 -define One  
29 - print_services  
30 - (  
31 - List(NetService) net_services  
32 - ) =  
33 -  
34 - map_forget((NetService net_s)|->  
35 - if net_s is net_service(version, id, name, domains, _) then  
36 - println(" id : 0x"+ to_hexa(id));  
37 - println(" version : " + to_String(version));  
38 - println(" name : "+ name );  
39 - println(" domains : ");  
40 - map_forget((String domain) |-> println(" : "+domain), domains);  
41 - println("----------------------------------------")  
42 - ,net_services).  
43 -  
44 - /** Try to find the service_id in services_list. If the service is found in that list  
45 - * the corresponding NetService object is return  
46 - */  
47 -define Maybe(NetService)  
48 - find_service  
49 - (  
50 - List(NetService) services_list,  
51 - Word32 service_id,  
52 - Word32 service_version,  
53 - String domain  
54 - )=  
55 - if services_list is  
56 - {  
57 - [] then failure,  
58 - [h . t] then  
59 - if h.id = service_id & h.version >=+ service_version then  
60 - if domain = "" then  
61 - success(h)  
62 - else if domain:h.domains then //this writing (a:b) means, is a belonging to b where b is list of type a  
63 - success(h)  
64 - else  
65 - find_service(t, service_id, service_version, domain)  
66 - else  
67 - find_service(t, service_id, service_version, domain)  
68 - }.  
69 -  
70 - /** Check if the muscle message msg has the correct fields for requesting a net_services  
71 - * if we found "service" and "version" fields on the message, we try to find if the service  
72 - * referenced in "service" is available in net_services list  
73 - */  
74 -define Maybe(NetService)  
75 - has_service  
76 - (  
77 - MessageQueue queue,  
78 - Message msg,  
79 - List(NetService) net_services  
80 - )=  
81 - if find_int32(msg, "SERVICE") is  
82 - {  
83 - failure then //send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE); failure,  
84 - // old names... should be removed soon  
85 - if find_int32(msg, "service") is  
86 - {  
87 - failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE); failure,  
88 - success(service_id) then  
89 - if find_int32(msg, "version") is  
90 - {  
91 - failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE);failure,  
92 - success(service_version) then  
93 - //if DOMAIN field exists, this mean we want to target only this domain  
94 - if find_string(msg, "DOMAIN") is  
95 - {  
96 - failure then find_service(net_services, service_id, service_version,""),  
97 - success(domain) then find_service(net_services, service_id, service_version, domain)  
98 - }  
99 - }  
100 - }  
101 -  
102 - success(service_id) then  
103 - if find_int32(msg, "VERSION") is  
104 - {  
105 - failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE);failure,  
106 - success(service_version) then  
107 - //if DOMAIN field exists, this mean we want to target only this domain  
108 - if find_string(msg, "DOMAIN") is  
109 - {  
110 - failure then find_service(net_services, service_id, service_version,""),  
111 - success(domain) then find_service(net_services, service_id, service_version, domain)  
112 - }  
113 - }  
114 - }.  
115 -  
116 -define String  
117 - get_time_stamp  
118 - =  
119 - with time = (UTime) unow,  
120 - "<"+virtual_machine_id+"@"+time.seconds+">".  
121 -  
122 - /** This message_received function just handle the negociation process the available net_services.  
123 - * In other words, it only recognize the _CXM_REQUEST_FOR_SERVICE message and try to launch the  
124 - * corresponding servcice  
125 - */  
126 -  
127 -define One  
128 - service_negociation  
129 - (  
130 - MessageQueue queue,  
131 - Message msg,  
132 - List(NetService) net_services,  
133 - String peer  
134 - )=  
135 - //println("Service NEGOCIATION [" + to_hexa(*msg.what) + "] received");  
136 - if * msg.what = _CXM_REQUEST_FOR_SERVICE then  
137 - if has_service(queue, msg, net_services) is  
138 - {  
139 - failure then  
140 - send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE, _CXM_UNKNOW_SERVICE, "Unknown service")  
141 - success(net_service) then  
142 - with result = message(0),  
143 - timestamp = get_time_stamp,  
144 - forget(add_string(result, "TIMESTAMP", timestamp));  
145 - send_ACK_ok(queue, _CXM_REQUEST_FOR_SERVICE, result);  
146 - net_service.handler(queue, peer, timestamp)  
147 - }  
148 - else  
149 - send_ACK_error(queue, *msg.what, _CXM_UNKNOW_CMD, "Unknown command [" + (*msg.what) + "]")  
150 - .  
151 -  
152 - /**  
153 - * this function unflatten muscle message and give the correct message to service_negociation function  
154 - */  
155 -public define One  
156 - message_receiver  
157 - (  
158 - MessageQueue queue,  
159 - List(NetService) net_services,  
160 - String peer  
161 - ) =  
162 - if queue.quit_requested(unique) then  
163 - unique  
164 - else  
165 - //println("PRE SERVICE message_receiver "+"["+virtual_machine_id + "]");  
166 - if queue.get_next_received_Message(1) is  
167 - {  
168 - timeout then //println("PRE timeout");  
169 - message_receiver(queue, net_services, peer),  
170 - closed then //println("PRE closed");  
171 - unique,  
172 - msg(msg) then unique; //println("PRE negociation");  
173 - service_negociation(queue, msg, net_services, peer);  
174 - message_receiver(queue, net_services, peer)  
175 - }.  
176 -  
177 -define Server -> (RWStream) -> One  
178 - net_services_handler  
179 - (  
180 - List(NetService) net_services,  
181 - ) =  
182 - (Server server) |-> (RWStream conn) |->  
183 - if remote_IP_address_and_port(conn) is (num_peer,_) then  
184 - //convert IP address of the client to string  
185 - with peer = ip_addr_to_string(num_peer),  
186 - //println("NET SERVICES Accepting connection with "+peer);  
187 -  
188 - //now managing the list of SERVICES  
189 - with queue = create_MessageQueue("CXM Net Services"),  
190 - message_transceiver(conn, queue);  
191 - message_receiver(queue, net_services, peer).  
192 -  
193 -  
194 -public define Maybe(Server)  
195 - start_net_services  
196 - (  
197 - List(NetService) net_services,  
198 - Word32 network_port,  
199 - )=  
200 - if start_server(0,  
201 - network_port,  
202 - net_services_handler(net_services),  
203 - (One u) |-> unique) is  
204 - {  
205 - cannot_create_the_socket then println("Cannot create the listening socket."); failure,  
206 - cannot_bind_to_port then println("Cannot bind to port " + network_port ); failure,  
207 - cannot_listen_on_port then println("Cannot listen on port " + network_port); failure,  
208 - ok(server) then  
209 - println("Net services started on port " + network_port);  
210 - println("------ Available services ------");  
211 - print_services(net_services);  
212 - success(server)  
213 - }. 1 +/*
  2 + *
  3 + * User: David RENE
  4 + * Date: 25/04/2007
  5 + * Time: 11:01
  6 + * (c) Calexium
  7 + *
  8 + */
  9 +read tools/basis.anubis
  10 +read system/convert.anubis
  11 +read system/string.anubis
  12 +read system/muscle.anubis
  13 +read system/data_io.anubis
  14 +read system/message_queue.anubis
  15 +read system/message_transceiver.anubis
  16 +read CXM_generic_protocol.anubis
  17 +read calexium_lib/CXM_message_constants.anubis
  18 +
  19 +public type NetService:
  20 + net_service(
  21 + Word32 version,
  22 + Word32 id,
  23 + String name,
  24 + List(String) domains,
  25 + (MessageQueue, String, String) -> One handler // Parameters are MessageQueue, peer IP and timestamp string
  26 + ).
  27 +
  28 +define One
  29 + print_services
  30 + (
  31 + List(NetService) net_services
  32 + ) =
  33 +
  34 + map_forget((NetService net_s)|->
  35 + if net_s is net_service(version, id, name, domains, _) then
  36 + println(" id : 0x"+ to_hexa(id));
  37 + println(" version : " + to_String(version));
  38 + println(" name : "+ name );
  39 + println(" domains : ");
  40 + map_forget((String domain) |-> println(" : "+domain), domains);
  41 + println("----------------------------------------")
  42 + ,net_services).
  43 +
  44 + /** Try to find the service_id in services_list. If the service is found in that list
  45 + * the corresponding NetService object is return
  46 + */
  47 +define Maybe(NetService)
  48 + find_service
  49 + (
  50 + List(NetService) services_list,
  51 + Word32 service_id,
  52 + Word32 service_version,
  53 + String domain
  54 + )=
  55 + if services_list is
  56 + {
  57 + [] then failure,
  58 + [h . t] then
  59 + if h.id = service_id & h.version >=+ service_version then
  60 + if domain = "" then
  61 + success(h)
  62 + else if domain:h.domains then //this writing (a:b) means, is a belonging to b where b is list of type a
  63 + success(h)
  64 + else
  65 + find_service(t, service_id, service_version, domain)
  66 + else
  67 + find_service(t, service_id, service_version, domain)
  68 + }.
  69 +
  70 + /** Check if the muscle message msg has the correct fields for requesting a net_services
  71 + * if we found "service" and "version" fields on the message, we try to find if the service
  72 + * referenced in "service" is available in net_services list
  73 + */
  74 +define Maybe(NetService)
  75 + has_service
  76 + (
  77 + MessageQueue queue,
  78 + Message msg,
  79 + List(NetService) net_services
  80 + )=
  81 + if find_int32(msg, "SERVICE") is
  82 + {
  83 + failure then //send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE); failure,
  84 + // old names... should be removed soon
  85 + if find_int32(msg, "service") is
  86 + {
  87 + failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE); failure,
  88 + success(service_id) then
  89 + if find_int32(msg, "version") is
  90 + {
  91 + failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE);failure,
  92 + success(service_version) then
  93 + //if DOMAIN field exists, this mean we want to target only this domain
  94 + if find_string(msg, "DOMAIN") is
  95 + {
  96 + failure then find_service(net_services, service_id, service_version,""),
  97 + success(domain) then find_service(net_services, service_id, service_version, domain)
  98 + }
  99 + }
  100 + }
  101 +
  102 + success(service_id) then
  103 + if find_int32(msg, "VERSION") is
  104 + {
  105 + failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE);failure,
  106 + success(service_version) then
  107 + //if DOMAIN field exists, this mean we want to target only this domain
  108 + if find_string(msg, "DOMAIN") is
  109 + {
  110 + failure then find_service(net_services, service_id, service_version,""),
  111 + success(domain) then find_service(net_services, service_id, service_version, domain)
  112 + }
  113 + }
  114 + }.
  115 +
  116 +define String
  117 + get_time_stamp
  118 + =
  119 + with time = (UTime) unow,
  120 + "<"+virtual_machine_id+"@"+time.seconds+">".
  121 +
  122 + /** This message_received function just handle the negociation process the available net_services.
  123 + * In other words, it only recognize the _CXM_REQUEST_FOR_SERVICE message and try to launch the
  124 + * corresponding servcice
  125 + */
  126 +
  127 +define One
  128 + service_negociation
  129 + (
  130 + MessageQueue queue,
  131 + Message msg,
  132 + List(NetService) net_services,
  133 + String peer
  134 + )=
  135 + //println("Service NEGOCIATION [" + to_hexa(*msg.what) + "] received");
  136 + if * msg.what = _CXM_REQUEST_FOR_SERVICE then
  137 + if has_service(queue, msg, net_services) is
  138 + {
  139 + failure then
  140 + send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE, _CXM_UNKNOW_SERVICE, "Unknown service")
  141 + success(net_service) then
  142 + with result = message(0),
  143 + timestamp = get_time_stamp,
  144 + forget(add_string(result, "TIMESTAMP", timestamp));
  145 + send_ACK_ok(queue, _CXM_REQUEST_FOR_SERVICE, result);
  146 + net_service.handler(queue, peer, timestamp)
  147 + }
  148 + else
  149 + send_ACK_error(queue, *msg.what, _CXM_UNKNOW_CMD, "Unknown command [" + (*msg.what) + "]")
  150 + .
  151 +
  152 + /**
  153 + * this function unflatten muscle message and give the correct message to service_negociation function
  154 + */
  155 +public define One
  156 + message_receiver
  157 + (
  158 + MessageQueue queue,
  159 + List(NetService) net_services,
  160 + String peer
  161 + ) =
  162 + if queue.quit_requested(unique) then
  163 + unique
  164 + else
  165 + //println("PRE SERVICE message_receiver "+"["+virtual_machine_id + "]");
  166 + if queue.get_next_received_Message(1) is
  167 + {
  168 + timeout then //println("PRE timeout");
  169 + message_receiver(queue, net_services, peer),
  170 + closed then //println("PRE closed");
  171 + unique,
  172 + msg(msg) then unique; //println("PRE negociation");
  173 + service_negociation(queue, msg, net_services, peer);
  174 + message_receiver(queue, net_services, peer)
  175 + }.
  176 +
  177 +define Server -> (RWStream) -> One
  178 + net_services_handler
  179 + (
  180 + List(NetService) net_services,
  181 + ) =
  182 + (Server server) |-> (RWStream conn) |->
  183 + if remote_IP_address_and_port(conn) is (num_peer,_) then
  184 + //convert IP address of the client to string
  185 + with peer = ip_addr_to_string(num_peer),
  186 + //println("NET SERVICES Accepting connection with "+peer);
  187 +
  188 + //now managing the list of SERVICES
  189 + with queue = create_MessageQueue("CXM Net Services", tcp(conn)),
  190 + message_transceiver(/*conn,*/ queue);
  191 + message_receiver(queue, net_services, peer).
  192 +
  193 +
  194 +public define Maybe(Server)
  195 + start_net_services
  196 + (
  197 + List(NetService) net_services,
  198 + Word32 network_port,
  199 + )=
  200 + if start_server(0,
  201 + network_port,
  202 + net_services_handler(net_services),
  203 + (One u) |-> unique) is
  204 + {
  205 + cannot_create_the_socket then println("Cannot create the listening socket."); failure,
  206 + cannot_bind_to_port then println("Cannot bind to port " + network_port ); failure,
  207 + cannot_listen_on_port then println("Cannot listen on port " + network_port); failure,
  208 + ok(server) then
  209 + println("Net services started on port " + network_port);
  210 + println("------ Available services ------");
  211 + print_services(net_services);
  212 + success(server)
  213 + }.
net_services_protocols/ftp_client.anubis
  Show/Hide comments   View file @8dac565
@@ -154,8 +154,8 @@ public define Maybe(One) @@ -154,8 +154,8 @@ public define Maybe(One)
154 { 154 {
155 error(_) then println("can't connect to ftp server"); failure, 155 error(_) then println("can't connect to ftp server"); failure,
156 ok(conn) then 156 ok(conn) then
157 - with queue = create_MessageQueue("ftp_get_file"),  
158 - message_transceiver(conn, queue); 157 + with queue = create_MessageQueue("ftp_get_file", tcp(conn)),
  158 + message_transceiver(/*conn,*/ queue);
159 println("request for ftp service"); 159 println("request for ftp service");
160 if request_for_service(queue) then 160 if request_for_service(queue) then
161 with result = get_file(queue, remote_file, local_file, ftp), 161 with result = get_file(queue, remote_file, local_file, ftp),