107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
# File 'lib/logstash/outputs/elasticsearch/protocol.rb', line 107
def bulk(actions)
bulk_response = @client.bulk(:body => actions.collect do |action, args, source|
if action == 'update'
if args[:_id]
source = { 'doc' => source }
if @options[:doc_as_upsert]
source['doc_as_upsert'] = true
else
source['upsert'] = args[:_upsert] if args[:_upsert]
end
else
raise(LogStash::ConfigurationError, "Specifying action => 'update' without a document '_id' is not supported.")
end
end
args.delete(:_upsert)
if source
next [ { action => args }, source ]
else
next { action => args }
end
end.flatten)
self.class.normalize_bulk_response(bulk_response)
end
|