You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@synapse.apache.org by ra...@apache.org on 2013/08/02 02:27:32 UTC
svn commit: r1509495 - in
/synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp:
./ connectionfactory/ ha/ pollingtask/
Author: rajikak
Date: Fri Aug 2 00:27:32 2013
New Revision: 1509495
URL: http://svn.apache.org/r1509495
Log:
added ha implementations.
Modified:
synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportEndpoint.java
synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportListener.java
synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/connectionfactory/AMQPTransportConnectionFactoryManager.java
synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/ha/AMQPTransportReconnectHandler.java
synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTask.java
synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTaskFactory.java
Modified: synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportEndpoint.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportEndpoint.java?rev=1509495&r1=1509494&r2=1509495&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportEndpoint.java (original)
+++ synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportEndpoint.java Fri Aug 2 00:27:32 2013
@@ -24,7 +24,6 @@ import org.apache.synapse.transport.amqp
import java.util.HashSet;
import java.util.Set;
-import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
/**
@@ -77,7 +76,8 @@ public class AMQPTransportEndpoint exten
service,
workerPool,
this,
- conFac);
+ conFac,
+ transportReceiver.getHaHandler());
} catch (AMQPTransportException e) {
throw new AxisFault("Could not load the AMQP endpoint configuration, " + e.getMessage(), e);
Modified: synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportListener.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportListener.java?rev=1509495&r1=1509494&r2=1509495&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportListener.java (original)
+++ synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/AMQPTransportListener.java Fri Aug 2 00:27:32 2013
@@ -41,7 +41,7 @@ public class AMQPTransportListener exten
private ExecutorService connectionFactoryES;
- private AMQPTransportReconnectHandler haHandler;
+ private AMQPTransportReconnectHandler haHandlerTask;
@Override
protected void doInit() throws AxisFault {
@@ -71,14 +71,14 @@ public class AMQPTransportListener exten
int maxReconnectionDuration = AMQPTransportUtils.getIntProperty(
AMQPTransportConstant.PARAM_MAX_RE_CONNECTION_DURATION, 1000 * 60 * 10);
- haHandler = new AMQPTransportReconnectHandler(
+ haHandlerTask = new AMQPTransportReconnectHandler(
connectionFactoryES,
maxReconnectionDuration,
reconnectionProgressionFactor,
initialReconnectDuration,
connectionFactoryManager);
- new Thread(haHandler, "AMQP-HA-handler-task").start();
+ new Thread(haHandlerTask, "AMQP-HA-handler-task").start();
log.info("AMQP transport listener initializing..");
}
@@ -137,4 +137,8 @@ public class AMQPTransportListener exten
return connectionFactoryManager.getConnectionFactory(
AMQPTransportConstant.DEFAULT_CONNECTION_FACTORY_NAME);
}
+
+ public AMQPTransportReconnectHandler getHaHandler(){
+ return haHandlerTask;
+ }
}
Modified: synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/connectionfactory/AMQPTransportConnectionFactoryManager.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/connectionfactory/AMQPTransportConnectionFactoryManager.java?rev=1509495&r1=1509494&r2=1509495&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/connectionfactory/AMQPTransportConnectionFactoryManager.java (original)
+++ synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/connectionfactory/AMQPTransportConnectionFactoryManager.java Fri Aug 2 00:27:32 2013
@@ -98,8 +98,10 @@ public class AMQPTransportConnectionFact
} catch (IOException e) {
throw new AMQPTransportException("Could not remove the connection '" + name + "'", e);
}
+ } else {
+ throw new AMQPTransportException("No connection factory found with the name '"
+ + name + "'");
}
- throw new AMQPTransportException("No connection factory found with the name '" + name + "'");
}
/**
Modified: synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/ha/AMQPTransportReconnectHandler.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/ha/AMQPTransportReconnectHandler.java?rev=1509495&r1=1509494&r2=1509495&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/ha/AMQPTransportReconnectHandler.java (original)
+++ synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/ha/AMQPTransportReconnectHandler.java Fri Aug 2 00:27:32 2013
@@ -97,7 +97,7 @@ public class AMQPTransportReconnectHandl
" The retry duration is set to initial reconnection duration " +
"value(" + initialReconnectDuration + "s)");
}
- log.error("The reconnection attempt number '" + count++ + "' failed. Next " +
+ log.info("The reconnection attempt number '" + count++ + "' failed. Next " +
"re-try will be after '" + (retryDuration / 1000) + "' seconds");
try {
Thread.sleep(retryDuration);
@@ -117,6 +117,7 @@ public class AMQPTransportReconnectHandl
connectionFactoryManager.removeConnectionFactory(name);
connectionFactoryManager.addConnectionFactory(
name, new AMQPTransportConnectionFactory(param, es));
+ log.info("A new connection factory was created for -> '" + name + "'");
}
String conFacName = entry.getConnectionFactoryName();
@@ -127,7 +128,6 @@ public class AMQPTransportReconnectHandl
new AMQPTransportHABrokerEntry(cf.getChannel(), cf.getConnection()));
entry.getLock().release();
-
while (blockedTasks.isEmpty()) {
entry = blockedTasks.take();
conFacName = entry.getConnectionFactoryName();
@@ -136,13 +136,23 @@ public class AMQPTransportReconnectHandl
connectionMap.put(
entry.getKey(),
new AMQPTransportHABrokerEntry(cf.getChannel(), cf.getConnection()));
+ log.info("The task with key '" + entry.getKey() + "' was combined with a new " +
+ "connection factory");
entry.getLock().release();
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (AMQPTransportException e) {
- log.error("High Availability handler just died!. It's time to re-start", e);
+ log.error("High Availability handler just died!. It's time to reboot the system.", e);
}
}
+
+ public BlockingQueue<AMQPTransportHAEntry> getBlockedTasks() {
+ return blockedTasks;
+ }
+
+ public ConcurrentMap<String, AMQPTransportHABrokerEntry> getConnectionMap() {
+ return connectionMap;
+ }
}
\ No newline at end of file
Modified: synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTask.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTask.java?rev=1509495&r1=1509494&r2=1509495&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTask.java (original)
+++ synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTask.java Fri Aug 2 00:27:32 2013
@@ -29,16 +29,17 @@ import org.apache.axis2.transport.http.H
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.synapse.transport.amqp.*;
+import org.apache.synapse.transport.amqp.ha.AMQPTransportHABrokerEntry;
+import org.apache.synapse.transport.amqp.ha.AMQPTransportHAEntry;
+import org.apache.synapse.transport.amqp.ha.AMQPTransportReconnectHandler;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
+import java.util.UUID;
+import java.util.concurrent.*;
/**
* The polling task deploy for each services exposed on AMQP transport. This task
@@ -207,6 +208,8 @@ public class AMQPTransportPollingTask {
private ScheduledFuture<?> pollingTaskFuture;
+ private AMQPTransportReconnectHandler haHandler;
+
public void setUseTx(boolean useTx) {
isUseTx = useTx;
}
@@ -383,6 +386,10 @@ public class AMQPTransportPollingTask {
this.responseConnectionFactory = responseConnectionFactory;
}
+ public void setHaHandler(AMQPTransportReconnectHandler haHandler) {
+ this.haHandler = haHandler;
+ }
+
/**
* Start the polling task for this service
*/
@@ -507,6 +514,26 @@ public class AMQPTransportPollingTask {
} catch (ShutdownSignalException e) {
log.error("Polling task for service '" + serviceName + "' received a " +
"shutdown signal", e);
+ Semaphore available = new Semaphore(0, true);
+ String key = UUID.randomUUID().toString();
+ haHandler.getBlockedTasks().add(new AMQPTransportHAEntry(
+ available, key, connectionFactoryName));
+ try {
+ available.acquire();
+ } catch (InterruptedException ie) {
+ log.error("The blocking semaphore received an interrupted", e);
+ Thread.currentThread().interrupt();
+ return;
+ }
+
+ AMQPTransportHABrokerEntry brokerEntry = haHandler.getConnectionMap().get(key);
+ if (brokerEntry == null) {
+ log.error("No new connection factory were found for key '" + key + "'");
+ } else {
+ setChannel(brokerEntry.getChannel());
+ this.queueingConsumer = new QueueingConsumer(channel);
+ }
+
} catch (ConsumerCancelledException e) {
log.error("Polling task for service '" + serviceName + "' received a " +
"cancellation signal");
Modified: synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTaskFactory.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTaskFactory.java?rev=1509495&r1=1509494&r2=1509495&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTaskFactory.java (original)
+++ synapse/trunk/java/modules/transports/optional/amqp/src/main/java/org/apache/synapse/transport/amqp/pollingtask/AMQPTransportPollingTaskFactory.java Fri Aug 2 00:27:32 2013
@@ -19,6 +19,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.synapse.transport.amqp.*;
import org.apache.synapse.transport.amqp.connectionfactory.AMQPTransportConnectionFactory;
+import org.apache.synapse.transport.amqp.ha.AMQPTransportReconnectHandler;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
@@ -36,7 +37,8 @@ public class AMQPTransportPollingTaskFac
AxisService service,
ScheduledExecutorService pool,
AMQPTransportEndpoint endpoint,
- AMQPTransportConnectionFactory connectionFactory) throws AxisFault {
+ AMQPTransportConnectionFactory connectionFactory,
+ AMQPTransportReconnectHandler haHandler) throws AxisFault {
Map<String, String> svcParam =
AMQPTransportUtils.getServiceStringParameters(service.getParameters());
@@ -48,6 +50,7 @@ public class AMQPTransportPollingTaskFac
pt.setServiceName(service.getName());
pt.setEndpoint(endpoint);
pt.setPollingTaskScheduler(pool);
+ pt.setHaHandler(haHandler);
// set buffers to hold request/response messages for this task
pt.setBuffers(new AMQPTransportBuffers());
@@ -89,7 +92,6 @@ public class AMQPTransportPollingTaskFac
pt.setInternalExchange(isInternalExchange);
}
-
pt.setChannel(connectionFactory.getChannel());
pt.setConnectionFactoryName(connectionFactory.getName());