CXM_net_services.anubis
6.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
/*
*
* User: David RENE
* Date: 25/04/2007
* Time: 11:01
* (c) Calexium
*
*/
read tools/basis.anubis
read system/convert.anubis
read system/string.anubis
read system/muscle.anubis
read system/data_io.anubis
read system/message_queue.anubis
read system/message_transceiver.anubis
read CXM_generic_protocol.anubis
read calexium_lib/CXM_message_constants.anubis
public type NetService:
net_service(
Word32 version,
Word32 id,
String name,
List(String) domains,
(MessageQueue, String, String) -> One handler // Parameters are MessageQueue, peer IP and timestamp string
).
define One
print_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).
/** Try to find the service_id in services_list. If the service is found in that list
* the corresponding NetService object is return
*/
define Maybe(NetService)
find_service
(
List(NetService) services_list,
Word32 service_id,
Word32 service_version
)=
if services_list is
{
[] then failure,
[h . t] then
if h.id = service_id & h.version >=+ service_version then
success(h)
else
find_service(t, service_id, service_version)
}.
/** 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
* referenced in "service" is available in net_services list
*/
define Maybe(NetService)
has_service
(
MessageQueue queue,
Message msg,
List(NetService) net_services
)=
if find_int32(msg, "SERVICE") is
{
failure then //send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE); failure,
// old names... should be removed soon
if find_int32(msg, "service") is
{
failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE); failure,
success(service_id) then
if find_int32(msg, "version") is
{
failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE);failure,
success(service_version) then find_service(net_services, service_id, service_version)
}
}
success(service_id) then
if find_int32(msg, "VERSION") is
{
failure then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE);failure,
success(service_version) then find_service(net_services, service_id, service_version)
}
}.
define String
get_time_stamp
=
with time = (UTime) unow,
"<"+virtual_machine_id+"@"+time.seconds+">".
/** This message_received function just handle the negociation process the available net_services.
* In other words, it only recognize the _CXM_REQUEST_FOR_SERVICE message and try to launch the
* corresponding servcice
*/
define One
service_negociation
(
MessageQueue queue,
Message msg,
List(NetService) net_services,
String peer
)=
//println("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
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)
}
else
send_ACK_error(queue, *msg.what, _CXM_UNKNOW_CMD, "Unknown command [" + (*msg.what) + "]")
.
/**
* this function unflatten muscle message and give the correct message to service_negociation function
*/
define One
message_receiver
(
MessageQueue queue,
List(NetService) net_services,
String peer
) =
if queue.quit_requested(unique) then
unique
else
//println("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");
unique,
msg(msg) then unique; //println("PRE negociation");
service_negociation(queue, msg, net_services, peer);
message_receiver(queue, net_services, peer)
}.
define Server -> (RWStream) -> One
net_services_handler
(
List(NetService) net_services,
) =
(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);
//now managing the list of SERVICES
with queue = create_MessageQueue("CXM Net Services"),
message_transceiver(conn, queue);
message_receiver(queue, net_services, peer).
public define Maybe(Server)
start_net_services
(
List(NetService) net_services,
Word32 network_port,
)=
if start_server(0,
network_port,
net_services_handler(net_services),
(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,
ok(server) then
println("Net services started on port " + network_port);
println("------ Available services ------");
print_services(net_services);
success(server)
}.