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
|
/*
* Licensed to Elasticsearch under one or more contributor
* license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright
* ownership. Elasticsearch licenses this file to you under
* the Apache License, Version 2.0 (the "License"); you may
* not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.rest.action.document;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.bulk.BulkShardRequest;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.node.NodeClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.logging.DeprecationLogger;
import org.elasticsearch.common.logging.Loggers;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.rest.BaseRestHandler;
import org.elasticsearch.rest.BytesRestResponse;
import org.elasticsearch.rest.RestController;
import org.elasticsearch.rest.RestRequest;
import org.elasticsearch.rest.RestResponse;
import org.elasticsearch.rest.action.RestBuilderListener;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import java.io.IOException;
import static org.elasticsearch.rest.RestRequest.Method.POST;
import static org.elasticsearch.rest.RestRequest.Method.PUT;
import static org.elasticsearch.rest.RestStatus.OK;
/**
* <pre>
* { "index" : { "_index" : "test", "_type" : "type1", "_id" : "1" }
* { "type1" : { "field1" : "value1" } }
* { "delete" : { "_index" : "test", "_type" : "type1", "_id" : "2" } }
* { "create" : { "_index" : "test", "_type" : "type1", "_id" : "1" }
* { "type1" : { "field1" : "value1" } }
* </pre>
*/
public class RestBulkAction extends BaseRestHandler {
private static final DeprecationLogger DEPRECATION_LOGGER =
new DeprecationLogger(Loggers.getLogger(RestBulkAction.class));
private final boolean allowExplicitIndex;
public RestBulkAction(Settings settings, RestController controller) {
super(settings);
controller.registerHandler(POST, "/_bulk", this);
controller.registerHandler(PUT, "/_bulk", this);
controller.registerHandler(POST, "/{index}/_bulk", this);
controller.registerHandler(PUT, "/{index}/_bulk", this);
controller.registerHandler(POST, "/{index}/{type}/_bulk", this);
controller.registerHandler(PUT, "/{index}/{type}/_bulk", this);
this.allowExplicitIndex = MULTI_ALLOW_EXPLICIT_INDEX.get(settings);
}
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
BulkRequest bulkRequest = Requests.bulkRequest();
String defaultIndex = request.param("index");
String defaultType = request.param("type");
String defaultRouting = request.param("routing");
FetchSourceContext defaultFetchSourceContext = FetchSourceContext.parseFromRestRequest(request);
String fieldsParam = request.param("fields");
if (fieldsParam != null) {
DEPRECATION_LOGGER.deprecated("Deprecated field [fields] used, expected [_source] instead");
}
String[] defaultFields = fieldsParam != null ? Strings.commaDelimitedListToStringArray(fieldsParam) : null;
String defaultPipeline = request.param("pipeline");
String waitForActiveShards = request.param("wait_for_active_shards");
if (waitForActiveShards != null) {
bulkRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
}
bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
bulkRequest.setRefreshPolicy(request.param("refresh"));
bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields,
defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex);
return channel -> client.bulk(bulkRequest, new RestBuilderListener<BulkResponse>(channel) {
@Override
public RestResponse buildResponse(BulkResponse response, XContentBuilder builder) throws Exception {
builder.startObject();
builder.field(Fields.TOOK, response.getTookInMillis());
if (response.getIngestTookInMillis() != BulkResponse.NO_INGEST_TOOK) {
builder.field(Fields.INGEST_TOOK, response.getIngestTookInMillis());
}
builder.field(Fields.ERRORS, response.hasFailures());
builder.startArray(Fields.ITEMS);
for (BulkItemResponse itemResponse : response) {
itemResponse.toXContent(builder, request);
}
builder.endArray();
builder.endObject();
return new BytesRestResponse(OK, builder);
}
});
}
static final class Fields {
static final String ITEMS = "items";
static final String ERRORS = "errors";
static final String TOOK = "took";
static final String INGEST_TOOK = "ingest_took";
}
}
|
