DATAES-510 - Add tests for resource clean up.

Original Pull Request: #231

src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientUnitTests.java

@@ -20,8 +20,10 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
 import static org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Receive.*;
 
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import java.io.IOException;
 import java.net.URI;
 import java.util.Collections;
 
@@ -38,12 +40,14 @@ import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.VersionType;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.reactivestreams.Publisher;
 import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider;
 import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Receive;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
+import org.springframework.util.StreamUtils;
 
 /**
  * @author Christoph Strobl
@@ -572,4 +576,51 @@ public class ReactiveElasticsearchClientUnitTests {
 				.verifyComplete();
 	}
 
+	// --> SCROLL
+
+	@Test // DATAES-510
+	public void scrollShouldReadAll() throws IOException {
+
+		byte[] start = StreamUtils.copyToByteArray(Receive.fromPath("search-ok-scroll").getInputStream());
+		byte[] next = StreamUtils.copyToByteArray(Receive.fromPath("scroll_ok").getInputStream());
+		byte[] end = StreamUtils.copyToByteArray(Receive.fromPath("scroll_no_more_results").getInputStream());
+		byte[] cleanup = StreamUtils.copyToByteArray(Receive.fromPath("scroll_clean").getInputStream());
+
+		hostProvider.when(HOST) //
+				.receive(Receive::json) //
+				.receive(response -> Mockito.when(response.body(any())).thenReturn(Mono.just(start), Mono.just(next),
+						Mono.just(end), Mono.just(cleanup)));
+
+		client.scroll(new SearchRequest("twitter")) //
+				.as(StepVerifier::create) //
+				.expectNextCount(4) //
+				.verifyComplete();
+
+		hostProvider.when(HOST).receive(response -> {
+			verify(response, times(4)).body(any());
+		});
+	}
+
+	@Test // DATAES-510
+	public void scrollShouldCleanUpResourcesOnError() throws IOException {
+
+		byte[] start = StreamUtils.copyToByteArray(Receive.fromPath("search-ok-scroll").getInputStream());
+		byte[] error = StreamUtils.copyToByteArray(Receive.fromPath("scroll_error").getInputStream());
+		byte[] cleanup = StreamUtils.copyToByteArray(Receive.fromPath("scroll_clean").getInputStream());
+
+		hostProvider.when(HOST) //
+				.receive(Receive::json) //
+				.receive(response -> Mockito.when(response.body(any())).thenReturn(Mono.just(start), Mono.just(error),
+						Mono.just(cleanup)));
+
+		client.scroll(new SearchRequest("twitter")) //
+				.as(StepVerifier::create) //
+				.expectNextCount(2) //
+				.verifyError();
+
+		hostProvider.when(HOST).receive(response -> {
+			verify(response, times(3)).body(any());
+		});
+	}
+
 }

src/test/resources/org/springframework/data/elasticsearch/client/scroll_clean.json

@@ -0,0 +1,4 @@
+{
+  "num_freed": 1,
+  "succeeded": true
+}

src/test/resources/org/springframework/data/elasticsearch/client/scroll_error.json

@@ -0,0 +1,77 @@
+{
+  "error": {
+    "caused_by": {
+      "reason": "No search context found for id [1]",
+      "type": "search_context_missing_exception"
+    },
+    "failed_shards": [
+      {
+        "index": null,
+        "reason": {
+          "reason": "No search context found for id [1]",
+          "type": "search_context_missing_exception"
+        },
+        "shard": -1
+      },
+      {
+        "index": null,
+        "reason": {
+          "reason": "No search context found for id [2]",
+          "type": "search_context_missing_exception"
+        },
+        "shard": -1
+      },
+      {
+        "index": null,
+        "reason": {
+          "reason": "No search context found for id [3]",
+          "type": "search_context_missing_exception"
+        },
+        "shard": -1
+      },
+      {
+        "index": null,
+        "reason": {
+          "reason": "No search context found for id [4]",
+          "type": "search_context_missing_exception"
+        },
+        "shard": -1
+      },
+      {
+        "index": null,
+        "reason": {
+          "reason": "No search context found for id [5]",
+          "type": "search_context_missing_exception"
+        },
+        "shard": -1
+      }
+    ],
+    "grouped": true,
+    "phase": "query",
+    "reason": "all shards failed",
+    "root_cause": [
+      {
+        "reason": "No search context found for id [1]",
+        "type": "search_context_missing_exception"
+      },
+      {
+        "reason": "No search context found for id [2]",
+        "type": "search_context_missing_exception"
+      },
+      {
+        "reason": "No search context found for id [3]",
+        "type": "search_context_missing_exception"
+      },
+      {
+        "reason": "No search context found for id [4]",
+        "type": "search_context_missing_exception"
+      },
+      {
+        "reason": "No search context found for id [5]",
+        "type": "search_context_missing_exception"
+      }
+    ],
+    "type": "search_phase_execution_exception"
+  },
+  "status": 404
+}

src/test/resources/org/springframework/data/elasticsearch/client/scroll_no_more_results.json

@@ -0,0 +1,16 @@
+{
+  "_shards": {
+    "failed": 0,
+    "skipped": 0,
+    "successful": 5,
+    "total": 5
+  },
+  "hits": {
+    "hits": [],
+    "max_score": 1.0,
+    "total": 100
+  },
+  "terminated_early": true,
+  "timed_out": false,
+  "took": 1
+}

src/test/resources/org/springframework/data/elasticsearch/client/scroll_ok.json

@@ -0,0 +1,39 @@
+{
+  "_shards": {
+    "failed": 0,
+    "skipped": 0,
+    "successful": 5,
+    "total": 5
+  },
+  "hits": {
+    "hits": [
+      {
+        "_index": "twitter",
+        "_type": "doc",
+        "_id": "2",
+        "_score": 0.2876821,
+        "_source": {
+          "user": "kimchy",
+          "post_date": "2009-11-15T14:12:12",
+          "message": "Another tweet, will it be indexed?"
+        }
+      },
+      {
+        "_index": "twitter",
+        "_type": "doc",
+        "_id": "1",
+        "_score": 0.2876821,
+        "_source": {
+          "user": "kimchy",
+          "post_date": "2009-11-15T13:12:00",
+          "message": "Trying out Elasticsearch, so far so good?"
+        }
+      }
+    ],
+    "max_score": 1.0,
+    "total": 100
+  },
+  "terminated_early": true,
+  "timed_out": false,
+  "took": 1
+}

src/test/resources/org/springframework/data/elasticsearch/client/search-ok-scroll.json

@@ -0,0 +1,39 @@
+{
+  "took": 52,
+  "timed_out": false,
+  "_scroll_id": "DnF1ZXJ5VGhlbkZldGNoBQAAAAAAAAAHFndhSE1uNUlLUXhXb1ZvQTNqOHNrMWcAAAAAAAAABhZ3YUhNbjVJS1F4V29Wb0EzajhzazFnAAAAAAAAAAgWd2FITW41SUtReFdvVm9BM2o4c2sxZwAAAAAAAAAJFndhSE1uNUlLUXhXb1ZvQTNqOHNrMWcAAAAAAAAAChZ3YUhNbjVJS1F4V29Wb0EzajhzazFn",
+  "_shards": {
+    "total": 5,
+    "successful": 5,
+    "skipped": 0,
+    "failed": 0
+  },
+  "hits": {
+    "total": 100,
+    "max_score": 0.2876821,
+    "hits": [
+      {
+        "_index": "twitter",
+        "_type": "doc",
+        "_id": "2",
+        "_score": 0.2876821,
+        "_source": {
+          "user": "kimchy",
+          "post_date": "2009-11-15T14:12:12",
+          "message": "Another tweet, will it be indexed?"
+        }
+      },
+      {
+        "_index": "twitter",
+        "_type": "doc",
+        "_id": "1",
+        "_score": 0.2876821,
+        "_source": {
+          "user": "kimchy",
+          "post_date": "2009-11-15T13:12:00",
+          "message": "Trying out Elasticsearch, so far so good?"
+        }
+      }
+    ]
+  }
+}