From 562b9814b9e9de1e518ae32067151b953fd48a10 Mon Sep 17 00:00:00 2001
From: fhanik <fhanik@13f79535-47bb-0310-9956-ffa450edef68>
Date: Fri, 27 Apr 2007 19:40:21 +0000
Subject: [PATCH] Undo Pero's checkin, sorry Peter, just can't have it in there
 until the solution is more robust, although I don't think it will ever be
 implemented in association with memberships 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
---
 .../catalina/tribes/transport/PooledSender.java    | 36 +++++++++-------------
 .../tribes/transport/nio/ParallelNioSender.java    | 14 ++++++++-
 2 files changed, 27 insertions(+), 23 deletions(-)

diff --git a/java/org/apache/catalina/tribes/transport/PooledSender.java b/java/org/apache/catalina/tribes/transport/PooledSender.java
index fa2dcb209..26afc7715 100644
--- 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 2b41edf59..01100eb14 100644
--- 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;
     }
 
-- 
2.11.0