update web/CXM_multihost_http_server to manage timeout in read function

add FTP constant definition change net_services functions that now use MessageQueue instead of RWStream socket

update web/CXM_multihost_http_server to manage timeout in read function
add FTP constant definition change net_services functions that now use MessageQueue instead of RWStream socket
David RENE
1 parent 75c12334
Showing 5 changed files with 48 additions and 40 deletions Show diff stats
calexium_lib/CXM_message_constants.anubis
calexium_lib/net_services/CXM_generic_protocol.anubis
calexium_lib/net_services/CXM_net_services.anubis
calexium_lib/net_services_protocols/pkg_updater_protocol.anubis
calexium_lib/web/CXM_multihost_http_server.anubis
@@ -39,3 +39,7 @@ public define Int32 _CXM_ACK_RESULT_MSG             = 0x33771002. //
 public define Int32 _CXM_FTP_SERVICE_MSG_BASIS      = 0x33771400.   //
 public define Int32 _CXM_FTP_SERVICE_ID             = 0x33771401.
+public define Int32 _CXM_FTP_GET_FILE               = 0x33771402.
+public define Int32 _CXM_FTP_START_TRANSFERT        = 0x33771403.
+public define Int32 _CXM_FTP_DATA                   = 0x33771404.
+
@@ -10,7 +10,7 @@
 read system/muscle.anubis
 read system/data_io.anubis
 read tools/basis.anubis
-
+read system/message_queue.anubis
 read calexium_lib/CXM_message_constants.anubis
 public define Int32 _CXM_ERROR      = 0.
@@ -20,31 +20,30 @@ public define Int32 _CXM_UNKNOW_CMD = 2.
 public define One
   send_ACK_error
   (
-    RWStream  conn,
+    MessageQueue  queue,
     Int32     cmd_id
   )=
   with err_msg = message(_CXM_ACK),
   forget(add_int32(err_msg, "CMD", cmd_id));
   forget(add_int32(err_msg, "STATUS", _CXM_ERROR));
-  forget(send_message_by_Stream(conn, err_msg))
-  .
+  forget(queue.add_Message_to_send(err_msg)).
 public define One
   send_ACK_ok
   (
-    RWStream  conn,
-    Int32     cmd_id
+    MessageQueue  queue,
+    Int32         cmd_id
   )=
   with ok_msg = message(_CXM_ACK),
   forget(add_int32(ok_msg, "CMD", cmd_id));
   forget(add_int32(ok_msg, "STATUS", _CXM_OK));
-  forget(send_message_by_Stream(conn, ok_msg))
+  forget(queue.add_Message_to_send(ok_msg))
   .
 public define One
   send_ACK_ok
   (
-    RWStream  conn,
+    MessageQueue  queue,
     Int32     cmd_id,
     Message   result
   )=
@@ -52,24 +51,24 @@ public define One
   forget(add_int32(ok_msg, "CMD", cmd_id));
   forget(add_int32(ok_msg, "STATUS", _CXM_OK));
   forget(add_message(ok_msg, "RESULT", result));
-  forget(send_message_by_Stream(conn, ok_msg))
+  forget(queue.add_Message_to_send(ok_msg))
   .
 public define One
   send_result
   (
-    RWStream        conn,
+    MessageQueue    queue,
     Int32           cmd_id,
     Maybe(Message)  mb_msg
   )=
   if mb_msg is
   {
-    failure       then send_ACK_error(conn, cmd_id),
-    success(msg)  then send_ACK_ok(conn, cmd_id, msg)
+    failure       then send_ACK_error(queue, cmd_id),
+    success(msg)  then send_ACK_ok(queue, cmd_id, msg)
   }
   .
-public define Maybe(Message)
+ public define Maybe(Message)
   wait_for_reply
   (
     RWStream conn
@@ -11,7 +11,8 @@ 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
@@ -20,7 +21,7 @@ public type NetService:
     Int32   version,
     Int32   id,
     String  name,
-    (RWStream) -> One handler
+    (MessageQueue) -> One handler
     ).
 define One
@@ -62,17 +63,17 @@ define Maybe(NetService)
 define Maybe(NetService)
   has_service
   (
-    RWStream  conn,
-    Message   msg,
+    MessageQueue      queue,
+    Message           msg,
     List(NetService)  net_services
   )=
   if find_int32(msg, "service") is
   {
-    failure             then send_ACK_error(conn, _CXM_REQUEST_FOR_SERVICE); failure,
+    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(conn, _CXM_REQUEST_FOR_SERVICE);failure,
+      failure                  then send_ACK_error(queue, _CXM_REQUEST_FOR_SERVICE);failure,
       success(service_version) then find_service(net_services, service_id, service_version)
     }
   }.
@@ -85,18 +86,18 @@ define Maybe(NetService)
 define One 
   service_negociation
   (
-    RWStream  conn,
-    Message   msg,
+    MessageQueue      queue,
+    Message           msg,
     List(NetService)  net_services
   )=
   print("MESSAGE [" + to_ascii(*msg.what) + "] received\n");
   if * msg.what = _CXM_REQUEST_FOR_SERVICE then
-    if has_service(conn, msg, net_services) is
+    if has_service(queue, msg, net_services) is
     {
       failure                   then unique,
       success(net_service)  then
-        send_ACK_ok(conn, _CXM_REQUEST_FOR_SERVICE);
-        net_service.handler(conn)
+        send_ACK_ok(queue, _CXM_REQUEST_FOR_SERVICE);
+        net_service.handler(queue)
     }
   else
     unique
@@ -110,19 +111,17 @@ define One
 define One   
   message_receiver
   (
-    RWStream conn,
+    MessageQueue      queue,
     List(NetService)  net_services
   ) =
-  if unflatten_message(make_data_io(weaken(conn))) is
+  if queue.get_next_received_Message(10) is
   {
-    failure       then 
-      //print("can't unflatten any message\n");
-      sleep(1);
-      message_receiver(conn, net_services),
-    success(msg)  then 
-      service_negociation(conn, msg, net_services);
-      message_receiver(conn, net_services)
+    timeout   then message_receiver(queue, net_services),
+    closed    then unique,
+    msg(msg)  then 
+      service_negociation(queue, msg, net_services);
+      message_receiver(queue, net_services)
   }.
 define Server -> (RWStream) -> One
@@ -137,7 +136,9 @@ define Server -&gt; (RWStream) -&gt; One
   // print("NET SERVICES Accepting connection with "+peer+"\n"); 
   //now managing the list of SERVICES
-  message_receiver(conn, net_services).
+  with queue = create_MessageQueue,
+  message_transceiver(conn, queue);
+  message_receiver(queue, net_services).
 public define Maybe(Server)
@@ -7,5 +7,6 @@
  *
  */
+
 public define Int32 no_update         = 0.
 public define Int32 update_available  = 1.
@@ -696,9 +696,10 @@ define Result(Error,Word8)
    //if now > dead_line then record_dubious_connection(connection,dead_line,dos) else
    if read(connection,1,600) is // the connection is closed after 10 minutes of inactivity
      {
-       failure      then error(cannot_read_from_connection),
+       error   then error(cannot_read_from_connection),
+       timeout then error(timeout(600)),
                          //record_dubious_connection(connection,dead_line,dos), 
-       success(ba)  then if nth(0,ba) is 
+       ok(ba)  then if nth(0,ba) is 
          {
            failure then error(cannot_read_from_connection),
            success(c) then ok(c)
@@ -1631,8 +1632,9 @@ define Result(Error,ByteArray)
    if retries =< 0 then error(cannot_read_from_connection) else
    if read(connection,body_size,60) is 
      {
-       failure then error(cannot_read_from_connection),
-       success(new_bytes) then with
+       error    then error(cannot_read_from_connection),
+       timeout  then error(timeout(60)),
+       ok(new_bytes) then with
         ba        = so_far + new_bytes,                     // contains all the bytes read so far
         nr        = length(ba),                             // total read since the beginning
         nn        = length(new_bytes),                      // number of bytes just read
@@ -1881,8 +1883,9 @@ define One
    if sent >= size then unique else
    if read(file,min(10000,size-sent),60) is 
      {
-       failure then log_journal_msg(desc,"Cannot read from file '"+filename+"'.\n"), 
-       success(ba) then 
+       error    then log_journal_msg(desc,"Cannot read from file '"+filename+"'.\n"),
+       timeout  then log_journal_msg(desc,"Cannot read from file timeoput'"+filename+"'.\n"),
+       ok(ba)   then 
          with nr = length(ba),                          // get the number of bytes read
          if reliable_write(connection,ba) is 
            {