Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merged in rev 474799 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/branches/activemq-4.1@474800 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2006-11-14 14:13:23 +00:00
parent 545c909ec9
commit 71c5a277d2
4 changed files with 182 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
activemq-core/src/main/java/org/apache/activemq/pool/ConnectionPool.java
View File

@ -17,18 +17,19 @@
*/
package org.apache.activemq.pool;
import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.commons.pool.ObjectPoolFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import javax.jms.JMSException;
import javax.jms.Session;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.transport.TransportListener;
import org.apache.commons.pool.ObjectPoolFactory;
import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
/**
* Holds a real JMS connection along with the session pools associated with it.
@ -36,13 +37,33 @@ import java.util.Map;
* @version $Revision$
*/
public class ConnectionPool {
private ActiveMQConnection connection;
private Map cache;
private AtomicBoolean started = new AtomicBoolean(false);
private int referenceCount;
private ObjectPoolFactory poolFactory;
private long lastUsed;
private boolean hasFailed;
private int idleTimeout = 30*1000;
public ConnectionPool(ActiveMQConnection connection, ObjectPoolFactory poolFactory) {
this(connection, new HashMap(), poolFactory);
// Add a transport Listener so that we can notice if this connection should be expired due to
// a connection failure.
connection.addTransportListener(new TransportListener(){
public void onCommand(Object command) {
}
public void onException(IOException error) {
synchronized(ConnectionPool.this) {
hasFailed = true;
}
}
public void transportInterupted() {
}
public void transportResumed() {
}
});
}
public ConnectionPool(ActiveMQConnection connection, Map cache, ObjectPoolFactory poolFactory) {
@ -57,7 +78,7 @@ public class ConnectionPool {
}
}
public ActiveMQConnection getConnection() {
synchronized public ActiveMQConnection getConnection() {
return connection;
}
@ -71,20 +92,58 @@ public class ConnectionPool {
return pool.borrowSession();
}
public void close() throws JMSException {
Iterator i = cache.values().iterator();
while (i.hasNext()) {
SessionPool pool = (SessionPool) i.next();
i.remove();
synchronized public void close() {
if( connection!=null ) {
Iterator i = cache.values().iterator();
while (i.hasNext()) {
SessionPool pool = (SessionPool) i.next();
i.remove();
try {
pool.close();
} catch (Exception e) {
}
}
try {
pool.close();
connection.close();
} catch (Exception e) {
}
catch (Exception e) {
throw JMSExceptionSupport.create(e);
}
}
connection.close();
connection = null;
connection = null;
}
}
synchronized public void incrementReferenceCount() {
referenceCount++;
}
synchronized public void decrementReferenceCount() {
referenceCount--;
if( referenceCount == 0 ) {
lastUsed = System.currentTimeMillis();
expiredCheck();
}
}
/**
* @return true if this connection has expired.
*/
synchronized public boolean expiredCheck() {
if( connection == null )
return true;
if( hasFailed || idleTimeout> 0 && System.currentTimeMillis() > lastUsed+idleTimeout ) {
if( referenceCount == 0 ) {
close();
}
return true;
}
return false;
}
public int getIdleTimeout() {
return idleTimeout;
}
public void setIdleTimeout(int idleTimeout) {
this.idleTimeout = idleTimeout;
}
}

8
activemq-core/src/main/java/org/apache/activemq/pool/PooledConnection.java
View File

@ -56,6 +56,7 @@ public class PooledConnection implements TopicConnection, QueueConnection {
public PooledConnection(ConnectionPool pool) {
this.pool = pool;
this.pool.incrementReferenceCount();
}
/**
@ -66,7 +67,10 @@ public class PooledConnection implements TopicConnection, QueueConnection {
}
public void close() throws JMSException {
pool = null;
if( this.pool!=null ) {
this.pool.decrementReferenceCount();
this.pool = null;
}
}
public void start() throws JMSException {
@ -133,7 +137,7 @@ public class PooledConnection implements TopicConnection, QueueConnection {
// Implementation methods
// -------------------------------------------------------------------------
protected ActiveMQConnection getConnection() throws JMSException {
ActiveMQConnection getConnection() throws JMSException {
assertNotClosed();
return pool.getConnection();
}

36
activemq-core/src/main/java/org/apache/activemq/pool/PooledConnectionFactory.java
View File

@ -17,22 +17,20 @@
*/
package org.apache.activemq.pool;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.Service;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.pool.ObjectPoolFactory;
import org.apache.commons.pool.impl.GenericObjectPoolFactory;
import org.apache.commons.pool.impl.GenericObjectPool.Config;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.Service;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.commons.pool.ObjectPoolFactory;
import org.apache.commons.pool.impl.GenericObjectPoolFactory;
/**
* A JMS provider which pools Connection, Session and MessageProducer instances
@ -79,6 +77,13 @@ public class PooledConnectionFactory implements ConnectionFactory, Service {
public synchronized Connection createConnection(String userName, String password) throws JMSException {
ConnectionKey key = new ConnectionKey(userName, password);
ConnectionPool connection = (ConnectionPool) cache.get(key);
// Now.. we might get a connection, but it might be that we need to
// dump it..
if( connection!=null && connection.expiredCheck() ) {
connection=null;
}
if (connection == null) {
ActiveMQConnection delegate = createConnection(key);
connection = new ConnectionPool(delegate, getPoolFactory());
@ -109,17 +114,10 @@ public class PooledConnectionFactory implements ConnectionFactory, Service {
}
public void stop() throws Exception {
ServiceStopper stopper = new ServiceStopper();
for (Iterator iter = cache.values().iterator(); iter.hasNext();) {
ConnectionPool connection = (ConnectionPool) iter.next();
try {
connection.close();
}
catch (JMSException e) {
stopper.onException(this, e);
}
connection.close();
}
stopper.throwFirstException();
}
public ObjectPoolFactory getPoolFactory() {

79
activemq-core/src/test/java/org/apache/activemq/pool/ConnectionFailureEvictsFromPool.java Normal file
View File

@ -0,0 +1,79 @@
/**
*
* 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.activemq.pool;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.test.TestSupport;
import org.apache.activemq.transport.mock.MockTransport;
public class ConnectionFailureEvictsFromPool extends TestSupport {
private BrokerService broker;
private ActiveMQConnectionFactory factory;
private PooledConnectionFactory pooledFactory;
protected void setUp() throws Exception {
broker = new BrokerService();
broker.setPersistent(false);
TransportConnector connector = broker.addConnector("tcp://localhost:0");
broker.start();
factory = new ActiveMQConnectionFactory("mock:"+connector.getConnectUri());
pooledFactory = new PooledConnectionFactory(factory);
}
public void testEviction() throws Exception {
Connection connection = pooledFactory.createConnection();
sendMessage(connection);
createConnectionFailure(connection);
try {
sendMessage(connection);
fail("Expected Error");
} catch ( JMSException e) {
}
// If we get another connection now it should be a new connection that works.
Connection connection2 = pooledFactory.createConnection();
sendMessage(connection2);
}
private void createConnectionFailure(Connection connection) throws Exception {
ActiveMQConnection c = ((PooledConnection)connection).getConnection();
MockTransport t = (MockTransport) c.getTransportChannel().narrow(MockTransport.class);
t.stop();
}
private void sendMessage(Connection connection) throws JMSException {
Session session = connection.createSession(false, 0);
MessageProducer producer = session.createProducer(new ActiveMQQueue("FOO"));
producer.send(session.createTextMessage("Test"));
session.close();
}
protected void tearDown() throws Exception {
broker.stop();
}
}