svn commit: r581056 - in /incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity: codec/AbstractDecoder.java transport/ChannelDelegate.java transport/Session.java

[rhs]

Author: rhs
Date: Mon Oct  1 13:10:52 2007
New Revision: 581056

URL: http://svn.apache.org/viewvc?rev=581056&view=rev
Log:
error checking for field table decode, actually close sessions on session 
close, and maintain cummulative mark in session

Modified:
    
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/codec/AbstractDecoder.java
    
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/ChannelDelegate.java
    
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java

Modified: 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/codec/AbstractDecoder.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/codec/AbstractDecoder.java?rev=581056&r1=581055&r2=581056&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/codec/AbstractDecoder.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/codec/AbstractDecoder.java
 Mon Oct  1 13:10:52 2007
@@ -208,7 +208,7 @@
         {
             String key = readShortstr();
             byte code = get();
-            Type t = Type.get(code);
+            Type t = getType(code);
             Object value = read(t);
             result.put(key, value);
         }
@@ -223,7 +223,7 @@
         while (count < start + size)
         {
             byte code = get();
-            Type t = Type.get(code);
+            Type t = getType(code);
             Object value = read(t);
             result.add(value);
         }
@@ -234,7 +234,7 @@
     {
         long size = readLong();
         byte code = get();
-        Type t = Type.get(code);
+        Type t = getType(code);
         long count = readLong();
 
         List<Object> result = new ArrayList<Object>();
@@ -244,6 +244,19 @@
             result.add(value);
         }
         return result;
+    }
+
+    private Type getType(byte code)
+    {
+        Type type = Type.get(code);
+        if (type == null)
+        {
+            throw new IllegalArgumentException("unknown code: " + code);
+        }
+        else
+        {
+            return type;
+        }
     }
 
     private long readSize(Type t)

Modified: 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/ChannelDelegate.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/ChannelDelegate.java?rev=581056&r1=581055&r2=581056&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/ChannelDelegate.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/ChannelDelegate.java
 Mon Oct  1 13:10:52 2007
@@ -48,4 +48,14 @@
                            attached.getDetachedLifetime());
     }
 
+    public @Override void sessionClosed(Channel channel, SessionClosed closed)
+    {
+        System.out.println("Session closed: [" + closed.getReplyCode() + "]" +
+                           closed.getReplyText());
+        channel.getSession().closed();
+        // XXX: should we remove the channel from the connection? It
+        // could have an external reference to it. Maybe we need a
+        // weak hash map in connection.
+    }
+
 }

Modified: 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java?rev=581056&r1=581055&r2=581056&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java
 Mon Oct  1 13:10:52 2007
@@ -82,6 +82,7 @@
 
     public void processed(Method command)
     {
+        System.out.printf("processed[%d]: %s\n", command.getId(), 
command.getClass());
         processed(command.getId());
     }
 
@@ -92,6 +93,7 @@
 
     public void processed(long lower, long upper)
     {
+
         processed(new Range(lower, upper));
     }
 
@@ -111,12 +113,24 @@
 
     void flushProcessed()
     {
+        long mark = -1;
+        boolean first = true;
+        RangeSet rest = new RangeSet();
         for (Range r: processed)
         {
             System.out.println("Completed Range [" + r.getLower() + "," + 
r.getUpper() +"]" );
+            if (first)
+            {
+                first = false;
+                mark = r.getUpper();
+            }
+            else
+            {
+                rest.add(r);
+            }
         }
         System.out.println("Notifying peer with execution complete");
-        executionComplete(0, processed);
+        executionComplete(mark, rest);
     }
 
     void syncPoint()