CXM_net_services.anubis
5.26 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
/*
*
* 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(
Int32 version,
Int32 id,
String name,
(MessageQueue) -> One handler
).
define One
print_services
(
List(NetService) net_services
) =
map_forget((NetService net_s)|->
println(" id : 0x"+ to_hexa(net_s.id));
println(" version : " + to_String(net_s.version));
println(" name : "+ net_s.name );
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,
Int32 service_id,
Int32 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,
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)
}
}.
/** 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
)=
//TODO DR add a real management of error
//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),
success(net_service) then
send_ACK_ok(queue, _CXM_REQUEST_FOR_SERVICE);
net_service.handler(queue)
}
else
unique
.
/**
* this function unflatten muscle message and give the correct message to service_negociation function
*/
define One
message_receiver
(
MessageQueue queue,
List(NetService) net_services
) =
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),
closed then //println("PRE closed");
unique,
msg(msg) then unique; //println("PRE negociation");
service_negociation(queue, msg, net_services);
message_receiver(queue, net_services)
}.
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).
public define Maybe(Server)
start_net_services
(
List(NetService) net_services,
Int32 network_port,
Var(Bool) shutdown_required
)=
//TODO change the port number in the real world
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)
}.