-
Notifications
You must be signed in to change notification settings - Fork 22
/
elastic_client.cpp
130 lines (107 loc) · 5.43 KB
/
elastic_client.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
#include <cpr/response.h>
#include <fc/io/json.hpp>
#include <fc/log/logger.hpp>
#include <boost/format.hpp>
#include <eosio/chain/exceptions.hpp>
#include "elastic_client.hpp"
#include "exceptions.hpp"
namespace eosio
{
namespace
{
bool is_2xx(int32_t status_code)
{
return status_code > 199 && status_code < 300;
}
} // namespace
bool elastic_client::head(const std::string &url_path)
{
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::HEAD, url_path, "");
if ( resp.status_code == 200 ) {
return true;
} else if ( resp.status_code == 404 ) {
return false;
} else {
EOS_THROW(chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
}
}
bool elastic_client::doc_exist(const std::string &index_name, const std::string &id)
{
auto url = boost::str(boost::format("%1%/_doc/%2%") % index_name % id );
return head(url);
}
void elastic_client::index(const std::string &index_name, const std::string &body, const std::string &id)
{
cpr::Response resp = client.index(index_name, "_doc", id, body);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
}
uint32_t elastic_client::create(const std::string &index_name, const std::string &body, const std::string &id)
{
auto url = boost::str(boost::format("%1%/_doc/%2%/_create") % index_name % id );
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::PUT, url, body);
if ( (!is_2xx(resp.status_code)) && (resp.status_code != 409) )
EOS_THROW(chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
return resp.status_code;
}
void elastic_client::init_index(const std::string &index_name, const std::string &mappings)
{
if ( !head(index_name) ) {
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::PUT, index_name, mappings);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
}
}
void elastic_client::delete_index(const std::string &index_name)
{
// retrn status code 404 if index not exists
client.performRequest(elasticlient::Client::HTTPMethod::DELETE, index_name, "");
}
uint64_t elastic_client::count_doc(const std::string &index_name, const std::string &query)
{
auto url = boost::str(boost::format("%1%/_doc/_count") % index_name );
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::GET, url, query);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
auto v = fc::json::from_string(resp.text);
return v["count"].as_uint64();
}
void elastic_client::get(const std::string &index_name, const std::string &id, fc::variant &res)
{
cpr::Response resp = client.get(index_name, "_doc", id);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
res = fc::json::from_string(resp.text);
}
void elastic_client::search(const std::string &index_name, fc::variant &v, const std::string &query)
{
cpr::Response resp = client.search(index_name, "_doc", query);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
v = fc::json::from_string(resp.text);
}
void elastic_client::delete_by_query(const std::string &index_name, const std::string &query)
{
auto url = boost::str(boost::format("%1%/_doc/_delete_by_query") % index_name );
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::POST, url, query);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
}
void elastic_client::bulk_perform(elasticlient::SameIndexBulkData &bulk)
{
auto index_name = bulk.indexName();
auto body = bulk.body();
auto url = boost::str(boost::format("%1%/_bulk") % index_name);
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::POST, url, body);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
fc::variant text_doc( fc::json::from_string(resp.text) );
EOS_ASSERT(text_doc["errors"].as_bool() == false, chain::bulk_fail_exception, "bulk perform errors: ${text}", ("text", resp.text));
}
void elastic_client::bulk_perform(const std::string &bulk)
{
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::POST, "_bulk", bulk);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
fc::variant text_doc( fc::json::from_string(resp.text) );
EOS_ASSERT(text_doc["errors"].as_bool() == false, chain::bulk_fail_exception, "bulk perform errors: ${text}", ("text", resp.text));
}
void elastic_client::update(const std::string &index_name, const std::string &id, const std::string &body)
{
auto url = boost::str(boost::format("%1%/_doc/%2%/_update") % index_name % id);
cpr::Response resp = client.performRequest(elasticlient::Client::HTTPMethod::POST, url, body);
EOS_ASSERT(is_2xx(resp.status_code), chain::response_code_exception, "${code} ${text}", ("code", resp.status_code)("text", resp.text));
}
} // namespace eosio