summaryrefslogtreecommitdiff
path: root/java/common/src
diff options
context:
space:
mode:
authorRobert Greig <rgreig@apache.org>2007-04-09 16:12:49 +0000
committerRobert Greig <rgreig@apache.org>2007-04-09 16:12:49 +0000
commitfbe47c01d43f20ac09998778bfa1676d351ab650 (patch)
tree823e66310978fe4e53800cc8b5af51077bd6c35d /java/common/src
parent63c1ee312fa77ec98c9fc8eccb0a7aa0a1689fe4 (diff)
downloadqpid-python-fbe47c01d43f20ac09998778bfa1676d351ab650.tar.gz
Got rid of some uses of System.out instead of log4j logging.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@526807 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java/common/src')
-rw-r--r--java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java101
1 files changed, 42 insertions, 59 deletions
diff --git a/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java b/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
index 552ecf6b66..c9c96925cb 100644
--- a/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
+++ b/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
@@ -7,9 +7,9 @@
* 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
@@ -20,16 +20,17 @@
*/
package org.apache.qpid.pool;
-import org.apache.qpid.pool.Event.CloseEvent;
-
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.log4j.Logger;
+
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoFilterAdapter;
import org.apache.mina.common.IoSession;
+import org.apache.qpid.pool.Event.CloseEvent;
+
public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionHandler
{
private static final Logger _logger = Logger.getLogger(PoolingFilter.class);
@@ -49,12 +50,12 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
void fireAsynchEvent(IoSession session, Event event)
{
Job job = getJobForSession(session);
- // job.acquire(); //prevents this job being removed from _jobs
+ // job.acquire(); //prevents this job being removed from _jobs
job.add(event);
- //Additional checks on pool to check that it hasn't shutdown.
+ // Additional checks on pool to check that it hasn't shutdown.
// The alternative is to catch the RejectedExecutionException that will result from executing on a shutdown pool
- if (job.activate() && _poolReference.getPool() != null && !_poolReference.getPool().isShutdown())
+ if (job.activate() && (_poolReference.getPool() != null) && !_poolReference.getPool().isShutdown())
{
_poolReference.getPool().execute(job);
}
@@ -70,16 +71,6 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
private Job getJobForSession(IoSession session)
{
return (Job) session.getAttribute(_name);
-
-/* if(job == null)
- {
- System.err.println("Error in " + _name);
- Thread.dumpStack();
- }
-
-
- job = _jobs.get(session);
- return job == null ? createJobForSession(session) : job;*/
}
private Job createJobForSession(IoSession session)
@@ -89,35 +80,36 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
private Job addJobForSession(IoSession session, Job job)
{
- //atomic so ensures all threads agree on the same job
+ // atomic so ensures all threads agree on the same job
Job existing = _jobs.putIfAbsent(session, job);
- return existing == null ? job : existing;
+
+ return (existing == null) ? job : existing;
}
- //Job.JobCompletionHandler
+ // Job.JobCompletionHandler
public void completed(IoSession session, Job job)
{
-// if (job.isComplete())
-// {
-// job.release();
-// if (!job.isReferenced())
-// {
-// _jobs.remove(session);
-// }
-// }
-// else
- if(!job.isComplete())
+ // if (job.isComplete())
+ // {
+ // job.release();
+ // if (!job.isReferenced())
+ // {
+ // _jobs.remove(session);
+ // }
+ // }
+ // else
+ if (!job.isComplete())
{
// ritchiem : 2006-12-13 Do we need to perform the additional checks here?
- // Can the pool be shutdown at this point?
- if (job.activate() && _poolReference.getPool() != null && !_poolReference.getPool().isShutdown())
+ // Can the pool be shutdown at this point?
+ if (job.activate() && (_poolReference.getPool() != null) && !_poolReference.getPool().isShutdown())
{
_poolReference.getPool().execute(job);
}
}
}
- //IoFilter methods that are processed by threads on the pool
+ // IoFilter methods that are processed by threads on the pool
public void sessionOpened(final NextFilter nextFilter, final IoSession session) throws Exception
{
@@ -129,37 +121,33 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
nextFilter.sessionClosed(session);
}
- public void sessionIdle(final NextFilter nextFilter, final IoSession session,
- final IdleStatus status) throws Exception
+ public void sessionIdle(final NextFilter nextFilter, final IoSession session, final IdleStatus status) throws Exception
{
nextFilter.sessionIdle(session, status);
}
- public void exceptionCaught(final NextFilter nextFilter, final IoSession session,
- final Throwable cause) throws Exception
+ public void exceptionCaught(final NextFilter nextFilter, final IoSession session, final Throwable cause) throws Exception
{
- nextFilter.exceptionCaught(session,cause);
+ nextFilter.exceptionCaught(session, cause);
}
- public void messageReceived(final NextFilter nextFilter, final IoSession session,
- final Object message) throws Exception
+ public void messageReceived(final NextFilter nextFilter, final IoSession session, final Object message) throws Exception
{
- nextFilter.messageReceived(session,message);
+ nextFilter.messageReceived(session, message);
}
- public void messageSent(final NextFilter nextFilter, final IoSession session,
- final Object message) throws Exception
+ public void messageSent(final NextFilter nextFilter, final IoSession session, final Object message) throws Exception
{
nextFilter.messageSent(session, message);
}
- public void filterWrite(final NextFilter nextFilter, final IoSession session,
- final WriteRequest writeRequest) throws Exception
+ public void filterWrite(final NextFilter nextFilter, final IoSession session, final WriteRequest writeRequest)
+ throws Exception
{
nextFilter.filterWrite(session, writeRequest);
}
- //IoFilter methods that are processed on current thread (NOT on pooled thread)
+ // IoFilter methods that are processed on current thread (NOT on pooled thread)
public void filterClose(NextFilter nextFilter, IoSession session) throws Exception
{
@@ -201,8 +189,8 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
super(refCountingPool, name);
}
- public void messageReceived(final NextFilter nextFilter, final IoSession session,
- final Object message) throws Exception
+ public void messageReceived(final NextFilter nextFilter, final IoSession session, final Object message)
+ throws Exception
{
fireAsynchEvent(session, new Event.ReceivedEvent(nextFilter, message));
@@ -223,9 +211,8 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
super(refCountingPool, name);
}
-
- public void filterWrite(final NextFilter nextFilter, final IoSession session,
- final WriteRequest writeRequest) throws Exception
+ public void filterWrite(final NextFilter nextFilter, final IoSession session, final WriteRequest writeRequest)
+ throws Exception
{
fireAsynchEvent(session, new Event.WriteEvent(nextFilter, writeRequest));
}
@@ -234,21 +221,17 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
{
fireAsynchEvent(session, new CloseEvent(nextFilter));
}
-
}
- public static PoolingFilter createAynschReadPoolingFilter(ReferenceCountingExecutorService refCountingPool,String name)
+ public static PoolingFilter createAynschReadPoolingFilter(ReferenceCountingExecutorService refCountingPool, String name)
{
- return new AsynchReadPoolingFilter(refCountingPool,name);
+ return new AsynchReadPoolingFilter(refCountingPool, name);
}
-
- public static PoolingFilter createAynschWritePoolingFilter(ReferenceCountingExecutorService refCountingPool,String name)
+ public static PoolingFilter createAynschWritePoolingFilter(ReferenceCountingExecutorService refCountingPool, String name)
{
- return new AsynchWritePoolingFilter(refCountingPool,name);
+ return new AsynchWritePoolingFilter(refCountingPool, name);
}
}
-
-