You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@synapse.apache.org by ru...@apache.org on 2008/05/08 20:18:18 UTC
svn commit: r654596 - in
/synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport:
base/AbstractTransportListener.java jms/JMSListener.java
mail/MailTransportListener.java vfs/VFSTransportListener.java
Author: ruwan
Date: Thu May 8 11:17:01 2008
New Revision: 654596
URL: http://svn.apache.org/viewvc?rev=654596&view=rev
Log:
Fixing the issue SYNAPSE-298 and minor changes
Modified:
synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/base/AbstractTransportListener.java
synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/jms/JMSListener.java
synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/mail/MailTransportListener.java
synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java
Modified: synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/base/AbstractTransportListener.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/base/AbstractTransportListener.java?rev=654596&r1=654595&r2=654596&view=diff
==============================================================================
--- synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/base/AbstractTransportListener.java (original)
+++ synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/base/AbstractTransportListener.java Thu May 8 11:17:01 2008
@@ -428,7 +428,6 @@
/**
* Utility method to allow transports to register MBeans
- * @param mbs the MBeanServer
* @param mbeanInstance bean instance
* @param objectName name
*/
Modified: synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/jms/JMSListener.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/jms/JMSListener.java?rev=654596&r1=654595&r2=654596&view=diff
==============================================================================
--- synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/jms/JMSListener.java (original)
+++ synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/jms/JMSListener.java Thu May 8 11:17:01 2008
@@ -19,16 +19,19 @@
import org.apache.axis2.Constants;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.context.ConfigurationContext;
-import org.apache.axis2.description.*;
+import org.apache.axis2.description.AxisService;
+import org.apache.axis2.description.Parameter;
+import org.apache.axis2.description.TransportInDescription;
import org.apache.synapse.transport.base.AbstractTransportListener;
+import org.apache.synapse.transport.base.BaseConstants;
import org.apache.synapse.transport.base.BaseUtils;
import org.apache.synapse.transport.base.ManagementSupport;
-import org.apache.synapse.transport.base.BaseConstants;
-import org.apache.commons.logging.LogFactory;
import javax.jms.JMSException;
import javax.naming.NamingException;
-import java.util.*;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
/**
* The JMS Transport listener implementation. A JMS Listner will hold one or
@@ -133,6 +136,10 @@
if (serviceName.indexOf('/') != -1) {
serviceName = serviceName.substring(0, serviceName.indexOf('/'));
}
+ // strip out the endpoint name if present
+ if (serviceName.indexOf('.') != -1) {
+ serviceName = serviceName.substring(0, serviceName.indexOf('.'));
+ }
return new EndpointReference[]{
new EndpointReference((String) serviceNameToEPRMap.get(serviceName))};
}
@@ -165,8 +172,8 @@
String destinationType = JMSUtils.getDestinationTypeForService(service);
- log.info("Starting to listen on destination : " + destinationName + " of type " + destinationType
- + " for service " + service.getName());
+ log.info("Starting to listen on destination : " + destinationName + " of type "
+ + destinationType + " for service " + service.getName());
cf.addDestination(destinationName, destinationType, service.getName());
cf.startListeningOnDestination(destinationName, destinationType);
}
Modified: synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/mail/MailTransportListener.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/mail/MailTransportListener.java?rev=654596&r1=654595&r2=654596&view=diff
==============================================================================
--- synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/mail/MailTransportListener.java (original)
+++ synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/mail/MailTransportListener.java Thu May 8 11:17:01 2008
@@ -58,7 +58,7 @@
public static final String MOVE = "MOVE";
/** Keep the list of directories/files and poll durations */
- private final List pollTable = new ArrayList();
+ private final List<PollTableEntry> pollTable = new ArrayList<PollTableEntry>();
/**
* Initializes the Mail transport
@@ -78,11 +78,8 @@
* it is time to scan the contents for new files
*/
public void onPoll() {
- Iterator iter = pollTable.iterator();
- while (iter.hasNext()) {
- PollTableEntry entry = (PollTableEntry) iter.next();
+ for (PollTableEntry entry : pollTable) {
long startTime = System.currentTimeMillis();
-
if (startTime > entry.getNextPollTime()) {
checkMail(entry, entry.getEmailAddress());
}
@@ -472,12 +469,10 @@
* @throws AxisFault not used
*/
public EndpointReference[] getEPRsForService(String serviceName, String ip) throws AxisFault {
- Iterator iter = pollTable.iterator();
- while (iter.hasNext()) {
- PollTableEntry entry = (PollTableEntry) iter.next();
- if (entry.getServiceName().equals(serviceName)) {
- return new EndpointReference[]{
- new EndpointReference(
+ for (PollTableEntry entry : pollTable) {
+ if (entry.getServiceName().equals(serviceName) ||
+ serviceName.startsWith(entry.getServiceName() + ".")) {
+ return new EndpointReference[]{new EndpointReference(
MailConstants.TRANSPORT_PREFIX + entry.getEmailAddress())};
}
}
@@ -580,9 +575,7 @@
}
protected void stopListeningForService(AxisService service) {
- Iterator iter = pollTable.iterator();
- while (iter.hasNext()) {
- PollTableEntry entry = (PollTableEntry) iter.next();
+ for (PollTableEntry entry : pollTable) {
if (service.getName().equals(entry.getServiceName())) {
cancelPoll(service);
pollTable.remove(entry);
Modified: synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java?rev=654596&r1=654595&r2=654596&view=diff
==============================================================================
--- synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java (original)
+++ synapse/branches/1.2/modules/transports/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java Thu May 8 11:17:01 2008
@@ -99,7 +99,7 @@
public static final String MOVE = "MOVE";
/** Keep the list of directories/files and poll durations */
- private final List pollTable = new ArrayList();
+ private final List<PollTableEntry> pollTable = new ArrayList<PollTableEntry>();
/** The VFS file system manager */
private FileSystemManager fsManager = null;
@@ -128,11 +128,8 @@
* it is time to scan the contents for new files
*/
public void onPoll() {
- Iterator iter = pollTable.iterator();
- while (iter.hasNext()) {
- PollTableEntry entry = (PollTableEntry) iter.next();
+ for (PollTableEntry entry : pollTable) {
long startTime = System.currentTimeMillis();
-
if (startTime > entry.getNextPollTime()) {
scanFileOrDirectory(entry, entry.getFileURI());
}
@@ -159,7 +156,7 @@
int maxRetryCount = entry.getMaxRetryCount();
long reconnectionTimeout = entry.getReconnectTimeout();
- while(wasError == true) {
+ while(wasError) {
try {
retryCount++;
fileObject = fsManager.resolveFile(fileURI);
@@ -178,7 +175,7 @@
return;
}
- if(wasError == true) {
+ if(wasError) {
try {
Thread.sleep(reconnectionTimeout);
} catch (InterruptedException e2) {
@@ -220,31 +217,31 @@
if (log.isDebugEnabled()) {
log.debug("File name pattern :" + entry.getFileNamePattern());
}
- for (int i = 0; i < children.length; i++) {
+ for (FileObject child : children) {
if (log.isDebugEnabled()) {
- log.debug("Matching file :" + children[i].getName().getBaseName());
+ log.debug("Matching file :" + child.getName().getBaseName());
}
- if ( (entry.getFileNamePattern() != null)
- && (children[i].getName().getBaseName().matches(entry.getFileNamePattern()))) {
+ if ((entry.getFileNamePattern() != null)
+ && (child.getName().getBaseName().matches(entry.getFileNamePattern()))) {
try {
if (log.isDebugEnabled()) {
- log.debug("Processing file :" + children[i]);
+ log.debug("Processing file :" + child);
}
- processFile(entry, children[i]);
+ processFile(entry, child);
successCount++;
// tell moveOrDeleteAfterProcessing() file was success
entry.setLastPollState(PollTableEntry.SUCCSESSFUL);
metrics.incrementMessagesReceived();
} catch (Exception e) {
- logException("Error processing File URI : " + children[i].getName(), e);
+ logException("Error processing File URI : " + child.getName(), e);
failCount++;
// tell moveOrDeleteAfterProcessing() file failed
entry.setLastPollState(PollTableEntry.FAILED);
metrics.incrementFaultsReceiving();
- }
+ }
- moveOrDeleteAfterProcessing(entry, children[i]);
+ moveOrDeleteAfterProcessing(entry, child);
}
}
@@ -355,7 +352,7 @@
metrics.incrementBytesReceived(content.getSize());
- Map transportHeaders = new HashMap();
+ Map<String, Object> transportHeaders = new HashMap<String, Object>();
transportHeaders.put(VFSConstants.FILE_PATH, filePath);
transportHeaders.put(VFSConstants.FILE_NAME, fileName);
@@ -363,7 +360,8 @@
transportHeaders.put(VFSConstants.FILE_LENGTH, Long.valueOf(content.getSize()));
} catch (FileSystemException ignore) {}
try {
- transportHeaders.put(VFSConstants.LAST_MODIFIED, Long.valueOf(content.getLastModifiedTime()));
+ transportHeaders.put(VFSConstants.LAST_MODIFIED,
+ Long.valueOf(content.getLastModifiedTime()));
} catch (FileSystemException ignore) {}
// compute the unique message ID
@@ -469,11 +467,10 @@
* @throws AxisFault not used
*/
public EndpointReference[] getEPRsForService(String serviceName, String ip) throws AxisFault {
- Iterator iter = pollTable.iterator();
- while (iter.hasNext()) {
- PollTableEntry entry = (PollTableEntry) iter.next();
- if (entry.getServiceName().equals(serviceName)) {
- return new EndpointReference[]{ new EndpointReference("vfs:" + entry.getFileURI())};
+ for (PollTableEntry entry : pollTable) {
+ if (entry.getServiceName().equals(serviceName)
+ || serviceName.startsWith(entry.getServiceName() + ".")) {
+ return new EndpointReference[]{new EndpointReference("vfs:" + entry.getFileURI())};
}
}
return null;