activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r777209 - in /activemq/trunk: activemq-core/src/main/java/org/apache/activemq/store/kahadb/ activemq-core/src/main/java/org/apache/activemq/transport/tcp/ kahadb/src/main/java/org/apache/kahadb/index/ kahadb/src/main/java/org/apache/kahadb/...
Date Thu, 21 May 2009 18:34:46 GMT
Author: chirino
Date: Thu May 21 18:34:45 2009
New Revision: 777209

URL: http://svn.apache.org/viewvc?rev=777209&view=rev
Log:
Extended the Marshaller interface so that users can determin if uses fixed size records. 
Kahadb should be able to make some optimizations 
on fixed sized objects.


Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/TempMessageDatabase.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashBin.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashIndex.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/BytesMarshaller.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/IntegerMarshaller.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/LongMarshaller.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/Marshaller.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/ObjectMarshaller.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/SequenceSet.java
    activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/StringMarshaller.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
Thu May 21 18:34:45 2009
@@ -70,6 +70,7 @@
 import org.apache.kahadb.util.LongMarshaller;
 import org.apache.kahadb.util.Marshaller;
 import org.apache.kahadb.util.StringMarshaller;
+import org.apache.kahadb.util.VariableMarshaller;
 
 public class MessageDatabase {
 
@@ -121,11 +122,7 @@
         }
     }
 
-    class MetadataMarshaller implements Marshaller<Metadata> {
-        public Class<Metadata> getType() {
-            return Metadata.class;
-        }
-
+    class MetadataMarshaller extends VariableMarshaller<Metadata> {
         public Metadata readPayload(DataInput dataIn) throws IOException {
             Metadata rc = new Metadata();
             rc.read(dataIn);
@@ -937,13 +934,9 @@
         }
     }
     
-    static protected class MessageKeysMarshaller implements Marshaller<MessageKeys>
{
+    static protected class MessageKeysMarshaller extends VariableMarshaller<MessageKeys>
{
         static final MessageKeysMarshaller INSTANCE = new MessageKeysMarshaller();
         
-        public Class<MessageKeys> getType() {
-            return MessageKeys.class;
-        }
-
         public MessageKeys readPayload(DataInput dataIn) throws IOException {
             return new MessageKeys(dataIn.readUTF(), LocationMarshaller.INSTANCE.readPayload(dataIn));
         }
@@ -967,10 +960,7 @@
         TreeMap<Long, HashSet<String>> ackPositions;
     }
 
-    protected class StoredDestinationMarshaller implements Marshaller<StoredDestination>
{
-        public Class<StoredDestination> getType() {
-            return StoredDestination.class;
-        }
+    protected class StoredDestinationMarshaller extends VariableMarshaller<StoredDestination>
{
 
         public StoredDestination readPayload(DataInput dataIn) throws IOException {
             StoredDestination value = new StoredDestination();
@@ -1002,10 +992,6 @@
     static class LocationMarshaller implements Marshaller<Location> {
         final static LocationMarshaller INSTANCE = new LocationMarshaller();
 
-        public Class<Location> getType() {
-            return Location.class;
-        }
-
         public Location readPayload(DataInput dataIn) throws IOException {
             Location rc = new Location();
             rc.setDataFileId(dataIn.readInt());
@@ -1017,14 +1003,22 @@
             dataOut.writeInt(object.getDataFileId());
             dataOut.writeInt(object.getOffset());
         }
-    }
 
-    static class KahaSubscriptionCommandMarshaller implements Marshaller<KahaSubscriptionCommand>
{
-        final static KahaSubscriptionCommandMarshaller INSTANCE = new KahaSubscriptionCommandMarshaller();
+        public int getFixedSize() {
+            return 8;
+        }
+
+        public Location deepCopy(Location source) {
+            return new Location(source);
+        }
 
-        public Class<KahaSubscriptionCommand> getType() {
-            return KahaSubscriptionCommand.class;
+        public boolean isDeepCopySupported() {
+            return true;
         }
+    }
+
+    static class KahaSubscriptionCommandMarshaller extends VariableMarshaller<KahaSubscriptionCommand>
{
+        final static KahaSubscriptionCommandMarshaller INSTANCE = new KahaSubscriptionCommandMarshaller();
 
         public KahaSubscriptionCommand readPayload(DataInput dataIn) throws IOException {
             KahaSubscriptionCommand rc = new KahaSubscriptionCommand();

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/TempMessageDatabase.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/TempMessageDatabase.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/TempMessageDatabase.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/TempMessageDatabase.java
Thu May 21 18:34:45 2009
@@ -47,6 +47,7 @@
 import org.apache.kahadb.util.LongMarshaller;
 import org.apache.kahadb.util.Marshaller;
 import org.apache.kahadb.util.StringMarshaller;
+import org.apache.kahadb.util.VariableMarshaller;
 
 public class TempMessageDatabase {
 
@@ -369,13 +370,9 @@
         }
     }
     
-    static protected class MessageKeysMarshaller implements Marshaller<MessageRecord>
{
+    static protected class MessageKeysMarshaller extends VariableMarshaller<MessageRecord>
{
         static final MessageKeysMarshaller INSTANCE = new MessageKeysMarshaller();
         
-        public Class<MessageRecord> getType() {
-            return MessageRecord.class;
-        }
-
         public MessageRecord readPayload(DataInput dataIn) throws IOException {
             return new MessageRecord(dataIn.readUTF(), ByteSequenceMarshaller.INSTANCE.readPayload(dataIn));
         }
@@ -398,7 +395,7 @@
         TreeMap<Long, HashSet<String>> ackPositions;
     }
 
-    protected class StoredDestinationMarshaller implements Marshaller<StoredDestination>
{
+    protected class StoredDestinationMarshaller extends VariableMarshaller<StoredDestination>
{
         public Class<StoredDestination> getType() {
             return StoredDestination.class;
         }
@@ -428,13 +425,9 @@
         }
     }
 
-    static class ByteSequenceMarshaller implements Marshaller<ByteSequence> {
+    static class ByteSequenceMarshaller extends VariableMarshaller<ByteSequence> {
         final static ByteSequenceMarshaller INSTANCE = new ByteSequenceMarshaller();
 
-        public Class<ByteSequence> getType() {
-            return ByteSequence.class;
-        }
-
         public ByteSequence readPayload(DataInput dataIn) throws IOException {
         	byte data[] = new byte[dataIn.readInt()];
         	dataIn.readFully(data);
@@ -447,13 +440,9 @@
         }
     }
 
-    static class KahaSubscriptionCommandMarshaller implements Marshaller<KahaSubscriptionCommand>
{
+    static class KahaSubscriptionCommandMarshaller extends VariableMarshaller<KahaSubscriptionCommand>
{
         final static KahaSubscriptionCommandMarshaller INSTANCE = new KahaSubscriptionCommandMarshaller();
 
-        public Class<KahaSubscriptionCommand> getType() {
-            return KahaSubscriptionCommand.class;
-        }
-
         public KahaSubscriptionCommand readPayload(DataInput dataIn) throws IOException {
             KahaSubscriptionCommand rc = new KahaSubscriptionCommand();
             rc.mergeFramed((InputStream)dataIn);

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
Thu May 21 18:34:45 2009
@@ -29,7 +29,9 @@
 import java.net.UnknownHostException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -119,6 +121,9 @@
     private Boolean tcpNoDelay;
     private Thread runnerThread;
 
+    private final ArrayBlockingQueue<Object> outbound = new ArrayBlockingQueue<Object>(100);
+    private Thread onewayThread;
+
     /**
      * Connect to a remote Node - e.g. a Broker
      * 
@@ -157,14 +162,37 @@
         this.localLocation = null;
         setDaemon(true);
     }
-
+    
     /**
      * A one way asynchronous send
      */
     public void oneway(Object command) throws IOException {
         checkStarted();
-        wireFormat.marshal(command, dataOut);
-        dataOut.flush();
+        try {
+            outbound.put(command);
+        } catch (InterruptedException e) {
+            throw new InterruptedIOException();
+        }
+    }
+
+    protected void sendOneways() {
+        try {
+            while(!isStopped()) {
+                Object command = outbound.poll(500, TimeUnit.MILLISECONDS);
+                if( command!=null ) {
+                    try {
+                        while( command!=null ) {
+                            wireFormat.marshal(command, dataOut);
+                            command = outbound.poll();
+                        }
+                        dataOut.flush();
+                    } catch (IOException e) {
+                        getTransportListener().onException(e);
+                    }
+                }
+            }
+        } catch (InterruptedException e) {
+        }
     }
 
     /**
@@ -399,6 +427,11 @@
 
     protected void doStart() throws Exception {
         connect();
+        onewayThread = new Thread(null, new Runnable(){
+            public void run() {
+                sendOneways();
+            }}, "ActiveMQ Transport Sender: " + toString(), getStackSize());
+        onewayThread.start();
         stoppedLatch.set(new CountDownLatch(1));
         super.doStart();
     }
@@ -487,7 +520,11 @@
                     LOG.debug("Caught exception closing socket",e);
                 }
             }
-           
+        }
+        if( onewayThread!=null ) {
+            onewayThread.join();
+            onewayThread = null;
+            outbound.clear();
         }
     }
 

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/BTreeNode.java Thu May 21
18:34:45 2009
@@ -26,11 +26,10 @@
 import java.util.NoSuchElementException;
 import java.util.Map.Entry;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.apache.kahadb.index.BTreeIndex.Prefixer;
 import org.apache.kahadb.page.Page;
 import org.apache.kahadb.page.Transaction;
+import org.apache.kahadb.util.VariableMarshaller;
 
 
 /**
@@ -144,17 +143,13 @@
      * @param <Key>
      * @param <Value>
      */
-    static public class Marshaller<Key,Value> implements org.apache.kahadb.util.Marshaller<BTreeNode<Key,Value>>
{
+    static public class Marshaller<Key,Value> extends VariableMarshaller<BTreeNode<Key,Value>>
{
         private final BTreeIndex<Key,Value> index;
         
         public Marshaller(BTreeIndex<Key,Value> index) {
             this.index = index;
         }
 
-        public Class<BTreeNode<Key,Value>> getType() {
-            return null;
-        }
-        
         public void writePayload(BTreeNode<Key,Value> node, DataOutput os) throws IOException
{
             // Write the keys
             short count = (short)node.keys.length; // cast may truncate value...

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashBin.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashBin.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashBin.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashBin.java Thu May 21 18:34:45
2009
@@ -17,15 +17,14 @@
 package org.apache.kahadb.index;
 
 import java.io.DataInput;
-import java.io.DataInputStream;
 import java.io.DataOutput;
-import java.io.DataOutputStream;
 import java.io.IOException;
 import java.util.Map;
 import java.util.TreeMap;
 
 import org.apache.kahadb.page.Page;
 import org.apache.kahadb.page.Transaction;
+import org.apache.kahadb.util.VariableMarshaller;
 
 /**
  * Bin in a HashIndex
@@ -35,17 +34,13 @@
 class HashBin<Key, Value> {
     
     
-    static public class Marshaller<Key, Value> implements org.apache.kahadb.util.Marshaller<HashBin<Key,
Value>> {
+    static public class Marshaller<Key, Value> extends VariableMarshaller<HashBin<Key,
Value>> {
         private final HashIndex<Key, Value> hashIndex;
 
         public Marshaller(HashIndex<Key, Value> index) {
             this.hashIndex = index;
         }
         
-        public Class<HashBin<Key, Value>> getType() {
-            return null;
-        }
-
         public HashBin<Key, Value> readPayload(DataInput is) throws IOException {
             HashBin<Key, Value> bin = new HashBin<Key, Value>();
             int size = is.readInt();

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashIndex.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashIndex.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashIndex.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/index/HashIndex.java Thu May 21
18:34:45 2009
@@ -30,6 +30,7 @@
 import org.apache.kahadb.page.PageFile;
 import org.apache.kahadb.page.Transaction;
 import org.apache.kahadb.util.Marshaller;
+import org.apache.kahadb.util.VariableMarshaller;
 
 /**
  * BTree implementation
@@ -104,11 +105,7 @@
             os.writeInt(binsActive);
         }
         
-        static class Marshaller implements org.apache.kahadb.util.Marshaller<Metadata>
{
-            public Class<Metadata> getType() {
-                return Metadata.class;
-            }
-
+        static class Marshaller extends VariableMarshaller<Metadata> {
             public Metadata readPayload(DataInput dataIn) throws IOException {
                 Metadata rc = new Metadata();
                 rc.read(dataIn);

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/BytesMarshaller.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/BytesMarshaller.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/BytesMarshaller.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/BytesMarshaller.java Thu May
21 18:34:45 2009
@@ -25,35 +25,31 @@
  * 
  * @version $Revision$
  */
-public class BytesMarshaller implements Marshaller {
-    /**
-     * Write the payload of this entry to the RawContainer
-     * 
-     * @param object
-     * @param dataOut
-     * @throws IOException
-     */
-    public void writePayload(Object object, DataOutput dataOut) throws IOException {
-        byte[] data = (byte[])object;
+public class BytesMarshaller implements Marshaller<byte[]> {
+
+    public void writePayload(byte[] data, DataOutput dataOut) throws IOException {
         dataOut.writeInt(data.length);
         dataOut.write(data);
     }
 
-    /**
-     * Read the entry from the RawContainer
-     * 
-     * @param dataIn
-     * @return unmarshalled object
-     * @throws IOException
-     */
-    public Object readPayload(DataInput dataIn) throws IOException {
+    public byte[] readPayload(DataInput dataIn) throws IOException {
         int size = dataIn.readInt();
         byte[] data = new byte[size];
         dataIn.readFully(data);
         return data;
     }
     
-    public Class getType() {
-        return byte[].class;
+    public int getFixedSize() {
+        return -1;
+    }
+
+    public byte[] deepCopy(byte[] source) {
+        byte []rc = new byte[source.length];
+        System.arraycopy(source, 0, rc, 0, source.length);
+        return rc;
+    }
+
+    public boolean isDeepCopySupported() {
+        return true;
     }
 }

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/IntegerMarshaller.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/IntegerMarshaller.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/IntegerMarshaller.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/IntegerMarshaller.java Thu
May 21 18:34:45 2009
@@ -36,8 +36,20 @@
     public Integer readPayload(DataInput dataIn) throws IOException {
         return dataIn.readInt();
     }
+
+    public int getFixedSize() {
+        return 4;
+    }
+
     
-    public Class<Integer> getType() {
-        return Integer.class;
+    /** 
+     * @return the source object since integers are immutable. 
+     */
+    public Integer deepCopy(Integer source) {
+        return source;
+    }
+
+    public boolean isDeepCopySupported() {
+        return true;
     }
 }

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/LongMarshaller.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/LongMarshaller.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/LongMarshaller.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/LongMarshaller.java Thu May
21 18:34:45 2009
@@ -29,29 +29,23 @@
     
     public static final LongMarshaller INSTANCE = new LongMarshaller();
     
-    /**
-     * Write the payload of this entry to the RawContainer
-     * 
-     * @param object
-     * @param dataOut
-     * @throws IOException
-     */
     public void writePayload(Long object, DataOutput dataOut) throws IOException {
         dataOut.writeLong(object);
     }
 
-    /**
-     * Read the entry from the RawContainer
-     * 
-     * @param dataIn
-     * @return unmarshalled object
-     * @throws IOException
-     */
     public Long readPayload(DataInput dataIn) throws IOException {
         return dataIn.readLong();
     }
-    
-    public Class<Long> getType() {
-        return Long.class;
+
+    public int getFixedSize() {
+        return 8;
+    }
+
+    public Long deepCopy(Long source) {
+        return source;
+    }
+
+    public boolean isDeepCopySupported() {
+        return true;
     }
 }

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/Marshaller.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/Marshaller.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/Marshaller.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/Marshaller.java Thu May 21
18:34:45 2009
@@ -21,16 +21,13 @@
 import java.io.IOException;
 
 /**
- * Marshaller for marshalling in/out objects to a RawContainer
- * 
  * @version $Revision$
  */
 public interface Marshaller<T> {
     
-    Class<T> getType();
-       
     /**
-     * Write the payload of this entry to the RawContainer
+     * Write the payload of the object to the DataOutput stream.
+     * 
      * @param object 
      * @param dataOut
      * @throws IOException
@@ -39,12 +36,28 @@
     
     
     /**
-     * Read the entry from the RawContainer
+     * Read the payload of the object from the DataInput stream.
+     * 
      * @param dataIn 
      * @return unmarshalled object
      * @throws IOException
      */
     T readPayload(DataInput dataIn) throws IOException;
 
+    /** 
+     * @return -1 if the object do not always marshall to a fixed size, otherwise return
that fixed size.
+     */
+    int getFixedSize();
+    
+    /**
+     * 
+     * @return true if the {@link #deepCopy(Object)} operations is supported.
+     */
+    boolean isDeepCopySupported();
+
+    /**
+     * @return a deep copy of the source object.
+     */
+    T deepCopy(T source);
    
 }

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/ObjectMarshaller.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/ObjectMarshaller.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/ObjectMarshaller.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/ObjectMarshaller.java Thu May
21 18:34:45 2009
@@ -29,15 +29,8 @@
  * 
  * @version $Revision$
  */
-public class ObjectMarshaller implements Marshaller {
+public class ObjectMarshaller extends VariableMarshaller<Object> {
 
-    /**
-     * Write the payload of this entry to the RawContainer
-     * 
-     * @param object
-     * @param dataOut
-     * @throws IOException
-     */
     public void writePayload(Object object, DataOutput dataOut) throws IOException {
         ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
         ObjectOutputStream objectOut = new ObjectOutputStream(bytesOut);
@@ -48,13 +41,6 @@
         dataOut.write(data);
     }
 
-    /**
-     * Read the entry from the RawContainer
-     * 
-     * @param dataIn
-     * @return unmarshalled object
-     * @throws IOException
-     */
     public Object readPayload(DataInput dataIn) throws IOException {
         int size = dataIn.readInt();
         byte[] data = new byte[size];
@@ -68,7 +54,4 @@
         }
     }
     
-    public Class getType() {
-        return Object.class;
-    }
 }

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/SequenceSet.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/SequenceSet.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/SequenceSet.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/SequenceSet.java Thu May 21
18:34:45 2009
@@ -36,10 +36,6 @@
 
         public static final Marshaller INSTANCE = new Marshaller();
         
-        public Class<SequenceSet> getType() {
-            return SequenceSet.class;
-        }
-
         public SequenceSet readPayload(DataInput in) throws IOException {
             SequenceSet value = new SequenceSet();
             int count = in.readInt();
@@ -70,6 +66,24 @@
                 sequence = sequence.getNext();
             }
         }
+
+        public int getFixedSize() {
+            return -1;
+        }
+
+        public SequenceSet deepCopy(SequenceSet value) {
+            SequenceSet rc = new SequenceSet();
+            Sequence sequence = value.getHead();
+            while (sequence != null ) {
+                rc.add(new Sequence(sequence.first, sequence.last));
+                sequence = sequence.getNext();
+            }
+            return rc;
+        }
+
+        public boolean isDeepCopySupported() {
+            return true;
+        }
     }
     
     public void add(Sequence value) {

Modified: activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/StringMarshaller.java
URL: http://svn.apache.org/viewvc/activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/StringMarshaller.java?rev=777209&r1=777208&r2=777209&view=diff
==============================================================================
--- activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/StringMarshaller.java (original)
+++ activemq/trunk/kahadb/src/main/java/org/apache/kahadb/util/StringMarshaller.java Thu May
21 18:34:45 2009
@@ -50,8 +50,17 @@
     public String readPayload(DataInput dataIn) throws IOException {
         return dataIn.readUTF();
     }
-    
-    public Class<String> getType() {
-        return String.class;
+
+
+    public int getFixedSize() {
+        return -1;
+    }
+
+    public String deepCopy(String source) {
+        return source;
+    }
+
+    public boolean isDeepCopySupported() {
+        return true;
     }
 }



Mime
View raw message