From 7bee2bf9320ad115ece1b97b0975ac2087a32063 Mon Sep 17 00:00:00 2001 From: Josh Long Date: Mon, 16 Aug 2010 06:51:36 +0000 Subject: [PATCH] Initial cut of the MetadataPersister support. Built two implementations - one for in-memory use cases and one for .properties-file based support. INT-1332 --- .../context/IntegrationContextUtils.java | 70 +++--- .../metadata/MapBasedMetadataPersister.java | 28 +++ .../endpoint/metadata/MetadataPersister.java | 17 ++ .../PropertiesBasedMetadataPersister.java | 219 ++++++++++++++++++ .../integration/handler/DelayHandler.java | 2 +- 5 files changed, 305 insertions(+), 31 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MapBasedMetadataPersister.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MetadataPersister.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/PropertiesBasedMetadataPersister.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java index 587c3f14a4..fa2ee5ec91 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java @@ -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 getBeanOfType(BeanFactory beanFactory, String beanName, Class 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 getBeanOfType(BeanFactory beanFactory, String beanName, Class type) { + if (!beanFactory.containsBean(beanName)) { + return null; + } + return beanFactory.getBean(beanName, type); + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MapBasedMetadataPersister.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MapBasedMetadataPersister.java new file mode 100644 index 0000000000..d383441e35 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MapBasedMetadataPersister.java @@ -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 the type of objects to be stored as values. Keys will always be {@link String} + */ +public class MapBasedMetadataPersister implements MetadataPersister { + + private ConcurrentHashMap metadataMap = new ConcurrentHashMap() ; + + 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); + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MetadataPersister.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MetadataPersister.java new file mode 100644 index 0000000000..c188f15c8a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/MetadataPersister.java @@ -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 (*.ini based). + *

+ * 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 { + void write(String key, V value); + V read(String key); +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/PropertiesBasedMetadataPersister.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/PropertiesBasedMetadataPersister.java new file mode 100644 index 0000000000..51a8ea1e1c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/metadata/PropertiesBasedMetadataPersister.java @@ -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. + *

+ * 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, 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 bootstrapResources = new ConcurrentSkipListSet(); + + 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); + } + } + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 8482f963cd..672b1e12f6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -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;