Undo Pero's checkin, sorry Peter, just can't have it in there until the solution...

For now, senders that are no longer connected on the remote end, will be removed during the keepalive heartbeat

git-svn-id: https://svn.apache.org/repos/asf/tomcat/tc6.0.x/trunk@533197 13f79535-47bb-0310-9956-ffa450edef68

diff --git a/java/org/apache/catalina/tribes/transport/PooledSender.java b/java/org/apache/catalina/tribes/transport/PooledSender.java
index fa2dcb2..26afc77 100644 (file)
--- a/java/org/apache/catalina/tribes/transport/PooledSender.java
+++ b/java/org/apache/catalina/tribes/transport/PooledSender.java
@@ -18,7 +18,6 @@ package org.apache.catalina.tribes.transport;
 
 import java.io.IOException;
 import java.util.List;
-
 import org.apache.catalina.tribes.Member;
 
 /**
@@ -86,13 +85,15 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS
     }
 
     public void add(Member member) {
-        // we don't need it senders are pooled...
+        // no op, senders created upon demans
     }
-    
-    public void remove(Member member) {
-        queue.remove(member) ;
-    }     
 
+    public void remove(Member member) {
+        //no op for now, should not cancel out any keys
+        //can create serious sync issues
+        //all TCP connections are cleared out through keepalive
+        //and if remote node disappears
+    }
     //  ----------------------------------------------------- Inner Class
 
     private class SenderQueue {
@@ -100,17 +101,17 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS
 
         PooledSender parent = null;
 
-        private List<DataSender> notinuse = null;
+        private List notinuse = null;
 
-        private List<DataSender> inuse = null;
+        private List inuse = null;
 
         private boolean isOpen = true;
 
         public SenderQueue(PooledSender parent, int limit) {
             this.limit = limit;
             this.parent = parent;
-            notinuse = new java.util.LinkedList<DataSender>();
-            inuse = new java.util.LinkedList<DataSender>();
+            notinuse = new java.util.LinkedList();
+            inuse = new java.util.LinkedList();
         }
 
         /**
@@ -149,18 +150,6 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS
             return result;
         }
 
-        // FIXME: remove also inuse senders. but then we must synch with sendMessage!
-        public synchronized void remove(Member member) {
-            if (isOpen) {
-                DataSender[] list = new DataSender[notinuse.size()];
-                notinuse.toArray(list);
-                for (int i=0; i<list.length; i++) {
-                    if(list[i] instanceof MultiPointSender)
-                        ((MultiPointSender)list[i]).remove(member);
-                }
-            }
-        }
-        
         public synchronized DataSender getSender(long timeout) {
             long start = System.currentTimeMillis();
             while ( true ) {
@@ -213,6 +202,9 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS
             notinuse.clear();
             inuse.clear();
             notify();
+            
+
+
         }
 
         public synchronized void open() {

diff --git a/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java b/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
index 2b41edf..01100eb 100644 (file)
--- a/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
+++ b/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
@@ -278,7 +278,6 @@ public class ParallelNioSender extends AbstractSender implements MultiPointSende
     }
 
     public boolean keepalive() {
-        //throw new UnsupportedOperationException("Method ParallelNioSender.checkKeepAlive() not implemented");
         boolean result = false;
         for ( Iterator i = nioSenders.entrySet().iterator(); i.hasNext();  ) {
             Map.Entry entry = (Map.Entry)i.next();
@@ -286,8 +285,21 @@ public class ParallelNioSender extends AbstractSender implements MultiPointSende
             if ( sender.keepalive() ) {
                 nioSenders.remove(entry.getKey());
                 result = true;
+            } else {
+                try {
+                    sender.read(null);
+                }catch ( IOException x ) {
+                    sender.disconnect();
+                    sender.reset();
+                    nioSenders.remove(entry.getKey());
+                    result = true;
+                }catch ( Exception x ) {
+                    log.warn("Error during keepalive test for sender:"+sender,x);
+                }
             }
         }
+        //clean up any cancelled keys
+        if ( result ) try { selector.selectNow(); }catch (Exception ignore){}
         return result;
     }