INT-4233: Add (S)FTP FileInfo Header (Streaming)

JIRA: https://jira.spring.io/browse/INT-4233

Add the complete file info as JSON (when Jackson or Boon available) to the message headers
when streaming inbound.

Provide a mechanism to configure Boon to provide similar output to Jackson.

Also allow subclasses to provide their own object mapper.

Also clean up after the AMQP DSL tests, and don't use a queue `foo`. I often have such a queue
with content; this caused tests to fail.

spring-integration-core/src/main/java/org/springframework/integration/support/json/BoonJsonObjectMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -28,6 +28,7 @@
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.Executors;
+import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -37,6 +38,7 @@
 import org.boon.json.JsonSerializerFactory;
 import org.boon.json.JsonSlurper;
 import org.boon.json.ObjectMapper;
+import org.boon.json.implementation.ObjectMapperImpl;
 
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.integration.mapping.support.JsonHeaders;
@@ -46,6 +48,7 @@
  * The Boon (@link https://github.com/RichardHightower/boon) {@link JsonObjectMapper} implementation.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  * @since 4.1
  */
 public class BoonJsonObjectMapper extends JsonObjectMapperAdapter<Map<String, Object>, Object>
@@ -67,6 +70,18 @@ public BoonJsonObjectMapper() {
 		this.objectMapper = JsonFactory.create();
 	}
 
+	public BoonJsonObjectMapper(Consumer<JsonParserFactory> jpfConfig, Consumer<JsonSerializerFactory> jsfConfig) {
+		JsonParserFactory jpf = new JsonParserFactory();
+		if (jpfConfig != null) {
+			jpfConfig.accept(jpf);
+		}
+		JsonSerializerFactory jsf = new JsonSerializerFactory();
+		if (jsfConfig != null) {
+			jsfConfig.accept(jsf);
+		}
+		this.objectMapper = new ObjectMapperImpl(jpf, jsf);
+	}
+
 	public BoonJsonObjectMapper(JsonParserFactory parserFactory, JsonSerializerFactory serializerFactory) {
 		this.objectMapper = JsonFactory.create(parserFactory, serializerFactory);
 	}

spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -38,16 +38,8 @@ private JacksonJsonUtils() {
 			ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper", classLoader) &&
 					ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator", classLoader);
 
-	private static final boolean jacksonPresent =
-			ClassUtils.isPresent("org.codehaus.jackson.map.ObjectMapper", classLoader) &&
-					ClassUtils.isPresent("org.codehaus.jackson.JsonGenerator", classLoader);
-
 	public static boolean isJackson2Present() {
 		return jackson2Present;
 	}
 
-	public static boolean isJacksonPresent() {
-		return jacksonPresent;
-	}
-
 }

spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperProvider.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -19,6 +19,13 @@
 
 import org.springframework.util.ClassUtils;
 
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 /**
  * Simple factory to provide {@linkplain JsonObjectMapper}
  * instances dependently of jackson-databind or boon libs in the classpath.
@@ -60,6 +67,25 @@ else if (boonPresent) {
 		}
 	}
 
+	/**
+	 * Return an object mapper builder if available.
+	 * @param preferBoon true to prefer boon if available.
+	 * @return the mapper builder.
+	 * @throws IllegalStateException if an implementation is not available.
+	 * @since 5.0
+	 */
+	public static JsonObjectMapperBuilder<?> newInstanceBuilder(boolean preferBoon) {
+		if (JacksonJsonUtils.isJackson2Present() && (!preferBoon || !boonPresent)) {
+			return new JacksonJsonObjectMapperBuilder();
+		}
+		else if (boonPresent) {
+			return new BoonJsonObjectMapperBuilder();
+		}
+		else {
+			throw new IllegalStateException("Neither jackson-databind.jar, nor boon.jar is present in the classpath.");
+		}
+	}
+
 	/**
 	 * Returns true if a supported JSON implementation is on the class path.
 	 * @return true if {@link #newInstance()} will return a mapper.
@@ -69,4 +95,76 @@ public static boolean jsonAvailable() {
 		return JacksonJsonUtils.isJackson2Present() || boonPresent;
 	}
 
+	public static abstract class JsonObjectMapperBuilder<B extends JsonObjectMapperBuilder<B>> {
+
+		protected boolean usePropertyOnly = true; // NOSONAR
+
+		protected boolean includeAllValues = true; // NOSONAR
+
+		public B usePropertyOnly(boolean use) {
+			this.usePropertyOnly = use;
+			return _this();
+		}
+
+		public B includeAllValues(boolean include) {
+			this.includeAllValues = include;
+			return _this();
+		}
+
+		@SuppressWarnings("unchecked")
+		protected final B _this() {
+			return (B) this;
+		}
+
+		public abstract JsonObjectMapper<?, ?> build();
+
+	}
+
+	private static class JacksonJsonObjectMapperBuilder
+			extends JsonObjectMapperBuilder<JacksonJsonObjectMapperBuilder> {
+
+		JacksonJsonObjectMapperBuilder() {
+			super();
+		}
+
+		@Override
+		public JsonObjectMapper<?, ?> build() {
+			ObjectMapper objectMapper = new ObjectMapper();
+			objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
+			objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+			if (this.usePropertyOnly) {
+				objectMapper.setVisibility(PropertyAccessor.GETTER, Visibility.PUBLIC_ONLY);
+				objectMapper.setVisibility(PropertyAccessor.FIELD, Visibility.NONE);
+			}
+			if (this.includeAllValues) {
+				objectMapper.setSerializationInclusion(JsonInclude.Include.ALWAYS);
+			}
+			return new Jackson2JsonObjectMapper(objectMapper);
+		}
+
+	}
+
+	private static class BoonJsonObjectMapperBuilder extends JsonObjectMapperBuilder<BoonJsonObjectMapperBuilder> {
+
+		BoonJsonObjectMapperBuilder() {
+			super();
+		}
+
+		@Override
+		public JsonObjectMapper<?, ?> build() {
+			return new BoonJsonObjectMapper(null, f -> {
+				f.useAnnotations();
+				if (this.usePropertyOnly) {
+					f.usePropertyOnly();
+				}
+				if (this.includeAllValues) {
+					f.includeDefaultValues()
+						.includeNulls()
+						.includeEmpty();
+				}
+			});
+		}
+
+	}
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -47,4 +47,10 @@ public abstract class FileHeaders {
 	 */
 	public static final String MARKER = PREFIX + "marker";
 
+	/**
+	 * JSON representation of remote file information (if a JSON object mapper is
+	 * available).
+	 */
+	public static final String REMOTE_FILE_INFO = PREFIX + "remoteFileInfo";
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -38,6 +38,10 @@
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.filters.ReversibleFileListFilter;
 import org.springframework.integration.file.remote.session.Session;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.integration.support.json.JacksonJsonUtils;
+import org.springframework.integration.support.json.JsonObjectMapper;
+import org.springframework.integration.support.json.JsonObjectMapperProvider;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 
@@ -58,6 +62,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 
 	private final Comparator<AbstractFileInfo<F>> comparator;
 
+	private final JsonObjectMapper<?, ?> objectMapper;
 	/**
 	 * the path on the remote server.
 	 */
@@ -74,6 +79,12 @@ protected AbstractRemoteFileStreamingMessageSource(RemoteFileTemplate<F> templat
 			Comparator<AbstractFileInfo<F>> comparator) {
 		this.remoteFileTemplate = template;
 		this.comparator = comparator;
+		if (JacksonJsonUtils.isJackson2Present()) {
+			this.objectMapper = JsonObjectMapperProvider.newInstanceBuilder(false).build();
+		}
+		else {
+			this.objectMapper = null;
+		}
 	}
 
 	/**
@@ -116,6 +127,16 @@ protected RemoteFileTemplate<F> getRemoteFileTemplate() {
 		return this.remoteFileTemplate;
 	}
 
+	/**
+	 * Override this method if you wish to provide your own object mapper for
+	 * the {@link AbstractFileInfo} header.
+	 * @return the object mapper.
+	 * @since 5.0
+	 */
+	protected JsonObjectMapper<?, ?> getObjectMapper() {
+		return this.objectMapper;
+	}
+
 	@Override
 	public final void afterPropertiesSet() {
 		Assert.state(this.remoteDirectoryExpression != null, "'remoteDirectoryExpression' must not be null");
@@ -136,11 +157,20 @@ protected Object doReceive() {
 			String remotePath = remotePath(file);
 			Session<?> session = this.remoteFileTemplate.getSession();
 			try {
-				return getMessageBuilderFactory().withPayload(session.readRaw(remotePath))
+				AbstractIntegrationMessageBuilder<InputStream> builder = getMessageBuilderFactory()
+						.withPayload(session.readRaw(remotePath))
 						.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
 						.setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
-						.setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
-						.build();
+						.setHeader(FileHeaders.REMOTE_FILE, file.getFilename());
+				if (getObjectMapper() != null) {
+					try {
+						builder.setHeader(FileHeaders.REMOTE_FILE_INFO, getObjectMapper().toJson(file));
+					}
+					catch (Exception e) {
+						logger.info("Failed to transform file info to json: " + file, e);
+					}
+				}
+				return builder.build();
 			}
 			catch (IOException e) {
 				throw new MessagingException("IOException when retrieving " + remotePath, e);

spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java

@@ -16,8 +16,10 @@
 
 package org.springframework.integration.file.remote;
 
+import static org.hamcrest.CoreMatchers.containsString;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.BDDMockito.willThrow;
@@ -37,6 +39,7 @@
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
@@ -45,6 +48,7 @@
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.splitter.FileSplitter;
+import org.springframework.integration.support.json.JsonObjectMapperProvider;
 import org.springframework.integration.transformer.StreamTransformer;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
@@ -73,14 +77,37 @@ public void testAllData() throws Exception {
 		assertEquals("foo\nbar", new String(received.getPayload()));
 		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
 		assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+		String fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO);
+		assertThat(fileInfo, containsString("remoteDirectory\":\"/foo"));
+		assertThat(fileInfo, containsString("permissions\":\"-rw-rw-rw"));
+		assertThat(fileInfo, containsString("size\":42"));
+		assertThat(fileInfo, containsString("directory\":false"));
+		assertThat(fileInfo, containsString("filename\":\"foo"));
+		assertThat(fileInfo, containsString("modified\":42000"));
+		assertThat(fileInfo, containsString("link\":false"));
+		assertThat(fileInfo, containsString("fileInfo"));
+		assertThat(fileInfo, containsString("name=/foo/foo"));
 
 		// close after list, transform
 		verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(2)).close();
 
+		new DirectFieldAccessor(streamer).setPropertyValue("objectMapper",
+				JsonObjectMapperProvider.newInstanceBuilder(true).build());
+
 		received = (Message<byte[]>) this.transformer.transform(streamer.receive());
 		assertEquals("baz\nqux", new String(received.getPayload()));
 		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
 		assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+		fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO);
+		assertThat(fileInfo, containsString("remoteDirectory\":\"/foo"));
+		assertThat(fileInfo, containsString("permissions\":\"-rw-rw-rw"));
+		assertThat(fileInfo, containsString("size\":42"));
+		assertThat(fileInfo, containsString("directory\":false")); // Boon doesn't emit default values (false)
+		assertThat(fileInfo, containsString("filename\":\"bar"));
+		assertThat(fileInfo, containsString("modified\":42000"));
+		assertThat(fileInfo, containsString("link\":false")); // Boon doesn't emit default values (false)
+		assertThat(fileInfo, containsString("fileInfo"));
+		assertThat(fileInfo, containsString("name=/foo/bar"));
 
 		// close after transform
 		verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(3)).close();
@@ -213,12 +240,12 @@ public boolean isLink() {
 
 		@Override
 		public long getSize() {
-			return 0;
+			return 42;
 		}
 
 		@Override
 		public long getModified() {
-			return 0;
+			return 42_000;
 		}
 
 		@Override
@@ -228,12 +255,16 @@ public String getFilename() {
 
 		@Override
 		public String getPermissions() {
-			return null;
+			return "-rw-rw-rw";
 		}
 
 		@Override
 		public String getFileInfo() {
-			return null;
+			return asString();
+		}
+
+		private String asString() {
+			return "StringFileInfo [name=" + this.name + "]";
 		}
 
 	}