SearchProxyController { @RequestMapping("/_search") public ResponseEntity<InputStreamResource> search(@RequestParam("q") String query, @RequestHeader(value = "X-ElasticSearch-Host", required = false) String elasticSearchHost) throws IOException { HttpHost httpHost = null; Collection<Task> tasks = scheduler.getTasks().values(); Stream<HttpHost> httpHostStream = tasks.stream().map(task -> new HttpHost(task.getHostname(), task.getClientAddress().getPort())); if (elasticSearchHost != null) { httpHost = httpHostStream.filter(host -> host.toHostString().equalsIgnoreCase(elasticSearchHost)).findAny().get(); } else { httpHost = httpHostStream.skip(RandomUtils.nextInt(0, tasks.size())).findAny().get(); } HttpResponse esSearchResponse = httpClient.execute(httpHost, new HttpGet("/_search?q=" + URLEncoder.encode(query, "UTF-8"))); InputStreamResource inputStreamResource = new InputStreamResource(esSearchResponse.getEntity().getContent()); return ResponseEntity.ok() .contentLength(esSearchResponse.getEntity().getContentLength()) .contentType(MediaType.APPLICATION_JSON) .header("X-ElasticSearch-host", httpHost.toHostString()) .body(inputStreamResource); }  @RequestMapping("/_cluster/stats") ResponseEntity<InputStreamResource> stats(); @RequestMapping("/_search") ResponseEntity<InputStreamResource> search(@RequestParam("q") String query, @RequestHeader(value = "X-ElasticSearch-Host", required = false) String elasticSearchHost);  }
@Test public void willForwardSearchRequestToARandomNode() throws Exception { when(elasticsearchScheduler.getTasks()).thenReturn(createTasksMap(3)); when(httpClient.execute(any(HttpHost.class), any(HttpRequest.class))).thenReturn(httpResponse); when(httpResponse.getEntity()).thenReturn(new InputStreamEntity(new ByteArrayInputStream("Search result".getBytes("UTF-8")))); final ResponseEntity<InputStreamResource> search = controller.search("test", null); assertEquals(200, search.getStatusCode().value()); final ArgumentCaptor<HttpHost> httpHostArgumentCaptor = ArgumentCaptor.forClass(HttpHost.class); final ArgumentCaptor<HttpRequest> httpRequestArgumentCaptor = ArgumentCaptor.forClass(HttpRequest.class); verify(httpClient).execute(httpHostArgumentCaptor.capture(), httpRequestArgumentCaptor.capture()); assertEquals(HOSTNAME, httpHostArgumentCaptor.getValue().getHostName()); assertEquals("/_search?q=test", httpRequestArgumentCaptor.getValue().getRequestLine().getUri()); }
@Test public void willForwardSearchRequestToAChosenNode() throws Exception { final String chosenNode = "1.0.0.1:1002"; when(elasticsearchScheduler.getTasks()).thenReturn(createTasksMap(3)); when(httpClient.execute(any(HttpHost.class), any(HttpRequest.class))).thenReturn(httpResponse); when(httpResponse.getEntity()).thenReturn(new InputStreamEntity(new ByteArrayInputStream("Search result".getBytes("UTF-8")))); final ResponseEntity<InputStreamResource> search = controller.search("test", chosenNode); assertEquals(200, search.getStatusCode().value()); assertEquals(chosenNode, search.getHeaders().getFirst("X-ElasticSearch-host")); final ArgumentCaptor<HttpHost> httpHostArgumentCaptor = ArgumentCaptor.forClass(HttpHost.class); final ArgumentCaptor<HttpRequest> httpRequestArgumentCaptor = ArgumentCaptor.forClass(HttpRequest.class); verify(httpClient).execute(httpHostArgumentCaptor.capture(), httpRequestArgumentCaptor.capture()); assertEquals(HOSTNAME, httpHostArgumentCaptor.getValue().getHostName()); assertEquals("/_search?q=test", httpRequestArgumentCaptor.getValue().getRequestLine().getUri()); }