From 3bf404cc2aa0ada5951716afb8ca078c2f0ba7d0 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Sun, 17 Oct 2010 22:35:35 -0400 Subject: [PATCH] INT-1527, INT-786 refactoread and renamed MetadataPersister to MetadataSore strategy, provided a very simple file-based implementation as FileBasedPropertiesStore which ises DefaultPropertiesPersister, modified FEED module to depend on it --- .../context/IntegrationContextUtils.java | 6 +- .../metadata/FileBasedPropertiesStore.java | 101 ++++++++ .../metadata/MapBasedMetadataPersister.java | 28 --- .../context/metadata/MetadataPersister.java | 15 -- .../context/metadata/MetadataStore.java | 39 ++++ .../PropertiesBasedMetadataPersister.java | 220 ------------------ ...PropertiesBasedMetadataPersisterTests.java | 113 --------- spring-integration-feed/.classpath | 1 - spring-integration-feed/pom.xml | 4 - .../feed/FeedEntryReaderMessageSource.java | 53 ++--- .../FeedEntryReaderMessageSourceTests.java | 2 +- ...essageSourceBeanDefinitionParserTests.java | 4 +- ...AbstractInboundTwitterEndpointSupport.java | 2 - 13 files changed, 162 insertions(+), 426 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MapBasedMetadataPersister.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataPersister.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java delete mode 100644 spring-integration-core/src/test/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersisterTests.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 4fbd85f8a6..48c2ee9d4f 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 @@ -21,7 +21,7 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.core.convert.ConversionService; import org.springframework.integration.MessageChannel; -import org.springframework.integration.context.metadata.MetadataPersister; +import org.springframework.integration.context.metadata.MetadataStore; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.scheduling.TaskScheduler; @@ -48,8 +48,8 @@ public abstract class IntegrationContextUtils { public static final String DEFAULT_POLLER_METADATA_BEAN_NAME = "org.springframework.integration.context.defaultPollerMetadata"; - public static MetadataPersister getMetadataPersister(BeanFactory beanFactory) { - return getBeanOfType(beanFactory, METADATA_PERSISTER_BEAN_NAME, MetadataPersister.class); + public static MetadataStore getMetadataPersister(BeanFactory beanFactory) { + return getBeanOfType(beanFactory, METADATA_PERSISTER_BEAN_NAME, MetadataStore.class); } public static MessageChannel getErrorChannel(BeanFactory beanFactory) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java new file mode 100644 index 0000000000..e30c1015ff --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java @@ -0,0 +1,101 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed 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.springframework.integration.context.metadata; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.Properties; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.util.DefaultPropertiesPersister; + +/** + * @author Oleg Zhurakousky + * @since 2.0 + */ +public class FileBasedPropertiesStore implements MetadataStore { + protected final Log logger = LogFactory.getLog(getClass()); + private final DefaultPropertiesPersister persister = new DefaultPropertiesPersister(); + private final String key; + private File persistentFile; + + public FileBasedPropertiesStore(String key){ + this.key = key; + String dirPath = System.getProperty("java.io.tmpdir") + "spring-integration/"; + String fileName = this.key + ".last.entry"; + File baseDir = new File(dirPath); + baseDir.mkdirs(); + persistentFile = new File(baseDir, fileName); + try { + if (!persistentFile.exists()){ + persistentFile.createNewFile(); + } + } catch (Exception e) { + e.printStackTrace(); + } + + } + + public void write(Properties metadata) { + FileOutputStream fo = null; + try { + fo = new FileOutputStream(persistentFile); + persister.store(metadata, fo, "Last feed entry"); + } + catch (IOException e) { + // not fatal for the functionality of the component + logger.warn("Failed to persist feed entry. This may result in a duplicate " + + "feed entry after this component is restarted", e); + } + finally { + try { + if (fo != null){ + fo.close(); + } + } + catch (IOException e) { + // not fatal for the functionality of he component + logger.warn("Failed to close FileOutputStream to " + persistentFile.getAbsolutePath(), e); + } + } + } + + public Properties load() { + Properties properties = new Properties(); + FileInputStream iStream = null; + try { + iStream = new FileInputStream(persistentFile); + persister.load(properties, iStream); + } catch (Exception e) { + // not fatal for the functionality of the component + logger.warn("Failed to load feed entry from the persistent store. This may result in a duplicate " + + "feed entry after this component is restarted", e); + } finally { + try { + if (iStream != null){ + iStream.close(); + } + } catch (Exception e2) { + // non fatal + logger.warn("Failed to close FileInputStream for: " + persistentFile.getAbsolutePath()); + } + } + return properties; + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MapBasedMetadataPersister.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MapBasedMetadataPersister.java deleted file mode 100644 index be08702ee0..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MapBasedMetadataPersister.java +++ /dev/null @@ -1,28 +0,0 @@ -package org.springframework.integration.context.metadata; - -import org.springframework.util.Assert; - -import java.util.concurrent.ConcurrentHashMap; - -/** - * Simple in-memory implementation of teh {@link org.springframework.integration.context.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/context/metadata/MetadataPersister.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataPersister.java deleted file mode 100644 index 6ba6464308..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataPersister.java +++ /dev/null @@ -1,15 +0,0 @@ -package org.springframework.integration.context.metadata; - - -/** - * 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/context/metadata/MetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java new file mode 100644 index 0000000000..fa243ea61c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java @@ -0,0 +1,39 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed 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.springframework.integration.context.metadata; + +import java.util.Properties; + +/** + * Strategy interface for persisting metadata from certain adapters / endpoints + * to avoid duplicate delivery of messages, for example. + * + * @author Josh Long + * @author Oleg Zhurakousky + * @since 2.0 + */ +public interface MetadataStore { + /** + * Wil write propertoes to a persistent store + * @param metadata + */ + void write(Properties metadata); + /** + * Will load Properties from the persistent store + * @return + */ + Properties load(); +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java deleted file mode 100644 index 7809bd4df2..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java +++ /dev/null @@ -1,220 +0,0 @@ -package org.springframework.integration.context.metadata; - -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.core.task.SimpleAsyncTaskExecutor; -import org.springframework.util.Assert; - -import java.io.*; -import java.util.*; -import java.util.concurrent.Executor; - - -/** - * Implementation of {@link org.springframework.integration.context.metadata.MetadataPersister} that knows how to write metadata - * to a {@link java.util.Properties} instance. - * - * @author Josh Long - */ -public class PropertiesBasedMetadataPersister implements MetadataPersister, InitializingBean { - /** - * Used to queue the writes asynchronously - */ - private Executor executor = new SimpleAsyncTaskExecutor(); - - /** - * 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; - - /** - * 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 Set bootstrapResources = new HashSet(); - private volatile File cachedLocationOfPropertiesFile; - - public PropertiesBasedMetadataPersister(Resource ultimateResourceToWhichToWriteFile) { - setLocationOfPropertiesOnDisk(ultimateResourceToWhichToWriteFile); - } - - @SuppressWarnings("unused") - public PropertiesBasedMetadataPersister(String uniqueName) { - this.uniqueName = uniqueName; - } - - @SuppressWarnings("unused") - public PropertiesBasedMetadataPersister() { - } - - @SuppressWarnings("unused") - public void setExecutor(Executor executor) { - this.executor = executor; - } - - public void setLocationOfPropertiesOnDisk(Resource locationOfPropertiesOnDisk) { - this.locationOfPropertiesOnDisk = locationOfPropertiesOnDisk; - } - - private File buildFileFromUniqueName() { - File tmpDir = new File(System.getProperty("java.io.tmpdir")); - - 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 - */ - @SuppressWarnings("unused") - public void setProperties(Properties properties) { - this.propertiesFactoryBean.setProperties(properties); - } - - public void write(String key, String value) { - Assert.notNull( key != null , "key can't be null"); - Assert.notNull( value != null , "value can't be null"); - synchronized (monitor) { - long now = System.nanoTime(); - this.properties.setProperty(key, value); - - if (this.supportAsyncWrites) { - this.executor.execute(new BackgroundWriterJob(now, key, value, this.properties)); - } else { - doWriteToDisk(now, key, value, this.properties); - } - } - } - - /** - * 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 - */ - @SuppressWarnings("unused") - public void setUniqueName(String uniqueName) { - this.uniqueName = uniqueName; - } - - private void doWriteToDisk(long timestamp, String newKey, String newValue, Properties pro) { - try { - FileOutputStream fileOutputStream = null; - - try { - fileOutputStream = new FileOutputStream (cachedLocationOfPropertiesFile); - pro.store(fileOutputStream, this.uniqueName); - } finally { - if (fileOutputStream != null) { - fileOutputStream.close(); - } - } - } catch (IOException e) { - throw new RuntimeException("couldn't write " + this.properties + " on submission of " + newKey + "=" + newValue + " 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.uniqueName == null) || this.uniqueName.trim().equals("")) { - this.uniqueName = UUID.randomUUID().toString(); - } - - 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); - } - - if (this.supportAsyncWrites) { - Assert.notNull(this.executor, "'executorService' must be set on this bean or defined in the context"); - } - - if (this.locationOfPropertiesOnDisk.exists()) { - this.bootstrapResources.add(locationOfPropertiesOnDisk); - } - - this.cachedLocationOfPropertiesFile = this.locationOfPropertiesOnDisk.getFile(); - - propertiesFactoryBean.setLocations(this.bootstrapResources.toArray(new Resource[bootstrapResources.size()])); - // we take the existing Resources [] and use them to bootstrap a Properties instance when this component wakes up again - propertiesFactoryBean.afterPropertiesSet(); - properties = propertiesFactoryBean.getObject(); - } - } - - @SuppressWarnings("unused") - public void setLocations(Resource[] locations) { - for (int i = 0, locationsLength = locations.length; i < locationsLength; i++) { - Resource r = locations[i]; - this.bootstrapResources.add(r); - } - } - - @SuppressWarnings("unused") - public void setLocation(Resource location) { - this.bootstrapResources.add(location); - } - - - - /** - * 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/test/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersisterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersisterTests.java deleted file mode 100644 index 7de8b67a98..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersisterTests.java +++ /dev/null @@ -1,113 +0,0 @@ -package org.springframework.integration.context.metadata; - -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; - -import org.springframework.core.io.FileSystemResource; -import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.integration.context.metadata.PropertiesBasedMetadataPersister; - -import java.io.*; - - -/** - * Tests the functionality of {@link PropertiesBasedMetadataPersister} - * - * @author Josh Long - */ -public class PropertiesBasedMetadataPersisterTests { - private FileSystemResource fileSystemResource; - private PropertiesBasedMetadataPersister propertiesBasedMetadataPersister; - - @Before - public void setUp() throws Throwable { - File tmpFile = new File(System.getProperty("java.io.tmpdir"), System.currentTimeMillis() + ".properties"); - fileSystemResource = new FileSystemResource(tmpFile); - - if (tmpFile.exists()) { - tmpFile.delete(); - } - } - - @After - public void tearDown() throws Throwable { - if ((this.fileSystemResource != null) && this.fileSystemResource.getFile().exists()) { - this.fileSystemResource.getFile().delete(); - } - } - - @Test - public void testMetadataPersistenceRecovery() throws Throwable { - propertiesBasedMetadataPersister = new PropertiesBasedMetadataPersister(fileSystemResource); - propertiesBasedMetadataPersister.afterPropertiesSet(); - - String timeString = System.currentTimeMillis() + ""; - propertiesBasedMetadataPersister.write("time", timeString); - - propertiesBasedMetadataPersister = new PropertiesBasedMetadataPersister(fileSystemResource); - propertiesBasedMetadataPersister.afterPropertiesSet(); - Assert.assertEquals(propertiesBasedMetadataPersister.read("time"), timeString); - } - - @Test - public void testAsyncMetadataPersistence() throws Throwable { - propertiesBasedMetadataPersister = new PropertiesBasedMetadataPersister(fileSystemResource); - propertiesBasedMetadataPersister.setSupportAsyncWrites(true); - propertiesBasedMetadataPersister.setExecutor(new SimpleAsyncTaskExecutor()); - propertiesBasedMetadataPersister.afterPropertiesSet(); - - for (int i = 1; i <= 30; i++) { - propertiesBasedMetadataPersister.write("sinceId", i + ""); - System.out.println("value written " + i + ", value retreived " + propertiesBasedMetadataPersister.read("sinceId")); - } - - Thread.sleep(1000); - Assert.assertTrue(contentsOfFile(fileSystemResource.getFile()).contains("sinceId=30")); - } - - private String contentsOfFile(File f) { - String txt = null; - int width = 300; - Reader reader = null; - - try { - StringBuffer stringBuffer = new StringBuffer(width); - reader = new FileReader(f); - - char[] values = new char[width]; - - while (reader.read(values) != -1) { - stringBuffer.append(values); - } - - txt = stringBuffer.toString().trim(); - } catch (Throwable e) { - throw new RuntimeException(e); - } finally { - try { - if (reader != null) { - reader.close(); - } - } catch (IOException e) { - // eat it - } - } - - return txt; - } - - @Test - public void testSyncMetadataPersistence() throws Throwable { - propertiesBasedMetadataPersister = new PropertiesBasedMetadataPersister(fileSystemResource); - propertiesBasedMetadataPersister.afterPropertiesSet(); - - for (int i = 1; i <= 30; i++) { - propertiesBasedMetadataPersister.write("sinceId", i + ""); - System.out.println("value written " + i + ", value retreived " + propertiesBasedMetadataPersister.read("sinceId")); - } - - Assert.assertTrue(contentsOfFile(fileSystemResource.getFile()).contains("sinceId=30")); - } -} diff --git a/spring-integration-feed/.classpath b/spring-integration-feed/.classpath index 96489ff1c9..d8b3f86eb6 100644 --- a/spring-integration-feed/.classpath +++ b/spring-integration-feed/.classpath @@ -3,7 +3,6 @@ - diff --git a/spring-integration-feed/pom.xml b/spring-integration-feed/pom.xml index d03f8c6c9b..9d68a9e4a6 100644 --- a/spring-integration-feed/pom.xml +++ b/spring-integration-feed/pom.xml @@ -19,10 +19,6 @@ org.springframework.integration spring-integration-core - - org.springframework.commons - spring-commons-serializer - commons-langcommons-lang2.5 diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java index 1ac9072ba0..bdcec40c7d 100644 --- a/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java @@ -15,10 +15,6 @@ */ package org.springframework.integration.feed; -import java.io.File; -import java.io.FileInputStream; -import java.io.FileOutputStream; -import java.io.IOException; import java.util.Collections; import java.util.Comparator; import java.util.List; @@ -28,10 +24,11 @@ import java.util.concurrent.ConcurrentLinkedQueue; import org.springframework.integration.Message; import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.integration.context.metadata.FileBasedPropertiesStore; +import org.springframework.integration.context.metadata.MetadataStore; import org.springframework.integration.core.MessageSource; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; -import org.springframework.util.DefaultPropertiesPersister; import org.springframework.util.StringUtils; import com.sun.syndication.feed.synd.SyndEntry; @@ -46,18 +43,17 @@ import com.sun.syndication.feed.synd.SyndFeed; * @author Oleg Zhurakousky */ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource{ - private final DefaultPropertiesPersister persister = new DefaultPropertiesPersister(); + private volatile MetadataStore metadataStore; + private volatile Properties lastPersistentEntry = new Properties(); private volatile Queue entries = new ConcurrentLinkedQueue(); private volatile FeedReaderMessageSource feedReaderMessageSource; private final Object monitor = new Object(); private volatile String feedMetadataIdKey; private volatile String persistentIdentifier; - private volatile boolean initialized; private volatile long lastTime = -1; - private volatile File persisterFile; - + private Comparator syndEntryComparator = new Comparator() { public int compare(SyndEntry syndEntry, SyndEntry syndEntry1) { long x = syndEntry.getPublishedDate().getTime() - @@ -81,6 +77,10 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple this.persistentIdentifier = persistentIdentifier; } + public void setMetadataStore(MetadataStore metadataStore) { + this.metadataStore = metadataStore; + } + public String getComponentType(){ return "feed:inbound-channel-adapter"; } @@ -125,14 +125,11 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple @Override protected void onInit() throws Exception { if (StringUtils.hasText(this.persistentIdentifier)){ - File dir = new File(System.getProperty("user.home") + "/temp/spring-integration"); - dir.mkdirs(); - persisterFile = new File(dir, this.persistentIdentifier + ".last.entry"); - if (!persisterFile.exists()){ - persisterFile.createNewFile(); - } - FileInputStream inStream = new FileInputStream(persisterFile); - persister.load(lastPersistentEntry, inStream); + if (this.metadataStore == null){ + logger.info("Creating FileBasedPropertiesStore"); + metadataStore = new FileBasedPropertiesStore(this.persistentIdentifier); + } + lastPersistentEntry = metadataStore.load(); } else { logger.info("Your '" + this.getComponentType() + "' is anonymous (no ID attribute), therefore no feed entries will be persisted " + @@ -158,26 +155,8 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple this.lastTime = next.getPublishedDate().getTime(); this.lastPersistentEntry.put(this.feedMetadataIdKey, this.lastTime + ""); - if (persisterFile != null){ - FileOutputStream fo = null; - try { - fo = new FileOutputStream(persisterFile); - persister.store(this.lastPersistentEntry, fo, "Last feed entry"); - } - catch (IOException e) { - // not fatal for the functionality of the component - logger.warn("Failed to persist feed entry. This may result in a duplicate " + - "feed entry after this component is restarted", e); - } - finally { - try { - fo.close(); - } - catch (IOException e) { - // not fatal for the functionality of he component - logger.warn("Failed to close output stream to " + persisterFile.getAbsolutePath(), e); - } - } + if (metadataStore != null){ + metadataStore.write(this.lastPersistentEntry); } return next; diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java index 286054fd43..202e709c97 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java @@ -41,7 +41,7 @@ import com.sun.syndication.feed.synd.SyndFeed; public class FeedEntryReaderMessageSourceTests { @Before public void prepare(){ - File persisterFile = new File(System.getProperty("user.home") + "/temp/spring-integration", "feedReader.last.entry"); + File persisterFile = new File(System.getProperty("java.io.tmpdir") + "spring-integration/", "feedReader.last.entry"); if (persisterFile.exists()){ persisterFile.delete(); } diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java index 66cce1826c..98df8c90b5 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java @@ -56,7 +56,7 @@ public class FeedMessageSourceBeanDefinitionParserTests { private static CountDownLatch latch; @Before public void prepare(){ - File persisterFile = new File(System.getProperty("user.home") + "/temp/spring-integration", "feedAdapter.last.entry"); + File persisterFile = new File(System.getProperty("java.io.tmpdir") + "spring-integration/", "feedAdapter.last.entry"); if (persisterFile.exists()){ persisterFile.delete(); } @@ -84,7 +84,7 @@ public class FeedMessageSourceBeanDefinitionParserTests { @Test public void validateSuccessfullNewsRetrievalWithFileUrlAndMessageHistory() throws Exception{ - File persisterFile = new File(System.getProperty("user.home") + "/temp/spring-integration", "feedAdapterUsage.last.entry"); + File persisterFile = new File(System.getProperty("java.io.tmpdir") + "spring-integration/", "feedAdapterUsage.last.entry"); if (persisterFile.exists()){ persisterFile.delete(); } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java index 9cfb41bd22..bd52c24af4 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java @@ -20,11 +20,9 @@ import java.util.ArrayList; import java.util.List; import org.apache.commons.lang.exception.ExceptionUtils; - import org.springframework.context.Lifecycle; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; -import org.springframework.integration.context.metadata.MetadataPersister; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.history.HistoryWritingMessagePostProcessor;