tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r381687 - in /tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes: ./ demos/ group/ group/interceptors/ io/ mcast/ tcp/
Date Tue, 28 Feb 2006 16:14:26 GMT
Author: fhanik
Date: Tue Feb 28 08:14:20 2006
New Revision: 381687

URL: http://svn.apache.org/viewcvs?rev=381687&view=rev
Log:
Added order protocol, created a command line channel creator

Added:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
Removed:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.jbx
Modified:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelMessage.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/MembershipService.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMembership.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastService.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/AsyncSocketSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/FastAsyncSocketSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/PooledSocketSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelMessage.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelMessage.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelMessage.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelMessage.java
Tue Feb 28 08:14:20 2006
@@ -16,6 +16,7 @@
 package org.apache.catalina.tribes;
 
 import java.io.Serializable;
+import org.apache.catalina.tribes.io.XByteBuffer;
 
 /**
  * @author Filip Hanik
@@ -61,11 +62,13 @@
      */
     public byte[] getUniqueId();
     
-    public void setMessage(byte[] data);
+    public void setMessage(XByteBuffer buf);
     
-    public byte[] getMessage();
+    public XByteBuffer getMessage();
     
     public int getOptions();
     public void setOptions(int options);
+    
+    public ChannelMessage clone();
 
 }

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/MembershipService.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/MembershipService.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/MembershipService.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/MembershipService.java
Tue Feb 28 08:14:20 2006
@@ -69,6 +69,8 @@
      */
     public boolean hasMembers();
     
+    
+    public Member getMember(Member mbr);
     /**
      * Returns a list of all the members in the cluster.
      */

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
Tue Feb 28 08:14:20 2006
@@ -1,3 +1,18 @@
+/*
+ * Copyright 1999,2004-2006 The Apache Software Foundation.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.catalina.tribes.demos;
 
 import org.apache.catalina.tribes.Member;
@@ -288,20 +303,15 @@
         System.out.println("Usage:\n\t"+
                            "java LoadTest [options]\n\t"+
                            "Options:\n\t\t"+
-                           "[-bind tcpbindaddress] \n\t\t"+
-                           "[-port tcplistenport]  \n\t\t"+
-                           "[-mbind multicastbindaddr]  \n\t\t"+
                            "[-mode receive|send|both]  \n\t\t"+
                            "[-debug]  \n\t\t"+
                            "[-count messagecount]  \n\t\t"+
                            "[-stats statinterval]  \n\t\t"+
-                           "[-ack true|false]  \n\t\t"+
-                           "[-sync true|false]  \n\t\t"+
-                           "[-gzip]  \n\t\t"+
                            "[-pause nrofsecondstopausebetweensends]  \n\t\t"+
-                           "[-sender pooled|fastasyncqueue]  \n\t\t"+
                            "[-threads numberofsenderthreads]  \n\t\t"+
                            "[-break (halts execution on exception)]\n"+
+                           "\tChannel options:"+
+                           ChannelCreator.usage()+"\n\n"+
                            "Example:\n\t"+
                            "java LoadTest -port 4004\n\t"+
                            "java LoadTest -bind 192.168.0.45 -port 4005\n\t"+
@@ -309,49 +319,28 @@
     }
     
     public static void main(String[] args) throws Exception {
-        String bind  = "auto";
-        int port = 4001;
         boolean send = true;
-        String mbind = null;
         boolean debug = false;
-        boolean ack = true;
-        boolean sync = true;
-        boolean gzip = false;
         long pause = 0;
         int count = 1000000;
         int stats = 10000;
         boolean breakOnEx = false;
         int threads = 1;
-        String sender = "pooled";
         if ( args.length == 0 ) {
             args = new String[] {"-help"};
         }
         for (int i = 0; i < args.length; i++) {
-            if ("-bind".equals(args[i])) {
-                bind = args[++i];
-            } else if ("-sender".equals(args[i])) {
-                sender = args[++i];
-            } else if ("-port".equals(args[i])) {
-                port = Integer.parseInt(args[++i]);
-            } else if ("-threads".equals(args[i])) {
+            if ("-threads".equals(args[i])) {
                 threads = Integer.parseInt(args[++i]);
             } else if ("-count".equals(args[i])) {
                 count = Integer.parseInt(args[++i]);
             } else if ("-pause".equals(args[i])) {
                 pause = Long.parseLong(args[++i])*1000;
-            } else if ("-gzip".equals(args[i])) {
-                gzip = true;
             } else if ("-break".equals(args[i])) {
                 breakOnEx = true;
-            } else if ("-ack".equals(args[i])) {
-                ack = Boolean.parseBoolean(args[++i]);
-            } else if ("-sync".equals(args[i])) {
-                sync = Boolean.parseBoolean(args[++i]);
             } else if ("-stats".equals(args[i])) {
                 stats = Integer.parseInt(args[++i]);
                 System.out.println("Stats every "+stats+" message");
-            } else if ("-mbind".equals(args[i])) {
-                mbind = args[++i];
             } else if ("-mode".equals(args[i])) {
                 if ( "receive".equals(args[++i]) ) send = false;
             } else if ("-debug".equals(args[i])) {
@@ -364,34 +353,7 @@
         }
         
         
-        ReplicationListener rl = new ReplicationListener();
-        rl.setTcpListenAddress(bind);
-        rl.setTcpListenPort(port);
-        rl.setTcpSelectorTimeout(100);
-        rl.setTcpThreadCount(4);
-        rl.getBind();
-        rl.setSendAck(ack);
-        rl.setSynchronized(sync);
-
-        ReplicationTransmitter ps = new ReplicationTransmitter();
-        ps.setReplicationMode(sender);
-        ps.setAckTimeout(15000);
-        ps.setAutoConnect(true);
-        ps.setWaitForAck(ack);
-
-        McastService service = new McastService();
-        service.setMcastAddr("228.0.0.5");
-        if ( mbind != null ) service.setMcastBindAddress(mbind);
-        service.setMcastFrequency(500);
-        service.setMcastDropTime(2000);
-        service.setMcastPort(45565);
-
-        ManagedChannel channel = new GroupChannel();
-        channel.setChannelReceiver(rl);
-        channel.setChannelSender(ps);
-        channel.setMembershipService(service);
-        
-        if ( gzip ) channel.addInterceptor(new GzipInterceptor());
+        ManagedChannel channel = (ManagedChannel)ChannelCreator.createChannel(args);
         
         LoadTest test = new LoadTest(channel,send,count,debug,pause,stats,breakOnEx);
         LoadMessage msg = new LoadMessage();

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java
Tue Feb 28 08:14:20 2006
@@ -22,9 +22,7 @@
 import org.apache.catalina.tribes.ChannelSender;
 import org.apache.catalina.tribes.ChannelReceiver;
 import org.apache.catalina.tribes.Channel;
-import java.io.IOException;
 import org.apache.catalina.tribes.InterceptorPayload;
-import org.apache.catalina.tribes.io.ClusterData;
 import org.apache.catalina.tribes.MessageListener;
 
 

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
Tue Feb 28 08:14:20 2006
@@ -103,8 +103,9 @@
                 b = XByteBuffer.serialize(msg);
             }
             data.setOptions(options);
-            
-            data.setMessage(b);
+            XByteBuffer buffer = new XByteBuffer(b.length+128);
+            buffer.append(b,0,b.length);
+            data.setMessage(buffer);
             getFirstInterceptor().sendMessage(destination, data, null);
         }catch ( Exception x ) {
             if ( x instanceof ChannelException ) throw (ChannelException)x;
@@ -118,9 +119,9 @@
             
             Serializable fwd = null;
             if ( (msg.getOptions() & BYTE_MESSAGE) == BYTE_MESSAGE ) {
-                fwd = new ByteMessage(msg.getMessage());
+                fwd = new ByteMessage(msg.getMessage().getBytes());
             } else {
-                fwd = XByteBuffer.deserialize(msg.getMessage());
+                fwd = XByteBuffer.deserialize(msg.getMessage().getBytes());
             }
             if ( channelListener != null && channelListener.accept(fwd,msg.getAddress()))

                 channelListener.messageReceived(fwd,msg.getAddress());

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
Tue Feb 28 08:14:20 2006
@@ -41,7 +41,9 @@
     
     public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload
payload) throws ChannelException {
         try {
-            msg.setMessage(compress(msg.getMessage()));
+            byte[] data = compress(msg.getMessage().getBytes());
+            msg.getMessage().trim(msg.getMessage().getLength());
+            msg.getMessage().append(data,0,data.length);
             getNext().sendMessage(destination, msg, payload);
         } catch ( IOException x ) {
             log.error("Unable to compress byte contents");
@@ -51,7 +53,9 @@
 
     public void messageReceived(ChannelMessage msg) {
         try {
-            msg.setMessage(decompress(msg.getMessage()));
+            byte[] data = decompress(msg.getMessage().getBytes());
+            msg.getMessage().trim(msg.getMessage().getLength());
+            msg.getMessage().append(data,0,data.length);
             getPrevious().messageReceived(msg);
         } catch ( IOException x ) {
             log.error("Unable to decompress byte contents",x);

Added: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java?rev=381687&view=auto
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
(added)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
Tue Feb 28 08:14:20 2006
@@ -0,0 +1,231 @@
+/*
+ * Copyright 1999,2004 The Apache Software Foundation.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ */
+
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.HashMap;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.InterceptorPayload;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.io.XByteBuffer;
+
+
+
+/**
+ *
+ * 
+ * @author Filip Hanik
+ * @version 1.0
+ */
+public class OrderInterceptor extends ChannelInterceptorBase {
+    private HashMap outcounter = new HashMap();
+    private HashMap incounter = new HashMap();
+    private HashMap incoming = new HashMap();
+    private long expire = 3000;
+    private boolean forwardExpired;
+
+    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload
payload) throws ChannelException {
+        for ( int i=0; i<destination.length; i++ ) {
+            ChannelMessage tmp = msg.clone();
+            int nr = incCounter(destination[i]);
+            tmp.getMessage().append(XByteBuffer.toBytes(nr),0,4);
+            getNext().sendMessage(new Member[] {destination[i]}, tmp, payload);
+        }
+    }
+
+    public void messageReceived(ChannelMessage msg) {
+        int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
+        msg.getMessage().trim(4);
+        MessageOrder order = new MessageOrder(msgnr,msg);
+        processIncoming(order);
+        //getPrevious().messageReceived(msg);
+    }
+    
+    public synchronized void processIncoming(MessageOrder order) {
+        Member member = order.getMessage().getAddress();
+        Counter cnt = getInCounter(member);
+        
+        MessageOrder tmp = (MessageOrder)incoming.get(member);
+        if ( tmp != null ) {
+            order = MessageOrder.add(tmp,order);
+        }
+        while ( (order!=null) && (order.getMsgNr() == cnt.getCounter()) ) {
+            //we are right on target. process orders
+            cnt.inc();
+            super.messageReceived(order.getMessage());
+            order.setMessage(null);
+            order = order.next;
+        }
+        MessageOrder head = order;
+        MessageOrder prev = null;
+        tmp = order;
+        while ( tmp != null ) {
+            //process expired messages
+            if ( tmp.isExpired(expire) ) {
+                //reset the head
+                if ( tmp == head ) head = tmp.next;
+                if ( getForwardExpired() ) super.messageReceived(tmp.getMessage());
+                tmp.setMessage(null);
+                tmp = tmp.next;
+                if ( prev != null ) prev.next = tmp;
+            } else {
+                prev = tmp;
+            }
+        }
+        if ( head == null ) incoming.remove(member);
+        else incoming.put(member,head);
+    }
+    
+    public void memberAdded(Member member) {
+        //notify upwards
+        getInCounter(member);
+        getOutCounter(member);
+        super.memberAdded(member);
+    }
+
+    public void memberDisappeared(Member member) {
+        //notify upwards
+        outcounter.remove(member);
+        incounter.remove(member);
+        super.memberDisappeared(member);
+    }
+    
+    public int incCounter(Member mbr) { 
+        Counter cnt = getOutCounter(mbr);
+        return cnt.inc();
+    }
+    
+    public synchronized Counter getInCounter(Member mbr) {
+        Counter cnt = (Counter)incounter.get(mbr);
+        if ( cnt == null ) {
+            cnt = new Counter();
+            cnt.inc(); //always start at 1 for incoming
+            incounter.put(mbr,cnt);
+        }
+        return cnt;
+    }
+
+    public synchronized Counter getOutCounter(Member mbr) {
+        Counter cnt = (Counter)outcounter.get(mbr);
+        if ( cnt == null ) {
+            cnt = new Counter();
+            outcounter.put(mbr,cnt);
+        }
+        return cnt;
+    }
+
+    public static class Counter {
+        private int value = 0;
+        
+        public int getCounter() {
+            return value;
+        }
+        
+        public synchronized int inc() {
+            return ++value;
+        }
+    }
+    
+    public static class MessageOrder {
+        private long received = System.currentTimeMillis();
+        private MessageOrder next;
+        private int msgNr;
+        private ChannelMessage msg = null;
+        public MessageOrder(int msgNr,ChannelMessage msg) {
+            this.msgNr = msgNr;
+            this.msg = msg;
+        }
+        
+        public boolean isExpired(long expireTime) {
+            return (System.currentTimeMillis()-received) > expireTime;
+        }
+        
+        public ChannelMessage getMessage() {
+            return msg;
+        }
+        
+        public void setMessage(ChannelMessage msg) {
+            this.msg = msg;
+        }
+        
+        public void setNext(MessageOrder order) {
+            this.next = order;
+        }
+        public MessageOrder getNext() {
+            return next;
+        }
+        
+        public static MessageOrder add(MessageOrder head, MessageOrder add) {
+            if ( head == null ) return add;
+            if ( add == null ) return head;
+            if ( head == add ) return add;
+
+            if ( head.getMsgNr() > add.getMsgNr() ) {
+                //add before
+                MessageOrder tmp = add.next;
+                add.next = head;
+                head.next = tmp;
+                return add;
+            }
+            
+            MessageOrder iter = head;
+            MessageOrder prev = null;
+            while ( iter.getMsgNr() < add.getMsgNr() && (iter.next !=null ) )
{
+                prev = iter;
+                iter = iter.next;
+            }
+            if ( iter.getMsgNr() < add.getMsgNr() ) {
+                //add after
+                add.next = iter.next;
+                iter.next = add;
+            } else if (iter.getMsgNr() > add.getMsgNr()) {
+                //add before
+                prev.next = add;
+                add.next = iter;
+                
+            } else {
+                throw new ArithmeticException("Message added has the same counter, synchronization
bug. Disable the order interceptor");
+            }
+            return head;
+        }
+        
+        public int getMsgNr() {
+            return msgNr;
+        }
+        
+        
+        
+    }
+
+    public void setExpire(long expire) {
+        this.expire = expire;
+    }
+
+    public void setForwardExpired(boolean forwardExpired) {
+        this.forwardExpired = forwardExpired;
+    }
+
+    public long getExpire() {
+        return expire;
+    }
+
+    public boolean getForwardExpired() {
+        return forwardExpired;
+    }
+
+}

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
Tue Feb 28 08:14:20 2006
@@ -39,7 +39,7 @@
 public class ClusterData implements ChannelMessage {
 
     private int options = 0 ;
-    private byte[] message ;
+    private XByteBuffer message ;
     private long timestamp ;
     private byte[] uniqueId ;
     private Member address;
@@ -60,7 +60,7 @@
      * @param message message data
      * @param timestamp message creation date
      */
-    public ClusterData(byte[] uniqueId, byte[] message, long timestamp) {
+    public ClusterData(byte[] uniqueId, XByteBuffer message, long timestamp) {
         this.uniqueId = uniqueId;
         this.message = message;
         this.timestamp = timestamp;
@@ -69,13 +69,13 @@
     /**
      * @return Returns the message.
      */
-    public byte[] getMessage() {
+    public XByteBuffer getMessage() {
         return message;
     }
     /**
      * @param message The message to set.
      */
-    public void setMessage(byte[] message) {
+    public void setMessage(XByteBuffer message) {
         this.message = message;
     }
     /**
@@ -150,7 +150,7 @@
 
      * @return byte[]
      */
-    public byte[] getDataPackage() throws IOException {
+    public byte[] getDataPackage()  {
         byte[] addr = ((McastMember)address).getData();
         int length = 
             4 + //options
@@ -160,7 +160,7 @@
             4 + //addr length off=12+uniqueId.length+4
             addr.length+ //member data off=12+uniqueId.length+4+add.length
             4 + //message length off=12+uniqueId.length+4+add.length+4
-            message.length;
+            message.getLength();
         byte[] data = new byte[length];
         int offset = 0;
         XByteBuffer.toBytes(options,data,offset);
@@ -175,14 +175,14 @@
         offset += 4; //addr.length
         System.arraycopy(addr,0,data,offset,addr.length);
         offset += addr.length; //addr data
-        XByteBuffer.toBytes(message.length,data,offset);
+        XByteBuffer.toBytes(message.getLength(),data,offset);
         offset += 4; //message.length
-        System.arraycopy(message,0,data,offset,message.length);
-        offset += message.length; //message data
+        System.arraycopy(message.getBytesDirect(),0,data,offset,message.getLength());
+        offset += message.getLength(); //message data
         return data;
     }
     
-    public static ClusterData getDataFromPackage(byte[] b) throws IOException {
+    public static ClusterData getDataFromPackage(byte[] b)  {
         ClusterData data = new ClusterData(false);
         int offset = 0;
         data.setOptions(XByteBuffer.toInt(b,offset));
@@ -198,30 +198,26 @@
         System.arraycopy(b,offset,addr,0,addr.length);
         data.setAddress(McastMember.getMember(addr));
         offset += addr.length; //addr data
-        data.message = new byte[XByteBuffer.toInt(b,offset)];
+        data.message = new XByteBuffer(new byte[XByteBuffer.toInt(b,offset)]);
         offset += 4; //message length
-        System.arraycopy(b,offset,data.message,0,data.message.length);
-        offset += data.message.length; //message data
+        System.arraycopy(b,offset,data.message.getBytesDirect(),0,data.message.getLength());
+        offset += data.message.getLength(); //message data
         return data;
     }
     
-    public static void main(String[] args) throws Exception {
-        ClusterData data1 = new ClusterData();
-        data1.setAddress(new McastMember("domain","127.0.0.1",1000,System.currentTimeMillis()));
-        data1.setMessage(new byte[1024]);
-        
-        byte[] b = data1.getDataPackage();
-        
-        ClusterData data2 = ClusterData.getDataFromPackage(b);
-        
-        if ( !(data1.getAddress().equals(data2.getAddress())) ||
-             !(Arrays.equals(data1.getMessage(),data2.getMessage())) ||
-             !(Arrays.equals(data1.getUniqueId(),data2.getUniqueId())) ||
-             !(data1.getTimestamp() == data2.timestamp) ||
-             !(data1.getOptions() == data2.getOptions() ) ) {
-            throw new Exception("Not Equal");
-        }
-                         
+    public int hashCode() {
+        return XByteBuffer.toInt(getUniqueId(),0);
+    }
+    
+    public boolean equals(Object o) {
+        if ( o instanceof ClusterData ) {
+            return Arrays.equals(getUniqueId(),((ClusterData)o).getUniqueId());
+        } else return false;
+    }
+    
+    public ClusterData clone() {
+        byte[] d = this.getDataPackage();
+        return ClusterData.getDataFromPackage(d);
     }
     
 }

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
Tue Feb 28 08:14:20 2006
@@ -91,6 +91,18 @@
     public XByteBuffer(int size) {
         buf = new byte[size];
     }
+    
+    public XByteBuffer(byte[] data) {
+        this(data,data.length+128);
+    }
+    
+    public XByteBuffer(byte[] data, int size) {
+        int length = Math.max(data.length,size);
+        buf = new byte[length];
+        System.arraycopy(data,0,buf,0,data.length);
+        bufSize = data.length;
+    }
+
 
     /**
      * Constructs a new XByteBuffer with an initial size of 1024 bytes
@@ -98,6 +110,20 @@
     public XByteBuffer()  {
         this(DEF_SIZE);
     }
+    
+    public int getLength() {
+        return bufSize;
+    }
+    
+    public void trim(int length) {
+        if ( (bufSize - length) < 0 ) 
+            throw new ArrayIndexOutOfBoundsException("Can't trim more bytes than are available.
length:"+bufSize+" trim:"+length);
+        bufSize -= length;
+    }
+            
+    public byte[] getBytesDirect() {
+        return this.buf;
+    }
 
     /**
      * Returns the bytes in the buffer, in its exact length
@@ -415,11 +441,6 @@
         ByteArrayOutputStream outs = new ByteArrayOutputStream();
         ObjectOutputStream out = new ObjectOutputStream(outs);
         out.writeObject(msg);
-        // flush out the gzip stream to byte buffer
-//        if(out != null) {
-//            out.flush();
-//            out.close();
-//        }
         byte[] data = outs.toByteArray();
         return data;
     }

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
Tue Feb 28 08:14:20 2006
@@ -135,7 +135,7 @@
      * @return - the bytes for this member deserialized
      * @throws Exception
      */
-    public byte[] getData() throws IOException {
+    public byte[] getData()  {
         //package looks like
         //alive - 8 bytes
         //port - 4 bytes

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMembership.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMembership.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMembership.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMembership.java
Tue Feb 28 08:14:20 2006
@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
+import org.apache.catalina.tribes.Member;
 
 /**
  * A <b>membership</b> implementation using simple multicast.
@@ -190,6 +191,19 @@
      */
     public synchronized boolean hasMembers() {
         return members.length > 0 ;
+    }
+    
+    
+    public synchronized McastMember getMember(Member mbr) {
+        if(hasMembers()) {
+            McastMember result = null;
+            for ( int i=0; i<this.members.length && result==null; i++ ) {
+                if ( members[i].equals(mbr) ) result = members[i];
+            }//for
+            return result;
+        } else {
+            return null;
+        }
     }
  
     /**

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastService.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastService.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastService.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastService.java
Tue Feb 28 08:14:20 2006
@@ -352,6 +352,11 @@
        if ( impl == null || impl.membership == null ) return false;
        return impl.membership.hasMembers();
     }
+    
+    public Member getMember(Member mbr) {
+        if ( impl == null || impl.membership == null ) return null;
+        return impl.membership.getMember(mbr);
+    }
 
     /**
      * Return all the members

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/AsyncSocketSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/AsyncSocketSender.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/AsyncSocketSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/AsyncSocketSender.java
Tue Feb 28 08:14:20 2006
@@ -166,12 +166,12 @@
         synchronized (this) {
             inQueueCounter++;
             if(queueThread != null)
-                queueThread.incQueuedNrOfBytes(data.getMessage().length);
+                queueThread.incQueuedNrOfBytes(data.getMessage().getLength());
        }
         if (log.isTraceEnabled())
             log.trace(sm.getString("AsyncSocketSender.queue.message",
                     getAddress().getHostAddress(), new Integer(getPort()), data.getUniqueId(),
new Long(
-                            data.getMessage().length)));
+                            data.getMessage().getLength())));
     }
 
     /*
@@ -268,7 +268,7 @@
                     int messagesize = 0;
                     try {
                         ChannelMessage data = (ChannelMessage) entry.getValue();
-                        messagesize = data.getMessage().length;
+                        messagesize = data.getMessage().getLength();
                         sender.pushMessage(data);
                     } catch (Exception x) {
                         log.warn(sm.getString("AsyncSocketSender.send.error",

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
Tue Feb 28 08:14:20 2006
@@ -827,8 +827,8 @@
             checkKeepAlive();
             if(doProcessingStats) addProcessingStats(time);
             if(messageTransfered) {
-                addStats(data.getMessage().length);
-                if (log.isTraceEnabled()) log.trace(sm.getString("IDataSender.send.message",
address.getHostAddress(),new Integer(port), data.getUniqueId(), new Long(data.getMessage().length)));
+                addStats(data.getMessage().getLength());
+                if (log.isTraceEnabled()) log.trace(sm.getString("IDataSender.send.message",
address.getHostAddress(),new Integer(port), data.getUniqueId(), new Long(data.getMessage().getLength())));
             } else {
                 dataFailureCounter++;
                 if ( exception != null ) throw exception;

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/FastAsyncSocketSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/FastAsyncSocketSender.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/FastAsyncSocketSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/FastAsyncSocketSender.java
Tue Feb 28 08:14:20 2006
@@ -319,12 +319,12 @@
         synchronized (this) {
             inQueueCounter++;
             if(queueThread != null)
-                queueThread.incQueuedNrOfBytes(data.getMessage().length);
+                queueThread.incQueuedNrOfBytes(data.getMessage().getLength());
         }
         if (log.isTraceEnabled())
             log.trace(sm.getString("AsyncSocketSender.queue.message",
                     getAddress().getHostAddress(), new Integer(getPort()), data.getUniqueId(),
new Long(
-                            data.getMessage().length)));
+                            data.getMessage().getLength())));
     }
 
     /**
@@ -480,7 +480,7 @@
                 int messagesize = 0;
                 try {
                     ChannelMessage data = (ChannelMessage) entry.data();
-                    messagesize = data.getMessage().length;
+                    messagesize = data.getMessage().getLength();
                     sender.pushMessage(data);
                 } catch (Exception x) {
                     log.warn(sm.getString(

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/PooledSocketSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/PooledSocketSender.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/PooledSocketSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/PooledSocketSender.java
Tue Feb 28 08:14:20 2006
@@ -132,7 +132,7 @@
             //return the connection to the pool
             senderQueue.returnSender(sender);
         }
-        addStats(data.getMessage().length);
+        addStats(data.getMessage().getLength());
     }
 
     public String toString() {

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java?rev=381687&r1=381686&r2=381687&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java
Tue Feb 28 08:14:20 2006
@@ -602,7 +602,7 @@
             }
             sender.sendMessage(data);
             sender.setSuspect(false);
-            addStats(data.getMessage().length);
+            addStats(data.getMessage().getLength());
         } catch (IOException x) {
             if (!sender.getSuspect()) {
                 if (log.isErrorEnabled() ) log.error("Unable to send replicated message,
is member ["+sender.toString()+"] down?",x);



---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org


Mime
View raw message