GH-3373: Support IPV6 in AbstractInboundFileSynch

Fixes #3373

The `AbstractInboundFileSynchronizer` doesn't consider that
`hostPort` from `Session` could be in an IPv6 syntax

* Parse the `hostPort` from `Session` in a manner that only the last
`:` is treated as a port delimiter

**Cherry-pick to 5.3.x & 5.2.x**

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -493,10 +493,13 @@ protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, F remoteF
 				if (this.preserveTimestamp && !localFile.setLastModified(modified)) {
 					throw new IllegalStateException("Could not sent last modified on file: " + localFile);
 				}
-				String[] hostPort = session.getHostPort().split(":");
+				String hostPort = session.getHostPort();
+				int colonIndex = hostPort.lastIndexOf(":");
+				String host = hostPort.substring(0, colonIndex);
+				String port = hostPort.substring(colonIndex + 1);
 				try {
 					String remoteFileMetadata =
-							new URI(protocol(), null, hostPort[0], Integer.parseInt(hostPort[1]),
+							new URI(protocol(), null, host, Integer.parseInt(port),
 									'/' + remoteDirectoryPath, null, remoteFileName)
 									.toString();
 					this.remoteFileMetadataStore.put(buildMetadataKey(localFile), remoteFileMetadata);

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.sftp.inbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
@@ -25,10 +26,12 @@
 
 import java.io.File;
 import java.io.FileInputStream;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Vector;
 
 import org.junit.jupiter.api.AfterEach;
@@ -65,11 +68,14 @@
  */
 public class SftpInboundRemoteFileSystemSynchronizerTests {
 
-	private static com.jcraft.jsch.Session jschSession = mock(com.jcraft.jsch.Session.class);
+	private static final com.jcraft.jsch.Session jschSession = mock(com.jcraft.jsch.Session.class);
 
 	@BeforeEach
 	@AfterEach
 	public void cleanup() {
+		willReturn("::1")
+				.given(jschSession)
+				.getHost();
 		File file = new File("test");
 		if (file.exists()) {
 			String[] files = file.list();
@@ -103,7 +109,7 @@ public void testCopyFileToLocalDir() throws Exception {
 		List<FileListFilter<LsEntry>> filters = new ArrayList<>();
 		filters.add(persistFilter);
 		filters.add(patternFilter);
-		CompositeFileListFilter<LsEntry> filter = new CompositeFileListFilter<LsEntry>(filters);
+		CompositeFileListFilter<LsEntry> filter = new CompositeFileListFilter<>(filters);
 		synchronizer.setFilter(filter);
 		synchronizer.setBeanFactory(mock(BeanFactory.class));
 		synchronizer.afterPropertiesSet();
@@ -129,6 +135,13 @@ public void testCopyFileToLocalDir() throws Exception {
 		assertThat(atestFile.getHeaders())
 				.containsKeys(FileHeaders.REMOTE_HOST_PORT, FileHeaders.REMOTE_DIRECTORY, FileHeaders.REMOTE_FILE);
 
+		@SuppressWarnings("unchecked")
+		Map<String, String> remoteFileMetadataStore =
+				TestUtils.getPropertyValue(synchronizer, "remoteFileMetadataStore.metadata", Map.class);
+
+		String next = remoteFileMetadataStore.values().iterator().next();
+		assertThat(URI.create(next).getHost()).isEqualTo("[::1]");
+
 		Message<File> btestFile = ms.receive();
 		assertThat(btestFile).isNotNull();
 		assertThat(btestFile.getPayload().getName()).isEqualTo("b.test");