rhizome over mdp now writes content for in-order packets, and

import gets triggered, but file hash currently doesn't match.

overlay_mdp_services.c

@@ -119,6 +119,9 @@ int overlay_mdp_service_rhizomerequest(overlay_mdp_frame *mdp)
 	  sqlite3_blob_read(blob,&reply.out.payload[1+16+8+8],
 			    blockBytes,0);	  
 	  reply.out.payload_length=1+16+8+8+blockBytes;
+
+	  // Mark terminal block if required
+	  if (blockOffset+blockBytes==blob_bytes) reply.out.payload[0]='T';
 	  // send packet
 	  overlay_mdp_dispatch(&reply,0 /* system generated */, NULL,0); 
 	} else break;
@@ -139,6 +142,7 @@ int overlay_mdp_service_rhizomeresponse(overlay_mdp_frame *mdp)
   int type=mdp->out.payload[0];
   switch (type) {
   case 'B': /* data block */
+  case 'T': /* terminal data block */
     {
       if (mdp->out.payload_length<(1+16+8+8+1)) RETURN(-1);
       unsigned char *bidprefix=&mdp->out.payload[1];
@@ -156,7 +160,7 @@ int overlay_mdp_service_rhizomeresponse(overlay_mdp_frame *mdp)
 	 a slot to capture this files as it is being requested
 	 by someone else.
       */
-      rhizome_received_content(bidprefix,version,offset,count,bytes);
+      rhizome_received_content(bidprefix,version,offset,count,bytes,type);
 
       RETURN(-1);
     }

rhizome.h

@@ -429,8 +429,9 @@ struct http_response {
   const char * body;
 };
 
-int rhizome_received_content(unsigned char *bid,uint64_t version, uint64_t offset,
+int rhizome_received_content(unsigned char *bidprefix,uint64_t version, 
-			     int count,unsigned char *bytes);
+			     uint64_t offset,int count,unsigned char *bytes,
+			     int type);
 int rhizome_server_set_response(rhizome_http_request *r, const struct http_response *h);
 int rhizome_server_free_http_request(rhizome_http_request *r);
 int rhizome_server_http_send_bytes(rhizome_http_request *r);

rhizome_fetch.c

@@ -1178,29 +1178,6 @@ void rhizome_fetch_write(struct rhizome_fetch_slot *slot)
   }
 }
 
-int rhizome_received_content(unsigned char *bidprefix,uint64_t version, uint64_t offset,
-			      int count,unsigned char *bytes)
-{
-  IN();
-  int i;
-  for(i=0;i<NQUEUES;i++) {
-    if (rhizome_fetch_queues[i].active.bidP) {
-      if (!bcmp(rhizome_fetch_queues[i].active.bid,bidprefix,
-		16))
-	{
-	  DEBUGF("This response matches slot 0x%p",
-		 rhizome_fetch_queues[i].active);
-	  RETURN(0);
-	}
-      else 
-	DEBUGF("Doesn't match this slot, because BIDs don't match: %s* vs %s",
-	       alloca_tohex(bidprefix,16),
-	       alloca_tohex_bid(rhizome_fetch_queues[i].active.bid));
-    }
-  }
-  RETURN(-1);
-}
-
 void rhizome_write_content(struct rhizome_fetch_slot *slot, char *buffer, int bytes)
 {
   if (bytes>(slot->file_len-slot->file_ofs))
@@ -1265,6 +1242,48 @@ void rhizome_write_content(struct rhizome_fetch_slot *slot, char *buffer, int by
   schedule(&slot->alarm);
 }
 
+int rhizome_received_content(unsigned char *bidprefix,uint64_t version, uint64_t offset,
+			     int count,unsigned char *bytes,int type)
+{
+  IN();
+  int i;
+  for(i=0;i<NQUEUES;i++) {
+    if (rhizome_fetch_queues[i].active.bidP) {
+      if (!bcmp(rhizome_fetch_queues[i].active.bid,bidprefix,
+		16))
+	{
+	  DEBUGF("This response matches slot 0x%p",
+		 rhizome_fetch_queues[i].active);
+	  struct rhizome_fetch_slot *slot=&rhizome_fetch_queues[i].active;
+	  if (slot->file_ofs==offset) {
+	    debug=DEBUG_RHIZOME_RX;
+	    /* We don't know the file length until we receive the last
+	       block. If it isn't the last block, lie, and claim the end of
+	       file is yet to come. */
+	    if (type=='T') slot->file_len=offset+count;
+	    else slot->file_len=offset+count+1;
+	    DEBUGF("Trying to write %d bytes @ %d (file len = %d)",
+		   count,(int)slot->file_ofs,(int)slot->file_len);
+	    rhizome_write_content(slot,(char *)bytes,count);
+	    debug=0;
+	    slot->mdpRXWindowStart=offset+count;
+	    // TODO: Shift bitmap
+	    RETURN(0);
+	  } else {
+	    // TODO: Implement out-of-order reception so that lost packets
+	    // don't cause wastage
+	  }
+	  RETURN(0);
+	}
+      else 
+	DEBUGF("Doesn't match this slot, because BIDs don't match: %s* vs %s",
+	       alloca_tohex(bidprefix,16),
+	       alloca_tohex_bid(rhizome_fetch_queues[i].active.bid));
+    }
+  }
+  RETURN(-1);
+}
+
 void rhizome_fetch_poll(struct sched_ent *alarm)
 {
   struct rhizome_fetch_slot *slot = (struct rhizome_fetch_slot *) alarm;