CXM_net_services.anubis
7.57 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
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
/*
*
* 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 system/logger.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, (LogLevel, String) -> One ) -> One handler // Parameters are MessageQueue, peer IP and timestamp string, logger
)
.
define String
dump_services
(
List(NetService) 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
*/
define Maybe(NetService)
find_service
(
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,
[h . t] then
if h.id = service_id & h.version >=+ service_version then
if domain = "" then
success(h)
else if domain:h.domains then //this writing (a:b) means, is a belonging to b where b is list of type a
success(h)
else
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
* 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
//if DOMAIN field exists, this mean we want to target only this domain
if find_string(msg, "DOMAIN") is
{
failure then find_service(net_services, service_id, service_version,""),
success(domain) then find_service(net_services, service_id, service_version, domain)
}
}
}
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
//if DOMAIN field exists, this mean we want to target only this domain
if find_string(msg, "DOMAIN") is
{
failure then find_service(net_services, service_id, service_version,""),
success(domain) then find_service(net_services, service_id, service_version, domain)
}
}
}
.
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,
(LogLevel, String) -> One logger
)=
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, logger)
}
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
*/
public define One
message_receiver
(
MessageQueue queue,
List(NetService) net_services,
String peer,
(LogLevel, String) -> One logger
) =
if queue.quit_requested(unique) then
unique
else
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, 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, logger);
message_receiver(queue, net_services, peer, logger)
}.
define Server -> (RWStream) -> One
net_services_handler
(
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),
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(queue);
message_receiver(queue, net_services, peer, logger).
public define Maybe(Server)
start_net_services
(
List(NetService) net_services,
Word32 network_port,
(LogLevel, String) -> One logger
)=
if start_server(0,
network_port,
net_services_handler(net_services, logger),
(One u) |-> unique) is
{
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
logger(logInfo, "Net services started on port " + network_port);
logger(logInfo, "------ Available services ------");
logger(logInfo, dump_services(net_services));
success(server)
}
.