QPID-615: Merge rupertlssmiths 581293

QPID-615, Added patched version of MINAs VM Pipe cleanup thread. Will replace once bug fix is in newer version of MINA.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@685151 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 250 insertions, 0 deletions

diff --git a/qpid/java/client/src/test/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java b/qpid/java/client/src/test/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java
new file mode 100644
index 0000000000..5323ad28bf
--- /dev/null
+++ b/qpid/java/client/src/test/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java
@@ -0,0 +1,125 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.mina.transport.vmpipe.support;
+
+import org.apache.mina.common.IdleStatus;
+
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * This file is a patch to override MINA, because of the IdentityHashMap bug. Workaround to be supplied in MINA 1.0.7.
+ * This patched file will be removed once upgraded onto a newer MINA.
+ *
+ * Dectects idle sessions and fires <tt>sessionIdle</tt> events to them.
+ *
+ * @author The Apache Directory Project (mina-dev@directory.apache.org)
+ */
+public class VmPipeIdleStatusChecker
+{
+    private static final VmPipeIdleStatusChecker INSTANCE = new VmPipeIdleStatusChecker();
+
+    public static VmPipeIdleStatusChecker getInstance()
+    {
+        return INSTANCE;
+    }
+
+    private final Map sessions = new HashMap(); // will use as a set
+
+    private final Worker worker = new Worker();
+
+    private VmPipeIdleStatusChecker()
+    {
+        worker.start();
+    }
+
+    public void addSession(VmPipeSessionImpl session)
+    {
+        synchronized (sessions)
+        {
+            sessions.put(session, session);
+        }
+    }
+
+    private class Worker extends Thread
+    {
+        private Worker()
+        {
+            super("VmPipeIdleStatusChecker");
+            setDaemon(true);
+        }
+
+        public void run()
+        {
+            for (;;)
+            {
+                try
+                {
+                    Thread.sleep(1000);
+                }
+                catch (InterruptedException e)
+                { }
+
+                long currentTime = System.currentTimeMillis();
+
+                synchronized (sessions)
+                {
+                    Iterator it = sessions.keySet().iterator();
+                    while (it.hasNext())
+                    {
+                        VmPipeSessionImpl session = (VmPipeSessionImpl) it.next();
+                        if (!session.isConnected())
+                        {
+                            it.remove();
+                        }
+                        else
+                        {
+                            notifyIdleSession(session, currentTime);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private void notifyIdleSession(VmPipeSessionImpl session, long currentTime)
+    {
+        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.BOTH_IDLE), IdleStatus.BOTH_IDLE,
+            Math.max(session.getLastIoTime(), session.getLastIdleTime(IdleStatus.BOTH_IDLE)));
+        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.READER_IDLE), IdleStatus.READER_IDLE,
+            Math.max(session.getLastReadTime(), session.getLastIdleTime(IdleStatus.READER_IDLE)));
+        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.WRITER_IDLE), IdleStatus.WRITER_IDLE,
+            Math.max(session.getLastWriteTime(), session.getLastIdleTime(IdleStatus.WRITER_IDLE)));
+    }
+
+    private void notifyIdleSession0(VmPipeSessionImpl session, long currentTime, long idleTime, IdleStatus status,
+        long lastIoTime)
+    {
+        if ((idleTime > 0) && (lastIoTime != 0) && ((currentTime - lastIoTime) >= idleTime))
+        {
+            session.increaseIdleCount(status);
+            session.getFilterChain().fireSessionIdle(session, status);
+        }
+    }
+
+}
diff --git a/qpid/java/systests/src/main/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java b/qpid/java/systests/src/main/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java
new file mode 100644
index 0000000000..5323ad28bf
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/mina/transport/vmpipe/support/VmPipeIdleStatusChecker.java
@@ -0,0 +1,125 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.mina.transport.vmpipe.support;
+
+import org.apache.mina.common.IdleStatus;
+
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * This file is a patch to override MINA, because of the IdentityHashMap bug. Workaround to be supplied in MINA 1.0.7.
+ * This patched file will be removed once upgraded onto a newer MINA.
+ *
+ * Dectects idle sessions and fires <tt>sessionIdle</tt> events to them.
+ *
+ * @author The Apache Directory Project (mina-dev@directory.apache.org)
+ */
+public class VmPipeIdleStatusChecker
+{
+    private static final VmPipeIdleStatusChecker INSTANCE = new VmPipeIdleStatusChecker();
+
+    public static VmPipeIdleStatusChecker getInstance()
+    {
+        return INSTANCE;
+    }
+
+    private final Map sessions = new HashMap(); // will use as a set
+
+    private final Worker worker = new Worker();
+
+    private VmPipeIdleStatusChecker()
+    {
+        worker.start();
+    }
+
+    public void addSession(VmPipeSessionImpl session)
+    {
+        synchronized (sessions)
+        {
+            sessions.put(session, session);
+        }
+    }
+
+    private class Worker extends Thread
+    {
+        private Worker()
+        {
+            super("VmPipeIdleStatusChecker");
+            setDaemon(true);
+        }
+
+        public void run()
+        {
+            for (;;)
+            {
+                try
+                {
+                    Thread.sleep(1000);
+                }
+                catch (InterruptedException e)
+                { }
+
+                long currentTime = System.currentTimeMillis();
+
+                synchronized (sessions)
+                {
+                    Iterator it = sessions.keySet().iterator();
+                    while (it.hasNext())
+                    {
+                        VmPipeSessionImpl session = (VmPipeSessionImpl) it.next();
+                        if (!session.isConnected())
+                        {
+                            it.remove();
+                        }
+                        else
+                        {
+                            notifyIdleSession(session, currentTime);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private void notifyIdleSession(VmPipeSessionImpl session, long currentTime)
+    {
+        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.BOTH_IDLE), IdleStatus.BOTH_IDLE,
+            Math.max(session.getLastIoTime(), session.getLastIdleTime(IdleStatus.BOTH_IDLE)));
+        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.READER_IDLE), IdleStatus.READER_IDLE,
+            Math.max(session.getLastReadTime(), session.getLastIdleTime(IdleStatus.READER_IDLE)));
+        notifyIdleSession0(session, currentTime, session.getIdleTimeInMillis(IdleStatus.WRITER_IDLE), IdleStatus.WRITER_IDLE,
+            Math.max(session.getLastWriteTime(), session.getLastIdleTime(IdleStatus.WRITER_IDLE)));
+    }
+
+    private void notifyIdleSession0(VmPipeSessionImpl session, long currentTime, long idleTime, IdleStatus status,
+        long lastIoTime)
+    {
+        if ((idleTime > 0) && (lastIoTime != 0) && ((currentTime - lastIoTime) >= idleTime))
+        {
+            session.increaseIdleCount(status);
+            session.getFilterChain().fireSessionIdle(session, status);
+        }
+    }
+
+}