ensure the send of the entire message throught the network

David René
1 parent 8a8a9890
Showing 2 changed files with 29 additions and 6 deletions Show diff stats
anubis_dev/library/system/message_transceiver.anubis
anubis_dev/library/system/muscle.anubis
@@ -44,6 +44,7 @@ define DataResult
       ok(bytes) then 
         with len_fragment = length(bytes),
              new_buffer = current + bytes,
+//             println("fragment "+len_fragment + " total : "+length(new_buffer));
              receive_data(conn, len_left - len_fragment, new_buffer)
     }
   .
@@ -71,10 +72,10 @@ define One
   else
   with buffer_len = length(current_buffer),
   if buffer_len = 8 then
-    //println("buffer len 8 bytes [" + to_hexa(current_buffer));
+//    println("buffer len 8 bytes [" + to_hexa(current_buffer));
     if is_header(current_buffer) then
       with message_len = lendian_to_host_Int32(extract(current_buffer, 0, 4)),
-    //  println("Message length "+message_len);
+//      println("Message length "+message_len);
       if receive_data(conn, message_len, constant_byte_array(0,0)) is
       {
         failure then 
@@ -97,7 +98,7 @@ define One
         //TODO manage the timeout
         message_receiver(conn, queue, current_buffer),
       ok(bytes) then 
-        //println("receiver bytes " + to_hexa(bytes));
+//        println("receiver bytes " + to_hexa(bytes));
         with new_buffer = current_buffer + bytes,
         message_receiver(conn, queue, new_buffer)
     }
@@ -1580,6 +1580,28 @@ public define Message_Send_Result
 		    }
 		}.
  
+public define Maybe(Int32)
+  send_buffer
+  (
+    RWStream  conn,
+    ByteArray buffer,
+    Int32     left_to_send,
+    Int32     so_far
+  )=
+  if left_to_send = 0 then
+    success(so_far)
+  else
+    if write(weaken(conn), buffer) is
+    {
+      failure     then failure,
+      success(len)then 
+      if left_to_send = len then
+        success(so_far+len)
+      else
+        send_buffer(conn, extract(buffer, len, length(buffer)), left_to_send - len, so_far+len)
+    }
+  .
+  
 public define Message_Send_Result
   send_message_by_Stream
   (
@@ -1593,15 +1615,15 @@ public define Message_Send_Result
     with header = host_to_lendian_Int32_ByteArray(length(buffer)) + 
 		              host_to_lendian_Int32_ByteArray(_MUSCLE_MESSAGE_ENCODING_DEFAULT),
     //send the header. that header is construct as follow (size of the following message + header mark)
-    if write(weaken(conn),header) is
+    if send_buffer(conn,header, length(header), 0) is
     {
       failure    then network_error,
 		  success(_) then 
 		  //now we send the real message
-      if write(weaken(conn), buffer) is
+      if send_buffer(conn, buffer, length(buffer), 0) is
       {
     	  failure     then network_error,
-    	  success(_)  then ok
+    	  success(len)then ok
       }
     }
   }.