java: fix streaming de/serializer

java-plan1/Unpacker.java

@@ -50,13 +50,26 @@ public class Unpacker {
 		return finished;
 	}
 
+	public void reset()
+	{
+		for(int i=0; i <= top; ++top) {
+			stack_ct[top] = 0;
+			stack_count[top] = 0;
+			stack_obj[top] = null;
+			stack_map_key[top] = null;
+		}
+		cs = CS_HEADER;
+		trail = 0;
+		top = -1;
+		finished = false;
+	}
 
 	@SuppressWarnings("unchecked")
 	public int execute(byte[] src, int off, int length) throws IOException
 	{
 		if(off >= length) { return off; }
 
-		int limit = off + length;
+		int limit = length;
 		int i = off;
 		int count;
 
@@ -64,7 +77,7 @@ public class Unpacker {
 
 		_out: do {
 		_header_again: {
-			//System.out.println("while i:"+i);
+			//System.out.println("while i:"+i+" limit:"+limit);
 
 			int b = src[i];
 
@@ -349,6 +362,7 @@ public class Unpacker {
 					((ArrayList)(stack_obj[top])).add(obj);
 					if(--stack_count[top] == 0) {
 						obj = stack_obj[top];
+						stack_obj[top] = null;
 						--top;
 						break _push;
 					}
@@ -363,6 +377,8 @@ public class Unpacker {
 					((HashMap)(stack_obj[top])).put(stack_map_key[top], obj);
 					if(--stack_count[top] == 0) {
 						obj = stack_obj[top];
+						stack_map_key[top] = null;
+						stack_obj[top] = null;
 						--top;
 						break _push;
 					}
@@ -379,7 +395,7 @@ public class Unpacker {
 		++i;
 		} while(i < limit);  // _out
 
-		return i - off;
+		return i;
 	}
 }
 

java-plan1/test.java

@@ -6,21 +6,88 @@ class OpenByteArrayOutputStream extends ByteArrayOutputStream {
 	byte[] getBuffer() { return buf; }
 }
 
+
 public class test {
+
 	public static void main(String[] args) throws IOException
+	{
+		testSimple();
+		testStreaming();
+	}
+
+
+	public static void testSimple() throws IOException
 	{
 		OpenByteArrayOutputStream out = new OpenByteArrayOutputStream();
 
 		Packer pk = new Packer(out);
 		pk.packArray(3)
-			.packInt(0)
+			.packInt(1)
-			.packByte((byte)1)
+			.packByte((byte)2)
-			.packDouble(0.1);
+			.packDouble(0.3);
 
 		Unpacker pac = new Unpacker();
 		int nlen = pac.execute(out.getBuffer(), 0, out.getCount());
+
 		if(pac.isFinished()) {
-			System.out.println(pac.getData());
+			System.out.println("testSimple: "+pac.getData());
+		}
+	}
+
+
+	public static void testStreaming() throws IOException
+	{
+		OpenByteArrayOutputStream out = new OpenByteArrayOutputStream();
+
+		////
+		// sender
+		//
+		// initialize the streaming serializer
+		Packer pk = new Packer(out);
+
+		// serialize 2 objects
+		pk.packArray(3)
+			.packInt(0)
+			.packByte((byte)1)
+			.packDouble(0.2);
+		pk.packArray(3)
+			.packInt(3)
+			.packByte((byte)4)
+			.packDouble(0.5);
+
+		// send it through the network
+		InputStream sock = new ByteArrayInputStream(out.getBuffer(), 0, out.getCount());
+
+
+		////
+		// receiver
+		//
+		// initialize the streaming deserializer
+		Unpacker pac = new Unpacker();
+		int parsed = 0;
+
+		byte[] buf = new byte[1024];
+		int buflen = 0;
+
+		while(true) {
+			// receive data from the network
+			int c = sock.read();
+			if(c < 0) { return; }
+
+			buf[buflen++] = (byte)c;
+
+			// deserialize
+			parsed = pac.execute(buf, parsed, buflen);
+			if(pac.isFinished()) {
+				// get an object
+				Object msg = pac.getData();
+				System.out.println("testStreaming: "+msg);
+
+				// reset the streaming deserializer
+				pac.reset();
+				buflen = 0;
+				parsed = 0;
+			}
 		}
 	}
 }