Я пытаюсь использовать API эластичного поиска, используя Spring webflux, чтобы моя конечная точка API не блокировалась.

Question

Я создаю конечную точку в Spring boot 2 и использую Spring webflux. В этой конечной точке я буду брать широту и долготу у вызывающего абонента и возвращать состояние на основании этого. Чтобы получить состояние, я вызываю API эластичного поиска для получения данных.

Я могу получить ответ от API поиска Elastic, как показано ниже:

{
  "took": 11,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "failed": 0
  },
  "hits": {
    "total": 117252,
    "max_score": null,
    "hits": [
      {
        "_index": "geolocation",
        "_type": "geolocationdata",
        "_id": "AWt0m6GJqkN7DgSP9Lsd",
        "_score": null,
        "_source": {
          "network": "117.254.200.0/22",
          "geonameId": 1262062,
          "registeredCountrygeonameId": 1269750,
          "representedCountrygeonameId": "",
          "postalCode": "370655",
          "location": "23.2667,68.8333",
          "accuracyRadius": 100,
          "localecode": "en",
          "continentcode": "AS",
          "continentname": "Asia",
          "countryisocode": "IN",
          "countryname": "India",
          "subdivision1isocode": "GJ",
          "subdivision1nname": "Gujarat",
          "subdivision2isocode": "",
          "subdivision2nname": "",
          "cityName": "Naliya",
          "metroCode": "",
          "timeZone": "Asia/Kolkata"
        },
        "sort": [
          6986.775031169917
        ]
      }
    ]
  }
}

Получив этот JSON, я хотел бы извлечь из него только необходимые поля и построить модель, необходимую для моего API, которая будет возвращена вызывающей стороне.

Вот как я использую API-интерфейс Elastic search и получаю результат

private WebClient webClient;

@PostConstruct
public void init() {
    this.webClient = WebClient.builder()
        .baseUrl("http://172.24.5.162:9200/geolocation")
        .defaultHeader(
            HttpHeaders.CONTENT_TYPE, 
            MediaType.APPLICATION_JSON_VALUE)
        .build();
}

public String getGeoname(String latitude, String longitude) throws Exception {
    try {
        String req = "{\"from\": 0,\"size\": 1,\"sort\": {\"_geo_distance\": {\"location\": {\"lat\": " + latitude
                + ",\"lon\": " + longitude
                + "},\"order\": \"asc\",\"unit\": \"km\",\"distance_type\": \"plane\"}}}";

        final String test;

        //result from Elastic search API
        Mono<String> result = webClient.post()
                                 .uri("/_search")
                                 .body(Mono.just(req), String.class)
                                 .retrieve().bodyToMono(String.class);

    } catch (Exception ex) {
        log.error("Exception while sending request to Elastic search Lat: " + latitude + " Long: " + longitude, ex);
        return gson.toJson(new ErrorModel(ErrorCodes.BAD_INPUT, "Bad Input"));
    }
    return "";
}

В переменной результата у меня есть JSON, как показано выше, как Mono. Если я буду использовать метод block () для переменной результата, чтобы получить строку, являющуюся JSON IWant, тогда он заблокирует основной поток и станет блокировать. Мое требование - использовать этот Mono, чтобы я мог выполнять операции, как показано ниже (в основном я строю GeoLocation для моей модели)

String hits = "";
JSONObject jsonObject = new JSONObject(o);

if (jsonObject.has("hits") && jsonObject.getJSONObject("hits").has("hits")) {
    hits = jsonObject.getJSONObject("hits")
        .getString("hits");

    hits = hits.substring(1);

    JSONObject hitsJson = new JSONObject(hits);
    JSONObject source = new JSONObject();
    if (hitsJson.has("_source")) {
        source = hitsJson.getJSONObject("_source");
        GeoLocation geolocation = new GeoLocation(source.getString("continentname"),
        source.getString("countryname"), 
        source.getString("subdivision1nname"),
        source.getString("cityName"));

        geoLocationResponse = Mono.just(gson.toJson(geolocation));

Как мне выполнить вышеуказанную операцию неблокирующим способом и вернуть результат моему абоненту конечной точки? Я думаю вернуть Mono из моего RestController

Answer 1

Прежде всего, вы не используете try catch в реактивном мире, мы создаем Mono.error, а затем воздействуем на эти ошибки позже в цепочке сигналов.

Вы действительно должны проверить некоторые учебники илиопробуйте основные принципы работы в проектной документации реактора.Это вам очень поможет.

Вы должны использовать Джексон, который поставляется в комплекте с весенней загрузкой, вместо использования JSONObject и Gson.

public Mono<GeoLocation> getGeoname(String latitude, String longitude) {
    final String req = "{\"from\": 0,\"size\": 1,\"sort\": {\"_geo_distance\": {\"location\": {\"lat\": " + latitude
                + ",\"lon\": " + longitude
                + "},\"order\": \"asc\",\"unit\": \"km\",\"distance_type\": \"plane\"}}}";

    return webClient.post()
            .uri("/_search")
            .body(Mono.just(req), String.class)
            .exchange()
            .flatMap(clientResponse -> {
                // Check status if not 200 create a mono error that we will act upon later
                if(clientResponse.rawStatusCode() != 200)
                    return Mono.error(RuntimeException::new);
                return clientResponse.bodyToMono(String.class);
            }).map(body -> {

                // if you want to work with strings then do your mapping here
                // otherwise replace bodyToMono(FooBar.class) the FooBar class with a              
                // representation of the returned body and jackson will map to it.

                return new GeoLocation();
            }).doOnError(throwable -> {
                // handle the mono error we created earlier here, and throw an exception.
                // Then handle the exception later in a global exception handler
                throw new RuntimeException(throwable);
            });
}