[WIP] Bulk index pivoted resources

mod_elasticsearch.erl

@@ -18,7 +18,8 @@
     terminate/2,
     code_change/3,
     start_link/1,
-    manage_schema/2
+    manage_schema/2,
+    bulk_index/2
 ]).
 
 -include("zotonic.hrl").
@@ -62,13 +63,22 @@ manage_schema(_Version, _Context) ->
         ]
     }.
 
+bulk_index(Tasks, Context) ->
+    Ids = [Id || #task{data = Id} <- Tasks],
+    case elasticsearch:bulk(Ids, Context) of
+        {ok, _Items} ->
+            ok;
+        {error, _Reason} ->
+            retry
+    end.
+
 handle_call({#search_query{} = Search, Context}, _From, State) ->
     {reply, search(Search, Context), State};
 handle_call(Message, _From, State) ->
     {stop, {unknown_call, Message}, State}.
 
 handle_cast(#rsc_pivot_done{id = Id}, State = #state{context = Context}) ->
-    elasticsearch:put_doc(Id, Context),
+    z_queue:queue(#task{callback = {?MODULE, bulk_index}, data = Id, batch = 100}, Context),
     {noreply, State};
 handle_cast(#rsc_delete{id = Id}, State = #state{context = Context}) ->
     elasticsearch:delete_doc(Id, Context),

support/elasticsearch.erl

@@ -5,11 +5,12 @@
 -export([
     connection/0,
     index/1,
+    put_mapping/3,
+    put_mapping/4,
     put_doc/2,
     put_doc/3,
     put_doc/5,
-    put_mapping/3,
-    put_mapping/4,
+    bulk/2,
     delete_doc/2,
     delete_doc/3,
     handle_response/1
@@ -45,18 +46,38 @@ put_mapping(Index, Type, Doc, _Context) ->
     handle_response(Response).
 
 %% Save a resource to Elasticsearch
+put_doc({Index, Type, Id, Data}, _Context) ->
+    Response = erlastic_search:index_doc_with_id(Index, z_convert:to_binary(Type), z_convert:to_binary(Id), Data),
+    handle_response(Response);
 put_doc(Id, Context) ->
-    put_doc(index(Context), Id, Context).
+    put_doc(map_doc(Id, Context), Context).
 
 put_doc(Index, Id, Context) ->
-    % All resource properties
-    Props = elasticsearch_mapping:map_rsc(Id, Context),
-    put_doc(Index, <<"resource">>, z_convert:to_binary(Id), Props, Context).
+    put_doc(map_doc(Index, Id, Context), Context).
 
 put_doc(Index, Type, Id, Data, Context) ->
-    NotifiedData = z_notifier:foldl(#elasticsearch_put{index = Index, type = Type, id = Id}, Data, Context),
-    Response = erlastic_search:index_doc_with_id(Index, Type, Id, NotifiedData),
+    put_doc(map_doc(Index, Type, Id, Data, Context), Context).
+
+-spec bulk([m_rsc:resource()], z:context()) -> {ok, list()} | {error, any()}.
+bulk(Ids, Context) ->
+    Docs = [map_doc(Id, Context) || Id <- Ids],
+    Response = erlastic_search:bulk_index_docs(connection(), Docs),
     handle_response(Response).
+
+map_doc(Id, Context) ->
+    map_doc(index(Context), Id, Context).
+
+map_doc(Index, Id, Context) ->
+    map_doc(Index, <<"resource">>, Id, Context).
+
+map_doc(Index, Type, Id, Context) ->
+    map_doc(Index, Type, Id, elasticsearch_mapping:map_rsc(Id, Context), Context).
+
+-spec map_doc(binary(), binary(), binary(), lists:proplist() | map(), z:context()) ->
+    {binary(), binary(), binary(), lists:proplist() | map()}.
+map_doc(Index, Type, Id, Data, Context) ->
+    Data2 = z_notifier:foldl(#elasticsearch_put{index = Index, type = Type, id = Id}, Data, Context),
+    {Index, Type, Id, Data2}.
 
 delete_doc(Id, Context) ->
     delete_doc(Id, index(Context), Context).
@@ -80,7 +101,8 @@ handle_response(Response) ->
             lager:error(
                 "Elasticsearch error: ~p with connection ~p",
                 [Reason, connection()]
-            );
+            ),
+            Response;
         {ok, _} ->
             Response
     end.