Add Unit Test and Docs for GcsStreamingMessageSource Comparator (spring-attic#2119)

Followup to spring-attic#2116; this adds an additional unit test and also mentions it in the docs.

Author: dzou

docs/src/main/asciidoc/spring-integration-storage.adoc

@@ -74,6 +74,8 @@ public MessageSource<InputStream> streamingAdapter(Storage gcs) {
 }
 ----
 
+If you would like to process the files in your bucket in a specific order, you may pass in a `Comparator<BlobInfo>` to the constructor `GcsStreamingMessageSource` to sort the files being processed.
+
 ==== Outbound channel adapter
 
 The outbound channel adapter allows files to be written to Google Cloud Storage.

spring-cloud-gcp-storage/src/main/java/org/springframework/cloud/gcp/storage/integration/inbound/GcsStreamingMessageSource.java

@@ -44,6 +44,14 @@ public GcsStreamingMessageSource(RemoteFileTemplate<BlobInfo> template) {
 		this(template, null);
 	}
 
+	/**
+	 * Creates a {@link GcsStreamingMessageSource} with a {@code comparator} which controls the order
+	 * that files are processed in.
+	 * @param template template making remote file calls to Google Cloud Storage
+	 * @param comparator defines the order that files should be processed based on {@link BlobInfo}.
+   *
+	 * @since 1.2
+	 */
 	public GcsStreamingMessageSource(RemoteFileTemplate<BlobInfo> template, Comparator<BlobInfo> comparator) {
 		super(template, comparator);
 		doSetFilter(new GcsPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "gcsStreamingMessageSource"));

spring-cloud-gcp-storage/src/test/java/org/springframework/cloud/gcp/storage/integration/inbound/GcsStreamingMessageSourceTests.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.gcp.storage.integration.inbound;
 
 import java.io.InputStream;
+import java.util.Comparator;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -60,24 +61,58 @@
 public class GcsStreamingMessageSourceTests {
 
 	@Autowired
-	private PollableChannel gcsChannel;
+	private PollableChannel unsortedChannel;
+
+	@Autowired
+	private PollableChannel sortedChannel;
 
 	@Test
 	public void testInboundStreamingChannelAdapter() {
-		Message<?> message = this.gcsChannel.receive(5000);
+		Message<?> message = this.unsortedChannel.receive(5000);
+		assertThat(message).isNotNull();
+		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("gamma");
+
+		message = this.unsortedChannel.receive(5000);
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("beta");
+		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
+
+		message = this.unsortedChannel.receive(5000);
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("alpha/alpha");
+		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
 
+		message = this.unsortedChannel.receive(10);
+		assertThat(message).isNull();
+	}
+
+	@Test
+	public void testSortedInboundChannelAdapter() {
+		// This uses the channel adapter with a custom comparator.
+		// Files will be processed in ascending order by name: alpha/alpha, beta, gamma
+		Message<?> message = this.sortedChannel.receive(5000);
 		assertThat(message).isNotNull();
 		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
-		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("folder1/gcsfilename");
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("alpha/alpha");
 
-		message = this.gcsChannel.receive(5000);
-		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("secondfilename");
+		message = this.sortedChannel.receive(5000);
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("beta");
 		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
 
-		message = this.gcsChannel.receive(10);
+		message = this.sortedChannel.receive(5000);
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("gamma");
+		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
+
+		message = this.sortedChannel.receive(10);
 		assertThat(message).isNull();
 	}
 
+	private static Blob createBlob(String bucket, String name) {
+		Blob blob = mock(Blob.class);
+		willAnswer((invocationOnMock) -> bucket).given(blob).getBucket();
+		willAnswer((invocationOnMock) -> name).given(blob).getName();
+		return blob;
+	}
+
 	/**
 	 * Spring config for the tests.
 	 */
@@ -87,35 +122,30 @@ public static class Config {
 
 		@Bean
 		public Storage gcsClient() {
-			Blob blob1 = mock(Blob.class);
-			Blob blob2 = mock(Blob.class);
-
-			willAnswer((invocationOnMock) -> "gcsbucket").given(blob1).getBucket();
-			willAnswer((invocationOnMock) -> "folder1/gcsfilename").given(blob1).getName();
-			willAnswer((invocationOnMock) -> "gcsbucket").given(blob2).getBucket();
-			willAnswer((invocationOnMock) -> "secondfilename").given(blob2).getName();
-
 			Storage gcs = mock(Storage.class);
 
 			willAnswer((invocationOnMock) ->
 				new PageImpl<>(null, null,
-						Stream.of(blob1, blob2)
+						Stream.of(
+								createBlob("gcsbucket", "gamma"),
+								createBlob("gcsbucket", "beta"),
+								createBlob("gcsbucket", "alpha/alpha"))
 								.collect(Collectors.toList())))
 					.given(gcs).list(eq("gcsbucket"));
 
-			ReadChannel channel1 = mock(ReadChannel.class);
-			ReadChannel channel2 = mock(ReadChannel.class);
-			willAnswer((invocationOnMock) -> channel1)
-					.given(gcs).reader(eq("gcsbucket"), eq("folder1/gcsfilename"));
-			willAnswer((invocationOnMock) -> channel2)
-					.given(gcs).reader(eq("gcsbucket"), eq("secondfilename"));
+			willAnswer((invocationOnMock) -> mock(ReadChannel.class))
+					.given(gcs).reader(eq("gcsbucket"), eq("alpha/alpha"));
+			willAnswer((invocationOnMock) -> mock(ReadChannel.class))
+					.given(gcs).reader(eq("gcsbucket"), eq("beta"));
+			willAnswer((invocationOnMock) -> mock(ReadChannel.class))
+					.given(gcs).reader(eq("gcsbucket"), eq("gamma"));
 
 			return gcs;
 		}
 
 		@Bean
-		@InboundChannelAdapter(value = "gcsChannel", poller = @Poller(fixedDelay = "100"))
-		public MessageSource<InputStream> adapter(Storage gcs) {
+		@InboundChannelAdapter(value = "unsortedChannel", poller = @Poller(fixedDelay = "100"))
+		public MessageSource<InputStream> unsortedChannelAdapter(Storage gcs) {
 			GcsStreamingMessageSource adapter =
 					new GcsStreamingMessageSource(new RemoteFileTemplate<>(new GcsSessionFactory(gcs)));
 			adapter.setRemoteDirectory("gcsbucket");
@@ -125,7 +155,26 @@ public MessageSource<InputStream> adapter(Storage gcs) {
 		}
 
 		@Bean
-		public PollableChannel gcsChannel() {
+		@InboundChannelAdapter(value = "sortedChannel", poller = @Poller(fixedDelay = "100"))
+		public MessageSource<InputStream> sortedChannelAdapter(Storage gcs) {
+			GcsStreamingMessageSource adapter =
+					new GcsStreamingMessageSource(
+							new RemoteFileTemplate<>(new GcsSessionFactory(gcs)),
+							Comparator.comparing(blob -> blob.getName()));
+
+			adapter.setRemoteDirectory("gcsbucket");
+			adapter.setFilter(new AcceptOnceFileListFilter<>());
+
+			return adapter;
+		}
+
+		@Bean
+		public PollableChannel unsortedChannel() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		public PollableChannel sortedChannel() {
 			return new QueueChannel();
 		}
 	}