tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r382509 - in /tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp: DataSender.java bio/PooledSocketSender.java nio/ParallelNioSender.java
Date Thu, 02 Mar 2006 20:53:56 GMT
Author: fhanik
Date: Thu Mar  2 12:53:55 2006
New Revision: 382509

URL: http://svn.apache.org/viewcvs?rev=382509&view=rev
Log:
more refactoring

Modified:
    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/bio/PooledSocketSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ParallelNioSender.java

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=382509&r1=382508&r2=382509&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
Thu Mar  2 12:53:55 2006
@@ -32,5 +32,8 @@
 public interface DataSender {
     public void connect() throws ChannelException;
     public void disconnect();
+    public boolean isConnected();
+    public void setRxBufSize(int size);
+    public void setTxBufSize(int size);
 
 }

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/PooledSocketSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/PooledSocketSender.java?rev=382509&r1=382508&r2=382509&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/PooledSocketSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/PooledSocketSender.java
Thu Mar  2 12:53:55 2006
@@ -22,6 +22,9 @@
 
 import org.apache.catalina.tribes.ChannelMessage;
 import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.tcp.PooledSender;
+import org.apache.catalina.tribes.tcp.DataSender;
+import org.apache.catalina.tribes.tcp.SenderState;
 
 /**
  * Send cluster messages with a pool of sockets (25).
@@ -31,7 +34,7 @@
  * @version 1.2
  */
 
-public class PooledSocketSender extends SinglePointDataSender {
+public class PooledSocketSender extends PooledSender {
 
     private static org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory
             .getLog(org.apache.catalina.tribes.tcp.bio.PooledSocketSender.class);
@@ -44,9 +47,16 @@
     // ----------------------------------------------------- Instance Variables
 
     private int maxPoolSocketLimit = 25;
+    private String domain;
+    private InetAddress host;
+    private int port;
+    private SenderState senderState;
+    private int keepAliveMaxRequestCount;
+    private long keepAliveTimeout;
+    private long ackTimeout;
+    private boolean resend;
+    private boolean waitForAck;
 
-    private SenderQueue senderQueue = null;
-    
     //  ----------------------------------------------------- Constructor
 
    /**
@@ -54,9 +64,12 @@
     * @param host replication node tcp address
     * @param port replication node tcp port
     */
-    public PooledSocketSender(String domain,InetAddress host, int port) {
-        super(domain,host, port);
-        senderQueue = new SenderQueue(this, maxPoolSocketLimit);
+    public PooledSocketSender(String domain,InetAddress host, int port, int poolSize) {
+        super(poolSize);
+        this.host = host;
+        this.domain = domain;
+        this.port = port;
+        this.maxPoolSocketLimit = poolSize;
     }
    
     //  ----------------------------------------------------- Public Properties
@@ -72,36 +85,82 @@
 
     }
 
-    public void setMaxPoolSocketLimit(int limit) {
-        maxPoolSocketLimit = limit;
-        senderQueue.setLimit(limit);
+    public void setDomain(String domain) {
+        this.domain = domain;
     }
 
-    public int getMaxPoolSocketLimit() {
-        return maxPoolSocketLimit;
+    public void setHost(InetAddress host) {
+        this.host = host;
     }
 
-    public int getInPoolSize() {
-        return senderQueue.getInPoolSize();
+    public void setPort(int port) {
+        this.port = port;
     }
 
-    public int getInUsePoolSize() {
-        return senderQueue.getInUsePoolSize();
+    public void setSenderState(SenderState senderState) {
+        this.senderState = senderState;
     }
 
-    //  ----------------------------------------------------- Public Methode
+    public void setKeepAliveMaxRequestCount(int keepAliveMaxRequestCount) {
+        this.keepAliveMaxRequestCount = keepAliveMaxRequestCount;
+    }
+
+    public void setKeepAliveTimeout(long keepAliveTimeout) {
+        this.keepAliveTimeout = keepAliveTimeout;
+    }
+
+    public void setAckTimeout(long ackTimeout) {
+        this.ackTimeout = ackTimeout;
+    }
+
+    public void setResend(boolean resend) {
+        this.resend = resend;
+    }
+
+    public void setWaitForAck(boolean waitForAck) {
+        this.waitForAck = waitForAck;
+    }
+
+    public String getDomain() {
+        return domain;
+    }
+
+    public InetAddress getHost() {
+        return host;
+    }
 
-    public synchronized void connect() throws ChannelException {
-        //do nothing, happens in the socket sender itself
-        senderQueue.open();
-        setSocketConnected(true);
+    public int getPort() {
+        return port;
     }
 
-    public synchronized void disconnect() {
-        senderQueue.close();
-        setSocketConnected(false);
+    public SenderState getSenderState() {
+        return senderState;
     }
 
+    public int getKeepAliveMaxRequestCount() {
+        return keepAliveMaxRequestCount;
+    }
+
+    public long getKeepAliveTimeout() {
+        return keepAliveTimeout;
+    }
+
+    public long getAckTimeout() {
+        return ackTimeout;
+    }
+
+    public boolean isResend() {
+        return resend;
+    }
+
+    public boolean getWaitForAck() {
+        return waitForAck;
+    }
+
+    //  ----------------------------------------------------- Public Methode
+
+    
+
     /**
      * send message and use a pool of DataSenders
      * 
@@ -117,9 +176,9 @@
                     connect();
             }
         }
-        SinglePointDataSender sender = senderQueue.getSender(0);
+        SinglePointDataSender sender = (SinglePointDataSender)getSender();
         if (sender == null) {
-            log.warn(sm.getString("PoolSocketSender.noMoreSender", this.getAddress(), new
Integer(this.getPort())));
+            log.warn("Sender queue is empty. Can not send any messages.");
             return;
         }
         //send the message
@@ -127,141 +186,30 @@
             sender.sendMessage(data);
         } finally {
             //return the connection to the pool
-            senderQueue.returnSender(sender);
+            returnSender(sender);
         }
     }
 
     public String toString() {
         StringBuffer buf = new StringBuffer("PooledSocketSender[");
-        buf.append(getAddress()).append(":").append(getPort()).append("]");
+        buf.append(getHost()).append(":").append(getPort()).append("]");
         return buf.toString();
     }
 
-    //  ----------------------------------------------------- Inner Class
-
-    private class SenderQueue {
-        private int limit = 25;
-
-        PooledSocketSender parent = null;
-
-        private LinkedList queue = new LinkedList();
-
-        private LinkedList inuse = new LinkedList();
-
-        private Object mutex = new Object();
-
-        private boolean isOpen = true;
-
-        public SenderQueue(PooledSocketSender parent, int limit) {
-            this.limit = limit;
-            this.parent = parent;
-        }
-
-        /**
-         * @return Returns the limit.
-         */
-        public int getLimit() {
-            return limit;
-        }
-        /**
-         * @param limit The limit to set.
-         */
-        public void setLimit(int limit) {
-            this.limit = limit;
-        }
-        /**
-         * @return
-         */
-        public int getInUsePoolSize() {
-            return inuse.size();
-        }
-
-        /**
-         * @return
-         */
-        public int getInPoolSize() {
-            return queue.size();
-        }
-
-        public SinglePointDataSender getSender(long timeout) {
-            SinglePointDataSender sender = null;
-            long start = System.currentTimeMillis();
-            long delta = 0;
-            do {
-                synchronized (mutex) {
-                    if (!isOpen)
-                        throw new IllegalStateException(
-                                "Socket pool is closed.");
-                    if (queue.size() > 0) {
-                        sender = (SinglePointDataSender) queue.removeFirst();
-                    } else if (inuse.size() < limit) {
-                        sender = getNewDataSender();
-                    } else {
-                        try {
-                            mutex.wait(timeout);
-                        } catch (Exception x) {
-                            PooledSocketSender.log.warn(sm.getString("PoolSocketSender.senderQueue.sender.failed",parent.getAddress(),new
Integer(parent.getPort())),x);
-                        }//catch
-                    }//end if
-                    if (sender != null) {
-                        inuse.add(sender);
-                    }
-                }//synchronized
-                delta = System.currentTimeMillis() - start;
-            } while ((isOpen) && (sender == null)
-                    && (timeout == 0 ? true : (delta < timeout)));
-            //to do
-            return sender;
-        }
-
-        public void returnSender(SinglePointDataSender sender) {
-            //to do
-            synchronized (mutex) {
-                queue.add(sender);
-                inuse.remove(sender);
-                mutex.notify();
-            }
-        }
-
-        private SinglePointDataSender getNewDataSender() {
-            //new DataSender(
+    public DataSender getNewDataSender() {
+        //new DataSender(
             SinglePointDataSender sender = new SinglePointDataSender(getDomain(),
-                                               parent.getAddress(),
-                                               parent.getPort(),
-                                               parent.getSenderState() );
-            sender.setKeepAliveMaxRequestCount(parent.getKeepAliveMaxRequestCount());
-            sender.setKeepAliveTimeout(parent.getKeepAliveTimeout());
-            sender.setAckTimeout(parent.getAckTimeout());
-            sender.setWaitForAck(parent.getWaitForAck());
-            sender.setResend(parent.isResend());
-            sender.setRxBufSize(parent.getRxBufSize());
-            sender.setTxBufSize(parent.getTxBufSize());
+                                               getHost(),
+                                               getPort(),
+                                               getSenderState() );
+            sender.setKeepAliveMaxRequestCount(getKeepAliveMaxRequestCount());
+            sender.setKeepAliveTimeout(getKeepAliveTimeout());
+            sender.setAckTimeout(getAckTimeout());
+            sender.setWaitForAck(getWaitForAck());
+            sender.setResend(isResend());
+            sender.setRxBufSize(getRxBufSize());
+            sender.setTxBufSize(getTxBufSize());
             return sender;
 
-        }
-
-        public void close() {
-            synchronized (mutex) {
-                for (int i = 0; i < queue.size(); i++) {
-                    SinglePointDataSender sender = (SinglePointDataSender) queue.get(i);
-                    sender.disconnect();
-                }//for
-                for (int i = 0; i < inuse.size(); i++) {
-                    SinglePointDataSender sender = (SinglePointDataSender) inuse.get(i);
-                    sender.disconnect();
-                }//for
-                queue.clear();
-                inuse.clear();
-                isOpen = false;
-                mutex.notifyAll();
-            }
-        }
-
-        public void open() {
-            synchronized (mutex) {
-                isOpen = true;
-                mutex.notifyAll();
-            }
-        }
     }
 }

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ParallelNioSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ParallelNioSender.java?rev=382509&r1=382508&r2=382509&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ParallelNioSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ParallelNioSender.java
Thu Mar  2 12:53:55 2006
@@ -17,16 +17,16 @@
 
 
 import java.io.IOException;
+import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.util.HashMap;
+import java.util.Iterator;
 
 import org.apache.catalina.tribes.ChannelException;
 import org.apache.catalina.tribes.ChannelMessage;
 import org.apache.catalina.tribes.Member;
 import org.apache.catalina.tribes.io.ClusterData;
 import org.apache.catalina.tribes.io.XByteBuffer;
-import java.util.Iterator;
-import java.nio.channels.SelectionKey;
 import org.apache.catalina.tribes.tcp.MultiPointSender;
 
 /**
@@ -53,7 +53,8 @@
     protected int rxBufSize = 43800;
     protected int txBufSize = 25188;
     protected boolean suspect = false;
-    
+    private boolean connected;
+
     public ParallelNioSender(long timeout, 
                              boolean waitForAck,
                              int retryAttempts,
@@ -189,6 +190,7 @@
     
     public void connect() {
         //do nothing, we connect on demand
+        setConnected(true);
     }
     
     
@@ -211,6 +213,7 @@
     
     public synchronized void disconnect() {
         try {close(); }catch (Exception x){}
+        setConnected(false);
     }
     
     public void finalize() {
@@ -220,6 +223,11 @@
     public boolean getSuspect() {
         return suspect;
     }
+
+    public boolean isConnected() {
+        return connected;
+    }
+
     public void setSuspect(boolean suspect) {
         this.suspect = suspect;
     }
@@ -247,7 +255,9 @@
     public void setTimeout(long timeout) {
         this.timeout = timeout;
     }
-    
-    
+
+    public void setConnected(boolean connected) {
+        this.connected = connected;
+    }
 
 }



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


Mime
View raw message