SpringCloud/spring-cloud-dataflow-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Batch file ingest updates for SFTP 2.0 source

Browse Source 
Resolves #55
This commit is contained in:
Chris Schaefer
2018-05-02 00:24:20 -04:00
parent ef2c13ef68
commit 596589a12f
17 changed files with 1013 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
batch/file-ingest-sftp/data/people.csv Normal file
View File

@@ -0,0 +1,3 @@
Jane,Doe
John,Doe
Joe,Doe
1 Jane Doe
2 John Doe
3 Joe Doe

117
batch/file-ingest-sftp/pom.xml Normal file
View File

@@ -0,0 +1,117 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework</groupId>
<artifactId>ingest-sftp</artifactId>
<version>1.0.0</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.1.RELEASE</version>
</parent>
<properties>
<java.version>1.8</java.version>
<maven.compiler.plugin.version>3.7.0</maven.compiler.plugin.version>
<spring.cloud.task.version>1.2.2.RELEASE</spring.cloud.task.version>
<spring.integration.version>5.0.4.RELEASE</spring.integration.version>
<commons.io.version>2.4</commons.io.version>
<sshd.core.version>1.6.0</sshd.core.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-task-core</artifactId>
<version>${spring.cloud.task.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-task-batch</artifactId>
<version>${spring.cloud.task.version}</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>${commons.io.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-file</artifactId>
<version>${spring.integration.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-sftp</artifactId>
<version>${spring.integration.version}</version>
</dependency>
<dependency>
<groupId>org.apache.sshd</groupId>
<artifactId>sshd-core</artifactId>
<version>${sshd.core.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.hsqldb</groupId>
<artifactId>hsqldb</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-test</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>${maven.compiler.plugin.version}</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
<testSource>${java.version}</testSource>
<testTarget>${java.version}</testTarget>
<compilerArgument>-Xlint:all</compilerArgument>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>repository.spring.milestone</id>
<name>Spring Milestone Repository</name>
<url>http://repo.spring.io/milestone</url>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</project>

18
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/Application.java Normal file
View File

@@ -0,0 +1,18 @@
package org.springframework.ingest;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.task.configuration.EnableTask;
/**
* Main entry point for the ingest sample application.
*
* @author Chris Schaefer
*/
@EnableTask
@SpringBootApplication
public class Application {
public static void main(String[] args) throws Exception {
SpringApplication.run(Application.class, args);
}
}

137
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfiguration.java Normal file
View File

@@ -0,0 +1,137 @@
package org.springframework.ingest.config;
import java.io.File;
import javax.sql.DataSource;
import org.apache.commons.io.FileUtils;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemStreamReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.ingest.domain.Person;
import org.springframework.ingest.mapper.fieldset.PersonFieldSetMapper;
import org.springframework.ingest.processor.PersonItemProcessor;
import org.springframework.ingest.resource.RemoteResource;
import org.springframework.ingest.resource.sftp.SftpRemoteResource;
/**
* Class used to configure the batch job related beans.
*
* @author Chris Schaefer
*/
@Configuration
@EnableBatchProcessing
@EnableConfigurationProperties(BatchConfigurationProperties.class)
public class BatchConfiguration {
private final DataSource dataSource;
private final JobBuilderFactory jobBuilderFactory;
private final StepBuilderFactory stepBuilderFactory;
private final BatchConfigurationProperties batchConfigurationProperties;
@Autowired
public BatchConfiguration(final DataSource dataSource, final JobBuilderFactory jobBuilderFactory,
final StepBuilderFactory stepBuilderFactory,
final BatchConfigurationProperties batchConfigurationProperties) {
this.dataSource = dataSource;
this.jobBuilderFactory = jobBuilderFactory;
this.stepBuilderFactory = stepBuilderFactory;
this.batchConfigurationProperties = batchConfigurationProperties;
}
@Bean
@StepScope
public StepExecutionListener ingestStepExecutionListener(@Value("#{jobParameters['remoteFilePath']}") String remoteFilePath,
@Value("#{jobParameters['localFilePath']}") String localFilePath) {
return new StepExecutionListener() {
@Override
public void beforeStep(StepExecution stepExecution) {
try {
Resource fetchedResource = remoteResource().getResource(remoteFilePath);
FileUtils.copyInputStreamToFile(fetchedResource.getInputStream(), new File(localFilePath));
}
catch (Exception e) {
throw new RuntimeException("Could not write remote file to local disk", e);
}
}
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
return null;
}
};
}
@Bean
@StepScope
public ItemStreamReader<Person> reader(@Value("#{jobParameters['localFilePath']}") String localFilePath) throws Exception {
return new FlatFileItemReaderBuilder<Person>()
.name("reader")
.resource(new FileSystemResource(localFilePath))
.delimited()
.names(new String[] {"firstName", "lastName"})
.fieldSetMapper(new PersonFieldSetMapper())
.build();
}
@Bean
public ItemProcessor<Person, Person> processor() {
return new PersonItemProcessor();
}
@Bean
public ItemWriter<Person> writer() {
return new JdbcBatchItemWriterBuilder<Person>()
.beanMapped()
.dataSource(this.dataSource)
.sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
.build();
}
@Bean
public Job ingestJob() throws Exception {
return jobBuilderFactory.get("ingestJob")
.incrementer(new RunIdIncrementer())
.flow(step1())
.end()
.build();
}
@Bean
public Step step1() throws Exception {
return stepBuilderFactory.get("ingest")
.<Person, Person>chunk(10)
.reader(reader(null))
.processor(processor())
.writer(writer())
.listener(ingestStepExecutionListener(null, null))
.build();
}
@Bean
public RemoteResource remoteResource() {
return new SftpRemoteResource(batchConfigurationProperties.getSftpHost(), batchConfigurationProperties.getSftpPort(),
batchConfigurationProperties.getSftpUsername(), batchConfigurationProperties.getSftpPassword());
}
}

48
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfigurationProperties.java Normal file
View File

@@ -0,0 +1,48 @@
package org.springframework.ingest.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* Configuration Properties for BatchConfiguration
*
* @author Chris Schaefer
*/
@ConfigurationProperties
public class BatchConfigurationProperties {
private String sftpHost;
private Integer sftpPort;
private String sftpUsername;
private String sftpPassword;
public String getSftpHost() {
return sftpHost;
}
public void setSftpHost(String sftpHost) {
this.sftpHost = sftpHost;
}
public Integer getSftpPort() {
return sftpPort;
}
public void setSftpPort(Integer sftpPort) {
this.sftpPort = sftpPort;
}
public String getSftpUsername() {
return sftpUsername;
}
public void setSftpUsername(String sftpUsername) {
this.sftpUsername = sftpUsername;
}
public String getSftpPassword() {
return sftpPassword;
}
public void setSftpPassword(String sftpPassword) {
this.sftpPassword = sftpPassword;
}
}

29
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/domain/Person.java Normal file
View File

@@ -0,0 +1,29 @@
package org.springframework.ingest.domain;
/**
* Domain object representing data about a Person.
*
* @author Chris Schaefer
*/
public class Person {
private final String firstName;
private final String lastName;
public Person(final String firstName, final String lastName) {
this.firstName = firstName;
this.lastName = lastName;
}
public String getFirstName() {
return firstName;
}
public String getLastName() {
return lastName;
}
@Override
public String toString() {
return "First name: " + firstName + " , last name: " + lastName;
}
}

21
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapper.java Normal file
View File

@@ -0,0 +1,21 @@
package org.springframework.ingest.mapper.fieldset;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.ingest.domain.Person;
/**
* Maps the provided FieldSet into a Person object.
*
* @author Chris Schaefer
*/
public class PersonFieldSetMapper implements FieldSetMapper<Person> {
@Override
public Person mapFieldSet(FieldSet fieldSet) {
String firstName = fieldSet.readString(0);
String lastName = fieldSet.readString(1);
return new Person(firstName, lastName);
}
}

29
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/processor/PersonItemProcessor.java Normal file
View File

@@ -0,0 +1,29 @@
package org.springframework.ingest.processor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.ingest.domain.Person;
/**
* Processes the providing record, transforming the data into
* uppercase characters.
*
* @author Chris Schaefer
*/
public class PersonItemProcessor implements ItemProcessor<Person, Person> {
private static final Logger LOGGER = LoggerFactory.getLogger(PersonItemProcessor.class);
@Override
public Person process(Person person) throws Exception {
String firstName = person.getFirstName().toUpperCase();
String lastName = person.getLastName().toUpperCase();
Person processedPerson = new Person(firstName, lastName);
LOGGER.info("Processed: " + person + " into: " + processedPerson);
return processedPerson;
}
}

12
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/RemoteResource.java Normal file
View File

@@ -0,0 +1,12 @@
package org.springframework.ingest.resource;
import org.springframework.core.io.Resource;
/**
* Interface definition for remote Resource implementations.
*
* @author Chris Schaefer
*/
public interface RemoteResource {
Resource getResource(String resourceLocation);
}

81
batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/sftp/SftpRemoteResource.java Normal file
View File

@@ -0,0 +1,81 @@
package org.springframework.ingest.resource.sftp;
import java.io.InputStream;
import org.apache.commons.io.IOUtils;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.Resource;
import org.springframework.ingest.resource.RemoteResource;
import org.springframework.integration.file.remote.InputStreamCallback;
import org.springframework.integration.file.remote.RemoteFileOperations;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
import org.springframework.util.Assert;
/**
* RemoteResource implementation utilizing a Spring Integration SftpRemoteFileTemplate
* to connect and return a file as a Spring Resource.
*
* @author Chris Schaefer
*/
public class SftpRemoteResource implements RemoteResource {
private static final Integer DEFAULT_SFTP_PORT = 22;
private static final String LOCALHOST = "127.0.0.1";
private final String host;
private final Integer port;
private final String username;
private final String password;
public SftpRemoteResource(String host, Integer port, String username, String password) {
Assert.hasText(username, "Username must be defined");
Assert.hasText(password, "Password must be defined");
this.host = host;
this.port = port;
this.username = username;
this.password = password;
}
@Override
public Resource getResource(String resourceLocation) {
DefaultSftpSessionFactory sessionFactory = getSessionFactory();
RemoteFileOperations remoteFileOperations = new SftpRemoteFileTemplate(sessionFactory);
FileFetcher filefetcher = new FileFetcher();
remoteFileOperations.get(resourceLocation, filefetcher);
return new ByteArrayResource(filefetcher.getBytes());
}
private DefaultSftpSessionFactory getSessionFactory() {
DefaultSftpSessionFactory sessionFactory = new DefaultSftpSessionFactory();
sessionFactory.setHost(host != null ? host : LOCALHOST);
sessionFactory.setPort(port != null ? port : DEFAULT_SFTP_PORT);
sessionFactory.setUser(username);
sessionFactory.setPassword(password);
sessionFactory.setAllowUnknownKeys(true);
return sessionFactory;
}
private static class FileFetcher implements InputStreamCallback {
private byte[] bytes;
@Override
public void doWithInputStream(InputStream inputStream) {
try {
bytes = IOUtils.toByteArray(inputStream);
}
catch (Exception e) {
throw new RuntimeException("Failed to convert InputStream to byte array", e);
}
}
public byte[] getBytes() {
return bytes;
}
}
}

1
batch/file-ingest-sftp/src/main/resources/application.properties Normal file
View File

@@ -0,0 +1 @@
spring.application.name=fileIngestSftp

5
batch/file-ingest-sftp/src/main/resources/schema-all.sql Normal file
View File

@@ -0,0 +1,5 @@
CREATE TABLE IF NOT EXISTS people (
person_id BIGINT IDENTITY NOT NULL PRIMARY KEY,
first_name VARCHAR(20),
last_name VARCHAR(20)
);

202
batch/file-ingest-sftp/src/test/java/org/springframework/ingest/config/BatchConfigurationTests.java Normal file
View File

@@ -0,0 +1,202 @@
package org.springframework.ingest.config;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.core.io.ResourceLoader;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseFactory;
import org.springframework.jdbc.datasource.init.DatabasePopulatorUtils;
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.ClassUtils;
import org.apache.commons.io.FileUtils;
import org.apache.sshd.common.NamedFactory;
import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory;
import org.apache.sshd.server.Command;
import org.apache.sshd.server.SshServer;
import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
import org.apache.sshd.server.subsystem.sftp.SftpSubsystemFactory;
import java.io.File;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import javax.annotation.PostConstruct;
import javax.sql.DataSource;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.rules.TemporaryFolder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/**
* BatchConfiguration test cases
*
* @author Chris Schaefer
*/
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
@DirtiesContext
@EnableConfigurationProperties(BatchConfigurationProperties.class)
public class BatchConfigurationTests {
private static int port;
private static SshServer server;
private static AnnotationConfigApplicationContext context;
private static final String SFTP_USER = "user";
private static final String SFTP_PASS = "pass";
private static final String SFTP_HOST = "127.0.0.1";
private static final String REMOTE_FILE = "people.csv";
private static final String HOST_KEY_FILE = "hostkey.ser";
@ClassRule
public static final TemporaryFolder remoteTemporaryFolder = new TemporaryFolder();
@BeforeClass
public static void createServer() throws Exception {
File createdFile = remoteTemporaryFolder.newFile(REMOTE_FILE);
FileUtils.writeStringToFile(createdFile, "Jill,Doe\nJoe,Doe\nJustin,Doe\nJane,Doe\nJohn,Doe");
server = SshServer.setUpDefaultServer();
server.setPasswordAuthenticator((username, password, session) -> true);
server.setPort(0);
server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File(HOST_KEY_FILE)));
server.setSubsystemFactories(Collections.<NamedFactory<Command>>singletonList(new SftpSubsystemFactory()));
server.setFileSystemFactory(new VirtualFileSystemFactory(remoteTemporaryFolder.getRoot().toPath()));
server.start();
port = server.getPort();
}
@AfterClass
public static void stopServer() throws Exception {
server.stop();
File hostkey = new File(HOST_KEY_FILE);
if (hostkey.exists()) {
hostkey.delete();
}
}
@Before
public void createContext() {
Map<String, Object> properties = new HashMap<String, Object>();
properties.put("sftp_host", SFTP_HOST);
properties.put("sftp_port", port);
properties.put("sftp_username", SFTP_USER);
properties.put("sftp_password", SFTP_PASS);
ConfigurableEnvironment environment = new StandardEnvironment();
environment.getPropertySources().addFirst(new MapPropertySource("sftpProperties", properties));
context = new AnnotationConfigApplicationContext();
context.register(BatchConfiguration.class, DataSourceConfiguration.class);
context.setEnvironment(environment);
context.refresh();
}
@After
public void closeContext() {
context.close();
}
@Test
public void testBatchConfigurationSuccess() throws Exception {
JobExecution jobExecution = testJob(REMOTE_FILE);
assertEquals("Incorrect batch status", BatchStatus.COMPLETED, jobExecution.getStatus());
assertEquals("Invalid number of step executions", 1, jobExecution.getStepExecutions().size());
}
@Test
public void testBatchConfigurationFail() throws Exception {
JobExecution jobExecution = testJob("missing-people-file.csv");
assertEquals("Incorrect batch status", BatchStatus.FAILED, jobExecution.getStatus());
}
@Test
public void testBatchDataProcessing() throws Exception {
JobExecution jobExecution = testJob(REMOTE_FILE);
assertEquals("Incorrect batch status", BatchStatus.COMPLETED, jobExecution.getStatus());
assertEquals("Invalid number of step executions", 1, jobExecution.getStepExecutions().size());
JdbcTemplate jdbcTemplate = new JdbcTemplate(context.getBean(DataSource.class));
List<Map<String, Object>> peopleList = jdbcTemplate.queryForList("select first_name, last_name from people");
assertEquals("Incorrect number of results", 5, peopleList.size());
for(Map<String, Object> person : peopleList) {
assertNotNull("Received null person", person);
String firstName = (String) person.get("first_name");
assertEquals("Invalid first name: " + firstName, firstName.toUpperCase(), firstName);
String lastName = (String) person.get("last_name");
assertEquals("Invalid last name: " + lastName, lastName.toUpperCase(), lastName);
}
}
private JobExecution testJob(String filePath) throws Exception {
Job job = context.getBean(Job.class);
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
File localFile = File.createTempFile("local", ".csv");
localFile.deleteOnExit();
JobParameters jobParameters = new JobParametersBuilder()
.addString("localFilePath", localFile.getAbsolutePath())
.addString("remoteFilePath", filePath)
.toJobParameters();
return jobLauncher.run(job, jobParameters);
}
@Configuration
public static class DataSourceConfiguration {
@Autowired
private ResourceLoader resourceLoader;
@PostConstruct
protected void initialize() {
ResourceDatabasePopulator populator = new ResourceDatabasePopulator();
populator.addScript(resourceLoader.getResource(ClassUtils.addResourcePathToPackagePath(Step.class, "schema-hsqldb.sql")));
populator.addScript(resourceLoader.getResource("classpath:schema-all.sql"));
populator.setContinueOnError(true);
DatabasePopulatorUtils.execute(populator, dataSource());
}
@Bean
public DataSource dataSource() {
return new EmbeddedDatabaseFactory().getDatabase();
}
}
}

36
batch/file-ingest-sftp/src/test/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapperTests.java Normal file
View File

@@ -0,0 +1,36 @@
package org.springframework.ingest.mapper.fieldset;
import org.junit.Test;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.DefaultFieldSet;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.ingest.domain.Person;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
/**
* Test cases for PersonFieldSetMapper.
*
* @author Chris Schaefer
*/
public class PersonFieldSetMapperTests {
private static final String[] TOKENS = new String[] { "jane", "doe" };
private static final String[] NAMES = new String[] { "firstName", "lastName" };
@Test
public void testPersonFieldMapping() throws Exception {
FieldSet fieldSet = new DefaultFieldSet(TOKENS, NAMES);
FieldSetMapper<Person> fieldSetMapper = new PersonFieldSetMapper();
Person person = fieldSetMapper.mapFieldSet(fieldSet);
assertNotNull("Received null Person", person);
assertNotNull("Received null first name", person.getFirstName());
assertNotNull("Received null last name", person.getLastName());
assertEquals("Received wrong first name", TOKENS[0], person.getFirstName());
assertEquals("Received wrong last name", TOKENS[1], person.getLastName());
}
}

35
batch/file-ingest-sftp/src/test/java/org/springframework/ingest/processor/PersonItemProcessorTests.java Normal file
View File

@@ -0,0 +1,35 @@
package org.springframework.ingest.processor;
import org.junit.Test;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.ingest.domain.Person;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
/**
* Test cases for PersonItemProcessor.
*
* @author Chris Schaefer
*/
public class PersonItemProcessorTests {
private static final String FIRST_NAME = "jane";
private static final String LAST_NAME = "doe";
@Test
public void testPersonProcessing() throws Exception {
Person person = new Person(FIRST_NAME, LAST_NAME);
ItemProcessor<Person, Person> personItemProcessor = new PersonItemProcessor();
Person transformedPerson = personItemProcessor.process(person);
assertNotNull("Received null Person", transformedPerson);
assertNotNull("Received null first name", transformedPerson.getFirstName());
assertNotNull("Received null last name", transformedPerson.getLastName());
assertEquals("Invalid first name processing, should be uppercase",
person.getFirstName().toUpperCase(), transformedPerson.getFirstName());
assertEquals("Invalid last name processing, should be uppercase",
person.getLastName().toUpperCase(), transformedPerson.getLastName());
}
}