Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Initial cut of the MetadataPersister support. Built two implementations - one for in-memory use cases and one for .properties-file based support. INT-1332

Browse Source
This commit is contained in:
Josh Long
2010-08-16 06:51:36 +00:00
parent c15ebb07d8
commit 7bee2bf932
5 changed files with 305 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
View File

@@ -13,61 +13,71 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.context;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.metadata.MetadataPersister;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
* Utility methods for accessing common integration components from the BeanFactory.
*
*
* @author Mark Fisher
* @author Josh Long
*/
public abstract class IntegrationContextUtils {
public static final String TASK_SCHEDULER_BEAN_NAME = "taskScheduler";
public static final String METADATA_PERSISTER_BEAN_NAME = "metadataPersister";
public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel";
public static final String TASK_SCHEDULER_BEAN_NAME = "taskScheduler";
public static final String NULL_CHANNEL_BEAN_NAME = "nullChannel";
public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel";
public static final String INTEGRATION_CONVERSION_SERVICE_BEAN_NAME = "integrationConversionService";
public static final String NULL_CHANNEL_BEAN_NAME = "nullChannel";
public static final String DEFAULT_POLLER_METADATA_BEAN_NAME = "org.springframework.integration.context.defaultPollerMetadata";
public static final String INTEGRATION_CONVERSION_SERVICE_BEAN_NAME = "integrationConversionService";
public static final String DEFAULT_POLLER_METADATA_BEAN_NAME = "org.springframework.integration.context.defaultPollerMetadata";
public static MessageChannel getErrorChannel(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, ERROR_CHANNEL_BEAN_NAME, MessageChannel.class);
}
public static MetadataPersister getMetadataPersister(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, METADATA_PERSISTER_BEAN_NAME, MetadataPersister.class);
}
public static TaskScheduler getTaskScheduler(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class);
}
public static MessageChannel getErrorChannel(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, ERROR_CHANNEL_BEAN_NAME, MessageChannel.class);
}
public static TaskScheduler getRequiredTaskScheduler(BeanFactory beanFactory) {
TaskScheduler taskScheduler = getTaskScheduler(beanFactory);
Assert.state(taskScheduler != null, "No such bean '" + TASK_SCHEDULER_BEAN_NAME + "'");
return taskScheduler;
}
public static TaskScheduler getTaskScheduler(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class);
}
public static PollerMetadata getDefaultPollerMetadata(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, DEFAULT_POLLER_METADATA_BEAN_NAME, PollerMetadata.class);
}
public static TaskScheduler getRequiredTaskScheduler(BeanFactory beanFactory) {
TaskScheduler taskScheduler = getTaskScheduler(beanFactory);
Assert.state(taskScheduler != null, "No such bean '" + TASK_SCHEDULER_BEAN_NAME + "'");
return taskScheduler;
}
public static ConversionService getConversionService(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, INTEGRATION_CONVERSION_SERVICE_BEAN_NAME, ConversionService.class);
}
public static PollerMetadata getDefaultPollerMetadata(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, DEFAULT_POLLER_METADATA_BEAN_NAME, PollerMetadata.class);
}
private static <T> T getBeanOfType(BeanFactory beanFactory, String beanName, Class<T> type) {
if (!beanFactory.containsBean(beanName)) {
return null;
}
return beanFactory.getBean(beanName, type);
}
public static ConversionService getConversionService(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, INTEGRATION_CONVERSION_SERVICE_BEAN_NAME, ConversionService.class);
}
private static <T> T getBeanOfType(BeanFactory beanFactory, String beanName, Class<T> type) {
if (!beanFactory.containsBean(beanName)) {
return null;
}
return beanFactory.getBean(beanName, type);
}
}

28
spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MapBasedMetadataPersister.java Normal file
View File

@@ -0,0 +1,28 @@
package org.springframework.integration.endpoint.metadata;
import org.springframework.util.Assert;
import java.util.concurrent.ConcurrentHashMap;
/**
* Simple in-memory implementation of teh {@link org.springframework.integration.endpoint.metadata.MetadataPersister}
* interface suitable for the use cases where it's assured that component only needs ephemeral metadata.
*
*
* @author Josh Long
* @param <T> the type of objects to be stored as values. Keys will always be {@link String}
*/
public class MapBasedMetadataPersister <T> implements MetadataPersister<T> {
private ConcurrentHashMap<String,T> metadataMap = new ConcurrentHashMap<String,T>() ;
public void write(String key, T value) {
Assert.notNull( key != null , "key can't be null");
Assert.notNull( value != null , "value can't be null");
this.metadataMap.put( key, value);
}
public T read(String key) {
return this.metadataMap.get(key);
}
}

17
spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MetadataPersister.java Normal file
View File

@@ -0,0 +1,17 @@
package org.springframework.integration.endpoint.metadata;
import java.util.Properties;
/**
* Envisioned as a strategy interface for persisting metadata from certain adapters / endpoints. Ideally,
* there will be at least two options - one ephemeral persister (RAM-only) and one durable (<code>*.ini</code> based).
* <p/>
* This is used to give adapters / endpoints a place to store metadata to avoid duplicate delivery of messages, for example.
*
* @author Josh Long
*/
public interface MetadataPersister<V> {
void write(String key, V value);
V read(String key);
}

219
spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/PropertiesBasedMetadataPersister.java Normal file
View File

@@ -0,0 +1,219 @@
package org.springframework.integration.endpoint.metadata;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Date;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentSkipListSet;
/**
* Implementation of {@link org.springframework.integration.endpoint.metadata.MetadataPersister} that knows how to write metadata
* to a {@link java.util.Properties} instance.
* <p/>
* TODO could this perhaps participate or at least be aware of our transaction synchronization mechanism? IE: no guarantees, but we at least try to write on commit()s?
*
* @author Josh Long
*/
public class PropertiesBasedMetadataPersister implements MetadataPersister<String>, InitializingBean, BeanFactoryAware {
/**
* Used to encapsulate acquisition of a {@link java.util.Properties} instance if it's prefered that we handled it on the client's behalf
*/
private PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
/**
* guard for initialization and writes
*/
private final Object monitor = new Object();
/**
* This would enable a background thread that would write as possible, but not block #write calls
*/
private volatile boolean supportAsyncWrites;
/**
* An existing {@link java.util.Properties} file that we can read in at startup. This is utlimately forwarded to {@link org.springframework.beans.factory.config.PropertiesFactoryBean} on startup
*/
private Properties properties;
/**
* An executor (only useful for the background writes if async writes are supported)
*/
private TaskScheduler taskScheduler;
/**
* Users can either provide a unique name and we can automatically setup #locationOfPropertiesOnDisk
*/
private String uniqueName;
/**
* Or, a user can stipulate a {@link org.springframework.core.io.Resource} directly
*/
private Resource locationOfPropertiesOnDisk;
private BeanFactory beanFactory;
private Set<Resource> bootstrapResources = new ConcurrentSkipListSet<Resource>();
public PropertiesBasedMetadataPersister(Resource ultimateResourceToWhichToWriteFile) {
this.locationOfPropertiesOnDisk = ultimateResourceToWhichToWriteFile;
}
public PropertiesBasedMetadataPersister(String uniqueName) {
this.uniqueName = uniqueName;
}
public PropertiesBasedMetadataPersister() {
}
public void setLocationOfPropertiesOnDisk(Resource locationOfPropertiesOnDisk) {
this.locationOfPropertiesOnDisk = locationOfPropertiesOnDisk;
}
private File buildFileFromUniqueName() {
File tmpDir = new File(System.getProperty("java.io.tmpdir"));
if ((this.uniqueName == null) || this.uniqueName.trim().equals("")) {
this.uniqueName = UUID.randomUUID().toString();
}
String un = this.uniqueName + ".properties";
return new File(tmpDir, un);
}
/**
* Optional - if there's already a {@link java.util.Properties} instance in play than we can simply use that one.
*
* @param properties existing properties, just in case
*/
public void setProperties(Properties properties) {
this.propertiesFactoryBean.setProperties(properties);
}
public void write(String key, String value) {
synchronized (monitor) {
long now = System.nanoTime();
this.properties.setProperty(key, value);
if (this.supportAsyncWrites) {
this.taskScheduler.schedule(new BackgroundWriterJob(now, key, value, this.properties), new Date(now));
} else {
doWriteToDisk(now, key, value, this.properties);
}
}
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
/**
* This is required to ensure contiuity across restarts. It must be meaningful to a given application of a given component.
*
* @param uniqueName the unqiue name to use in constructing a {@link org.springframework.core.io.Resource} for the {@link java.util.Properties} file
*/
public void setUniqueName(String uniqueName) {
this.uniqueName = uniqueName;
}
private void doWriteToDisk(long timestamp, String newKey, String newValue, Properties pro) {
try {
FileWriter fileWriter = new FileWriter(this.locationOfPropertiesOnDisk.getFile());
pro.store(fileWriter, this.uniqueName);
} catch (IOException e) {
throw new RuntimeException("couldn't write " + this.properties + " on submission of to disk at " + new Date(timestamp).toString());
}
}
public String read(String key) {
return this.properties.getProperty(key);
}
public void setSupportAsyncWrites(boolean supportAsyncWrites) {
this.supportAsyncWrites = supportAsyncWrites;
}
public void afterPropertiesSet() throws Exception {
synchronized (this.monitor) {
if ((this.locationOfPropertiesOnDisk == null) && (this.uniqueName == null)) {
throw new RuntimeException("you must either specify a property file Resource or a uniqueName that can be used in generated a path that will be input into creating a Resource");
}
if ((this.locationOfPropertiesOnDisk == null)) {
File pathOfPropertiesFileOnDisk = buildFileFromUniqueName();
this.locationOfPropertiesOnDisk = new FileSystemResource(pathOfPropertiesFileOnDisk);
}
taskScheduler = (this.taskScheduler == null) ? IntegrationContextUtils.getTaskScheduler(this.beanFactory) : taskScheduler;
if (this.supportAsyncWrites) {
Assert.notNull(this.taskScheduler, "'taskScheduler' must be set on this bean or defined in the context");
}
if (this.locationOfPropertiesOnDisk.exists()) {
this.bootstrapResources.add(locationOfPropertiesOnDisk);
}
// we take the existing Resources [] and use them to bootstrap a Property file when this component wakes up again
propertiesFactoryBean.afterPropertiesSet();
properties = propertiesFactoryBean.getObject();
}
}
public void setLocations(Resource[] locations) {
for (int i = 0, locationsLength = locations.length; i < locationsLength; i++) {
Resource r = locations[i];
this.bootstrapResources.add(r);
}
}
public void setLocation(Resource location) {
this.bootstrapResources.add(location);
}
public void setBeanFactory(BeanFactory beanFactory)
throws BeansException {
this.beanFactory = beanFactory;
}
/**
* This class is used to ensure that the properies are persisted to the right place as soon as capacity / the task Scheduler allows
*/
private class BackgroundWriterJob implements Runnable {
private volatile Properties properties;
private String key;
private String value;
private long now;
public BackgroundWriterJob(long now, String key, String value, Properties properties) {
this.properties = properties;
this.now = now;
this.key = key;
this.value = value;
}
public void run() {
synchronized (monitor) {
doWriteToDisk(this.now, this.key, this.value, this.properties);
}
}
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -78,7 +78,7 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan
private volatile String delayHeaderName;
private boolean waitForTasksToCompleteOnShutdown = false;;
private boolean waitForTasksToCompleteOnShutdown = false;
private volatile MessageChannel outputChannel;