Java 类org.elasticsearch.action.search.SearchScrollRequest 实例源码

@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    String scrollId = request.param("scroll_id");
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    searchScrollRequest.scrollId(scrollId);
    String scroll = request.param("scroll");
    if (scroll != null) {
        searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    request.withContentOrSourceParamParserOrNull(xContentParser -> {
        if (xContentParser != null) {
            // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
            try {
                buildFromContent(xContentParser, searchScrollRequest);
            } catch (IOException e) {
                throw new IllegalArgumentException("Failed to parse request body", e);
            }
        }});
    return channel -> client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<>(channel));
}

public static void buildFromContent(XContentParser parser, SearchScrollRequest searchScrollRequest) throws IOException {
    if (parser.nextToken() != XContentParser.Token.START_OBJECT) {
        throw new IllegalArgumentException("Malformed content, must start with an object");
    } else {
        XContentParser.Token token;
        String currentFieldName = null;
        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
            if (token == XContentParser.Token.FIELD_NAME) {
                currentFieldName = parser.currentName();
            } else if ("scroll_id".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                searchScrollRequest.scrollId(parser.text());
            } else if ("scroll".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                searchScrollRequest.scroll(new Scroll(TimeValue.parseTimeValue(parser.text(), null, "scroll")));
            } else {
                throw new IllegalArgumentException("Unknown parameter [" + currentFieldName
                        + "] in request body or parameter is of the wrong type[" + token + "] ");
            }
        }
    }
}

@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    String scrollId = request.param("scroll_id");
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    searchScrollRequest.scrollId(scrollId);
    String scroll = request.param("scroll");
    if (scroll != null) {
        searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    if (RestActions.hasBodyContent(request)) {
        XContentType type = XContentFactory.xContentType(RestActions.getRestContent(request));
        if (type == null) {
            if (scrollId == null) {
                scrollId = RestActions.getRestContent(request).toUtf8();
                searchScrollRequest.scrollId(scrollId);
            }
        } else {
            // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
            buildFromContent(RestActions.getRestContent(request), searchScrollRequest);
        }
    }
    client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<SearchResponse>(channel));
}

public static void buildFromContent(BytesReference content, SearchScrollRequest searchScrollRequest) {
    try (XContentParser parser = XContentHelper.createParser(content)) {
        if (parser.nextToken() != XContentParser.Token.START_OBJECT) {
            throw new IllegalArgumentException("Malforrmed content, must start with an object");
        } else {
            XContentParser.Token token;
            String currentFieldName = null;
            while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                if (token == XContentParser.Token.FIELD_NAME) {
                    currentFieldName = parser.currentName();
                } else if ("scroll_id".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                    searchScrollRequest.scrollId(parser.text());
                } else if ("scroll".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                    searchScrollRequest.scroll(new Scroll(TimeValue.parseTimeValue(parser.text(), null, "scroll")));
                } else {
                    throw new IllegalArgumentException("Unknown parameter [" + currentFieldName + "] in request body or parameter is of the wrong type[" + token + "] ");
                }
            }
        }
    } catch (IOException e) {
        throw new IllegalArgumentException("Failed to parse request body", e);
    }
}

@Override
protected void doStartNextScroll(String scrollId, TimeValue extraKeepAlive, Consumer<? super Response> onResponse) {
    searchWithRetry(listener -> {
        SearchScrollRequest request = new SearchScrollRequest();
        // Add the wait time into the scroll timeout so it won't timeout while we wait for throttling
        request.scrollId(scrollId).scroll(timeValueNanos(firstSearchRequest.scroll().keepAlive().nanos() + extraKeepAlive.nanos()));
        client.searchScroll(request, listener);
    }, r -> consume(r, onResponse));
}

public void testParseSearchScrollRequest() throws Exception {
    XContentParser content = createParser(XContentFactory.jsonBuilder()
        .startObject()
            .field("scroll_id", "SCROLL_ID")
            .field("scroll", "1m")
        .endObject());

    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    RestSearchScrollAction.buildFromContent(content, searchScrollRequest);

    assertThat(searchScrollRequest.scrollId(), equalTo("SCROLL_ID"));
    assertThat(searchScrollRequest.scroll().keepAlive(), equalTo(TimeValue.parseTimeValue("1m", null, "scroll")));
}

public void testParseSearchScrollRequestWithUnknownParamThrowsException() throws Exception {
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    XContentParser invalidContent = createParser(XContentFactory.jsonBuilder()
            .startObject()
                .field("scroll_id", "value_2")
                .field("unknown", "keyword")
            .endObject());

    Exception e = expectThrows(IllegalArgumentException.class,
            () -> RestSearchScrollAction.buildFromContent(invalidContent, searchScrollRequest));
    assertThat(e.getMessage(), startsWith("Unknown parameter [unknown]"));
}

public ShardFetchRequest(SearchScrollRequest request, long id, IntArrayList list, ScoreDoc lastEmittedDoc) {
    super(request);
    this.id = id;
    this.docIds = list.buffer;
    this.size = list.size();
    this.lastEmittedDoc = lastEmittedDoc;
}

@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    String scrollId = request.param("scroll_id");
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    searchScrollRequest.scrollId(scrollId);
    String scroll = request.param("scroll");
    if (scroll != null) {
        searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    if (RestActions.hasBodyContent(request)) {
        XContentType type = XContentFactory.xContentType(RestActions.getRestContent(request));
        if (type == null) {
            if (scrollId == null) {
                scrollId = RestActions.getRestContent(request).toUtf8();
                searchScrollRequest.scrollId(scrollId);
            }
        } else {
            // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
            RestSearchScrollAction.buildFromContent(RestActions.getRestContent(request), searchScrollRequest);
        }
    }

    final String sparseFormat = request.param("sparse_format", "dict");
    client.searchScroll(searchScrollRequest, new RestBuilderListener<SearchResponse>(channel) {
        @Override
        public RestResponse buildResponse(SearchResponse resp, XContentBuilder builder) throws Exception {
            SearchVectorizeResponse searchVectorizeResponse = new SearchVectorizeResponse(resp);
            searchVectorizeResponse.setFormat(sparseFormat);
            searchVectorizeResponse.toXContent(builder, ToXContent.EMPTY_PARAMS);
            return new BytesRestResponse(OK, builder);
        }
    });
}

public void execute(SearchScrollRequest request, final ActionListener<SearchResponse> listener) {
    logger.debug("Search scroll request {}", request);
    try {
        ActionRequestValidationException validation = request.validate();
        if(validation != null && !validation.validationErrors().isEmpty()) {
            throw validation;
        }

        RequestUriBuilder uriBuilder = new RequestUriBuilder()
                .addEndpoint("_search/scroll");

        if (request.scroll() != null) {
            uriBuilder.addQueryParameter("scroll", request.scroll().keepAlive().toString());
        }
        uriBuilder.addQueryParameter("scroll_id", request.scrollId());

        httpClient.getHttpClient().submit(HttpClientRequest.createGet(uriBuilder.toString()))
                .flatMap(ErrorHandler.AS_FUNC)
                .flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<SearchResponse>>() {
                    @Override
                    public Observable<SearchResponse> call(HttpClientResponse<ByteBuf> response) {
                        return response.getContent().flatMap(new Func1<ByteBuf, Observable<SearchResponse>>() {
                            @Override
                            public Observable<SearchResponse> call(ByteBuf byteBuf) {
                                return SearchResponse.parse(byteBuf);
                            }
                        });
                    }
                })
                .single()
                .subscribe(new ListenerCompleterObserver<>(listener));

    } catch (Exception e) {
        listener.onFailure(e);
    }
}

@Test
public void should_fail_when_scroll_id_is_missing() throws Exception {
    try {
        httpClient.searchScroll(new SearchScrollRequest()).get();
        fail();
    } catch (ExecutionException e) {
        Assertions.assertThat(e).hasCauseInstanceOf(ActionRequestValidationException.class);
        Assertions.assertThat(e.getCause()).hasMessageContaining("scrollId is missing");
    }
}

ActionResponse execute(final ActionRequest request) throws IOException {
    if (request instanceof BulkRequest) {
        return bulk((BulkRequest) request);
    } else if (request instanceof IndexRequest) {
        return index((IndexRequest) request);
    } else if (request instanceof DeleteRequest) {
        return delete((DeleteRequest) request);
    } else if (request instanceof ClearScrollRequest) {
        return clearScroll((ClearScrollRequest) request);
    } else if (request instanceof SearchScrollRequest) {
        return searchScroll((SearchScrollRequest) request);
    }

    return null;
}

public InternalScrollSearchRequest(SearchScrollRequest request, long id) {
    this.id = id;
    this.scroll = request.scroll();
}

@Override
public ActionFuture<SearchResponse> searchScroll(final SearchScrollRequest request) {
    return execute(SearchScrollAction.INSTANCE, request);
}

@Override
public void searchScroll(final SearchScrollRequest request, final ActionListener<SearchResponse> listener) {
    execute(SearchScrollAction.INSTANCE, request, listener);
}

public InternalScrollSearchRequest(SearchScrollRequest request, long id) {
    super(request);
    this.id = id;
    this.scroll = request.scroll();
}

@Override
public ActionFuture<SearchResponse> searchScroll(SearchScrollRequest request) {
    // TODO Auto-generated method stub
    return null;
}

@Override
public void searchScroll(SearchScrollRequest request, ActionListener<SearchResponse> listener) {
    // TODO Auto-generated method stub

}

public void searchScroll(SearchScrollRequest request, ActionListener<SearchResponse> listener) {
    searchScrollActionHandler.execute(request, listener);
}

public Future<SearchResponse> searchScroll(SearchScrollRequest request) {
    PlainActionFuture<SearchResponse> future = PlainActionFuture.newFuture();
    searchScroll(request, future);
    return future;
}

@Override
protected SearchResponse scrollSearchResponse(final String scrollId) throws IOException {
    return _client.searchScroll(new SearchScrollRequest(scrollId).scroll(
            AbstractElasticSearchDataContext.TIMEOUT_SCROLL));
}

public SearchScrollRequestBuilder(Client client, JsonToString<JsonInput> jsonToString, StringToJson<JsonOutput> stringToJson) {
    super(client, new SearchScrollRequest(), jsonToString, stringToJson);
}

@Override
protected ActionFuture<SearchResponse> doExecute(SearchScrollRequest request) {
    return client.searchScroll(request);
}

@Override
protected XContentBuilder toXContent(SearchScrollRequest request, SearchResponse response, XContentBuilder builder) throws IOException {
    return super.toXContent(request, response, builder).endObject();
}

/**
 * A search scroll request to continue searching a previous scrollable search request.
 *
 * @param request The search scroll request
 * @return The result future
 * @see Requests#searchScrollRequest(String)
 */
ActionFuture<SearchResponse> searchScroll(SearchScrollRequest request);

/**
 * A search scroll request to continue searching a previous scrollable search request.
 *
 * @param request  The search scroll request
 * @param listener A listener to be notified of the result
 * @see Requests#searchScrollRequest(String)
 */
void searchScroll(SearchScrollRequest request, ActionListener<SearchResponse> listener);

/**
 * Creates a search scroll request allowing to continue searching a previous search request.
 *
 * @param scrollId The scroll id representing the scrollable search
 * @return The search scroll request
 * @see org.elasticsearch.client.Client#searchScroll(org.elasticsearch.action.search.SearchScrollRequest)
 */
public static SearchScrollRequest searchScrollRequest(String scrollId) {
    return new SearchScrollRequest(scrollId);
}

/**
 * Creates a search scroll request allowing to continue searching a previous search request.
 *
 * @param scrollId The scroll id representing the scrollable search
 * @return The search scroll request
 * @see org.elasticsearch.client.Client#searchScroll(org.elasticsearch.action.search.SearchScrollRequest)
 */
public static SearchScrollRequest searchScrollRequest(String scrollId) {
    return new SearchScrollRequest(scrollId);
}