[GNUnet-SVN] r18734 - in gnunet-java: . src/org/gnunet/exceptions src/org/gnunet/messages src/org/gnunet/util

[gnunet]

Author: dold
Date: 2011-12-21 11:19:09 +0100 (Wed, 21 Dec 2011)
New Revision: 18734

Modified:
   gnunet-java/ISSUES
   gnunet-java/src/org/gnunet/exceptions/InternalLogicError.java
   gnunet-java/src/org/gnunet/messages/MessageLoader.java
   gnunet-java/src/org/gnunet/util/Client.java
   gnunet-java/src/org/gnunet/util/Scheduler.java
Log:
implemented message loading

Modified: gnunet-java/ISSUES
===================================================================
--- gnunet-java/ISSUES  2011-12-21 10:08:21 UTC (rev 18733)
+++ gnunet-java/ISSUES  2011-12-21 10:19:09 UTC (rev 18734)
@@ -82,23 +82,31 @@
 
 ====
 
-(undocumented) issues with the Runabout: visit methods cannot access private 
properties/classes that are
-accessible from the scope of the visit method
-java.lang.IllegalAccessError: tried to access class 
org.gnunet.construct.Construct$ParserGenerator from class 
org.grothoff.GenCode00000003
-when Construct.ParserGenerator is declared private
+* (undocumented) issues with the Runabout: visit methods cannot access private 
properties/classes that are
+  accessible from the scope of the visit method
+  java.lang.IllegalAccessError: tried to access class 
org.gnunet.construct.Construct$ParserGenerator from class 
org.grothoff.GenCode00000003
+  when Construct.ParserGenerator is declared private
 
 
-Runabout.java line 373: string comparison with "==" instead of .equals
-Does this rely on string interning?
+* Runabout.java line 373: string comparison with "==" instead of .equals
+ * Does this rely on string interning?
 
 
-* Buffer sizes / buffer allocation in general
+* buffer sizes / buffer allocation / buffer management  in general
 
+* Should Construct switch to ByteBuffer?
+ * otherwise we may get performance trouble with direct buffers?
 
-why is there no receive_cancel in client.c (but in connection.c)
+* why is there no receive_cancel in client.c (but in connection.c)
 
+* what about the timeout in all the transmit calls: what is the timeout 
relative to? call of the function?
+  the beginning of trying to select() to send?
 
-notifyTransmitReady: how often do we know the exact size of the message we 
want to send, even if the message
-has not been constructed yet?
 
-when do we issue a transmit but cancel it?
+
+(* notifyTransmitReady: how often do we know the exact size of the message we 
want to send, even if the message
+  has not been constructed yet?)
+
+* when do we issue a transmit but cancel it (except in shutdown)
+
+

Modified: gnunet-java/src/org/gnunet/exceptions/InternalLogicError.java
===================================================================
--- gnunet-java/src/org/gnunet/exceptions/InternalLogicError.java       
2011-12-21 10:08:21 UTC (rev 18733)
+++ gnunet-java/src/org/gnunet/exceptions/InternalLogicError.java       
2011-12-21 10:19:09 UTC (rev 18734)
@@ -9,5 +9,8 @@
  */
 @SuppressWarnings("serial")
 public class InternalLogicError extends Error {
+    public InternalLogicError(String s) {
+        super(s);
+    }
 
 }

Modified: gnunet-java/src/org/gnunet/messages/MessageLoader.java
===================================================================
--- gnunet-java/src/org/gnunet/messages/MessageLoader.java      2011-12-21 
10:08:21 UTC (rev 18733)
+++ gnunet-java/src/org/gnunet/messages/MessageLoader.java      2011-12-21 
10:19:09 UTC (rev 18734)
@@ -23,6 +23,10 @@
 package org.gnunet.messages;
 
 
+import org.gnunet.construct.Construct;
+import org.gnunet.exceptions.InternalLogicError;
+import org.gnunet.exceptions.MessageFormatException;
+
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -31,7 +35,8 @@
 
 public class MessageLoader {
     private static HashMap<Integer, String> msgmap;
-    static void loadMessageMap() {
+
+    public static void loadMessageMap() {
         URL loc = MessageLoader.class.getResource("MsgMap.txt");
         if (loc == null) {
             throw new RuntimeException("could not load message map");
@@ -57,4 +62,22 @@
             throw new RuntimeException("could not read message map");
         }
     }
+
+    public static Message loadMessage(int type, byte[] data, int offset) {
+        String className = msgmap.get(type);
+        if (className == null) {
+            // todo: error message
+            throw new MessageFormatException();
+        }
+        
+        ClassLoader cl = MessageLoader.class.getClassLoader();
+        Class msgClass;
+        try {
+            msgClass = cl.loadClass(className);
+        } catch (ClassNotFoundException e) {
+            throw new InternalLogicError("message class not found in 
classpath");
+        }
+        
+        return Construct.parseAs(data, offset, msgClass);
+    }
 }

Modified: gnunet-java/src/org/gnunet/util/Client.java
===================================================================
--- gnunet-java/src/org/gnunet/util/Client.java 2011-12-21 10:08:21 UTC (rev 
18733)
+++ gnunet-java/src/org/gnunet/util/Client.java 2011-12-21 10:19:09 UTC (rev 
18734)
@@ -19,8 +19,11 @@
  */
 package org.gnunet.util;
 
+import org.gnunet.construct.Construct;
 import org.gnunet.exceptions.InterfaceViolationException;
 import org.gnunet.messages.Message;
+import org.gnunet.messages.MessageHeader;
+import org.gnunet.messages.MessageLoader;
 import org.gnunet.util.Scheduler.Task;
 
 import java.io.IOException;
@@ -35,6 +38,8 @@
  * Represents a connection to a service.
  */
 public class Client {
+    
+    final int HEADER_SIZE=4; // XXX: this is the wrong place for this
 
 
     public class TransmitHandle {
@@ -53,23 +58,60 @@
     private class ReceiveHelperTask implements Task {
 
         private MessageReceiver receiver;
+        private RelativeTime timeout;
+        private boolean headerProcessed = false;
+        private MessageHeader msgh;
 
-        public ReceiveHelperTask(MessageReceiver receiver) {
+        public ReceiveHelperTask(MessageReceiver receiver, RelativeTime 
timeout) {
+
             this.receiver = receiver;
+            this.timeout = timeout;
+            recvActive = true;
         }
 
+        public void dispatchMessage() {
+            receiver.process(MessageLoader.loadMessage(msgh.type, 
recvBuffer.array(), 0));
+        }
+
         @Override
         public void run(Context ctx) {
             if (ctx.reasons.contains(Scheduler.Reason.TIMEOUT)) {
                 receiver.handleTimeout();
             } else if (ctx.reasons.contains(Scheduler.Reason.READ_READY)) {
-                //chan.read();
-                // recvBuffer.
+                try {
+                    chan.read(recvBuffer);
+                } catch (IOException e) {
+                    throw new RuntimeException("read failed");
+                }
+                if (recvBuffer.remaining() == 0) {
+                    if (headerProcessed) {
+                        recvActive = false;
+                        dispatchMessage();
+                    } else {
+                        msgh = (MessageHeader) 
Construct.parseAs(recvBuffer.array(), 0, MessageHeader.class);
+                        headerProcessed = true;
+                        // XXX: is the message size with or without the header`
+                        if (msgh.size > HEADER_SIZE) {
+                            recvBuffer.limit(msgh.size);
+                            schedule();
+                        } else {
+                            recvActive = false;
+                            dispatchMessage();
+                        }
+                    }
+                } else {
+                    schedule();
+                }
+
             } else {
                 // XXX: what to do here?
-                throw new RuntimeException();
+                throw new RuntimeException("receive failed");
             }
         }
+
+        public void schedule() {
+            Scheduler.addRead(timeout, chan, this);
+        }
     }
 
     private class TransmitHelperTask implements Task {
@@ -87,12 +129,14 @@
 
 
     private final int INITIAL_BUFFER_SIZE=128;
-    private SocketChannel chan;
+    SocketChannel chan;
     boolean recvActive=false;
     ByteBuffer recvBuffer = ByteBuffer.allocateDirect(INITIAL_BUFFER_SIZE);
 
+
+    boolean transmitActive=false;
     Object transmitQueue;
-    Object transmitBuffer;
+    ByteBuffer transmitBuffer;
 
 
     /**
@@ -151,8 +195,10 @@
         if (recvActive) {
             throw new InterfaceViolationException("receive must not be called 
while receiving");
         }
-        ReceiveHelperTask task = new ReceiveHelperTask(receiver);
-        Scheduler.addRead(timeout, chan, task);
+        recvBuffer.reset();
+        recvBuffer.limit(HEADER_SIZE);
+        ReceiveHelperTask rh = new ReceiveHelperTask(receiver, timeout);
+        rh.schedule();
     }
 
     public TransmitHandle notifyTransmitReady(int size, RelativeTime timeout,

Modified: gnunet-java/src/org/gnunet/util/Scheduler.java
===================================================================
--- gnunet-java/src/org/gnunet/util/Scheduler.java      2011-12-21 10:08:21 UTC 
(rev 18733)
+++ gnunet-java/src/org/gnunet/util/Scheduler.java      2011-12-21 10:19:09 UTC 
(rev 18734)
@@ -269,7 +269,8 @@
 
     private static void queueReady(TaskIdentifier tid) {
         assert tid.priority != null;
-        ready.get(tid.priority.ordinal()).add(tid);
+        int idx = tid.priority.ordinal();
+        ready.get(idx).add(tid);
         ready_count++;
     }