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

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

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-10-17 22:35:35 -04:00
parent 6adb950390
commit 3bf404cc2a
13 changed files with 162 additions and 426 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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) {

101
spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java Normal file
View File

@@ -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;
}
}

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

@@ -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 <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);
}
}

15
spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataPersister.java
View File

@@ -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 (<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);
}

39
spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java Normal file
View File

@@ -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();
}

220
spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java
View File

@@ -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<String>, 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<Resource> bootstrapResources = new HashSet<Resource>();
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);
}
}
}
}

113
spring-integration-core/src/test/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersisterTests.java
View File

@@ -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"));
}
}