added first version for stream api

Author: zookzook

README.md

@@ -217,22 +217,36 @@ unordered bulk writes.
 
 ```elixir
 
-    bulk = "bulk"
-           |> new()
-           |> insert_one(%{name: "Greta"})
-           |> insert_one(%{name: "Tom"})
-           |> insert_one(%{name: "Waldo"})
-           |> update_one(%{name: "Greta"}, %{"$set": %{kind: "dog"}})
-           |> update_one(%{name: "Tom"}, %{"$set": %{kind: "dog"}})
-           |> update_one(%{name: "Waldo"}, %{"$set": %{kind: "dog"}})
-           |> update_many(%{kind: "dog"}, %{"$set": %{kind: "cat"}})
-           |> delete_one(%{kind: "cat"})
-           |> delete_one(%{kind: "cat"})
-           |> delete_one(%{kind: "cat"})
-
-    result = Mongo.BulkWrite.bulk_write(:mongo, bulk, w: 1)
+bulk = "bulk"
+       |> new()
+       |> insert_one(%{name: "Greta"})
+       |> insert_one(%{name: "Tom"})
+       |> insert_one(%{name: "Waldo"})
+       |> update_one(%{name: "Greta"}, %{"$set": %{kind: "dog"}})
+       |> update_one(%{name: "Tom"}, %{"$set": %{kind: "dog"}})
+       |> update_one(%{name: "Waldo"}, %{"$set": %{kind: "dog"}})
+       |> update_many(%{kind: "dog"}, %{"$set": %{kind: "cat"}})
+       |> delete_one(%{kind: "cat"})
+       |> delete_one(%{kind: "cat"})
+       |> delete_one(%{kind: "cat"})
+
+result = Mongo.BulkWrite.bulk_write(:mongo, bulk, w: 1)
 ```
 
+In the following example we import 1.000.000 integers into the MongoDB using the stream api:
+
+We need to create an insert operation for each number. Then we call the `ongo.UnorderedBulk.stream`
+function to import it. This function returns a stream function which accumulate 
+all inserts operations until the limit `1000` is reached. In this case the operation group is send to
+MongoDB. So using the stream api you can reduce the memory using while 
+importing big volume of data.
+
+```elixir
+stream = 1..1_000_000 
+    |> Stream.map(fn i -> Mongo.BulkUtils.get_insert_one(%{number: i}) end) 
+    |> Mongo.UnorderedBulk.stream(top, "bulk", 1_000)
+    |> Stream.run()
+```
 ### Examples
 
 Using `$and`

lib/mongo/bulk_utils.ex

@@ -0,0 +1,23 @@
+defmodule Mongo.BulkUtils do
+
+  def get_insert_one(doc), do: {:insert, doc}
+
+  def get_delete_one(doc, opts \\ []), do: {:delete, {doc, Keyword.put(opts, :limit, 1)}}
+  def get_delete_many(doc, opts \\ []), do: {:delete, {doc, Keyword.put(opts, :limit, 0)}}
+
+  def get_update_one(filter, update, opts \\ []) do
+    ## _ = modifier_docs(update, :update)
+    {:update, {filter, update, Keyword.put(opts, :multi, false)}}
+  end
+
+  def get_update_many(filter, update, opts \\ []) do
+    ## _ = modifier_docs(update, :update)
+    {:update, {filter, update, Keyword.put(opts, :multi, true)}}
+  end
+
+  def get_replace_one(filter, replacement, opts \\ []) do
+    ## _ = modifier_docs(replacement, :replace)
+    {:update, {filter, replacement, Keyword.put(opts, :multi, false)}}
+  end
+
+end

lib/mongo/bulk_write.ex

@@ -13,6 +13,8 @@ defmodule Mongo.BulkWrite do
   alias Mongo.UnorderedBulk
   alias Mongo.OrderedBulk
 
+  @max_batch_size 100_000
+
   @doc """
   Unordered bulk write operations:
   Executes first insert commands, then updates commands and after that all delete commands are executed. If a group (inserts, updates or deletes) exceeds the limit
@@ -174,7 +176,7 @@ defmodule Mongo.BulkWrite do
 
   defp run_commands(conn, cmds, opts) do
 
-    IO.puts "Running cmds #{inspect cmds}"
+    # IO.puts "Running cmds #{inspect cmds}"
 
     cmds
     |> Enum.map(fn cmd -> Mongo.direct_command(conn, cmd, opts) end)
@@ -183,12 +185,10 @@ defmodule Mongo.BulkWrite do
 
   def get_insert_cmds(coll, docs, write_concern, _opts) do
 
-    max_batch_size = 10 ## only for test maxWriteBatchSize
-
     {_ids, docs} = assign_ids(docs)
 
     docs
-    |> Enum.chunk_every(max_batch_size)
+    |> Enum.chunk_every(@max_batch_size)
     |> Enum.map(fn inserts -> get_insert_cmd(coll, inserts, write_concern) end)
 
   end
@@ -201,9 +201,8 @@ defmodule Mongo.BulkWrite do
 
   defp get_delete_cmds(coll, docs, write_concern, opts) do
 
-    max_batch_size = 10 ## only for test maxWriteBatchSize
     docs
-    |> Enum.chunk_every(max_batch_size)
+    |> Enum.chunk_every(@max_batch_size)
     |> Enum.map(fn deletes -> get_delete_cmd(coll, deletes, write_concern, opts) end)
 
   end
@@ -222,9 +221,8 @@ defmodule Mongo.BulkWrite do
 
   defp get_update_cmds(coll, docs, write_concern, opts) do
 
-    max_batch_size = 10 ## only for test maxWriteBatchSize
     docs
-    |> Enum.chunk_every(max_batch_size)
+    |> Enum.chunk_every(@max_batch_size)
     |> Enum.map(fn updates -> get_update_cmd(coll, updates, write_concern, opts) end)
 
   end

lib/mongo/ordered_bulk.ex

@@ -65,22 +65,6 @@ defmodule Mongo.OrderedBulk do
     IO.puts inspect result
   end
 
-  def test2() do
-
-    # create a streaming bulk write with max 1024 operations
-    bulk_stream = "bulk" |> new_stream(:mongo, 1024, w: 1)
-
-    # now streaming a long text file with small memory usage
-    File.stream!(file)
-    |> Stream.with_index
-    #|> Stream.map(fn {name, i} -> insert_one(%{line: i, name: name}) end) # {:insert, %{line: i, name: name}}
-    # |> Stream.into(bulk_stream, (fn {name, i} -> insert_one(%{line: i, name: name}) end))
-    |> Stream.map(fn {name, i} -> bulk_stream.insert_one(%{line: i, name: name}) end)
-    |> Stream.reduce()
-
-    File.stream!(src_filename, [], 512) |> Stream.into(bulk_stream) |> Stream.run()
-
-  end
 
 
 end

lib/mongo/unordered_bulk.ex

@@ -6,42 +6,51 @@ defmodule Mongo.UnorderedBulk do
   """
 
   alias Mongo.UnorderedBulk
+  import Mongo.BulkUtils
 
   defstruct coll: nil, inserts: [], updates: [], deletes: []
 
   def new(coll) do
     %UnorderedBulk{coll: coll}
   end
 
-
-  def insert_one(%UnorderedBulk{inserts: rest} = bulk, doc) do
+  def push({:insert, doc}, %UnorderedBulk{inserts: rest} = bulk) do
     %UnorderedBulk{bulk | inserts: [doc | rest] }
   end
+  def push({:update, doc}, %UnorderedBulk{updates: rest} = bulk) do
+    %UnorderedBulk{bulk | updates: [doc | rest] }
+  end
+  def push({:delete, doc}, %UnorderedBulk{deletes: rest} = bulk) do
+    %UnorderedBulk{bulk | deletes: [doc | rest] }
+  end
+
+  def insert_one(%UnorderedBulk{} = bulk, doc) do
+    get_insert_one(doc) |> push(bulk)
+  end
 
-  def delete_one(%UnorderedBulk{deletes: rest} = bulk, doc, opts \\ []) do
-    %UnorderedBulk{bulk | deletes: [{doc, Keyword.put(opts, :limit, 1)} | rest] }
+  def delete_one(%UnorderedBulk{} = bulk, doc, opts \\ []) do
+    get_delete_one(doc, opts) |> push(bulk)
   end
 
-  def delete_many(%UnorderedBulk{deletes: rest} = bulk, doc, opts \\ []) do
-    %UnorderedBulk{bulk | deletes: [{doc, Keyword.put(opts, :limit, 0)} | rest] }
+  def delete_many(%UnorderedBulk{} = bulk, doc, opts \\ []) do
+    get_delete_many(doc, opts) |> push(bulk)
   end
 
-  def replace_one(%UnorderedBulk{updates: rest} = bulk, filter, replacement, opts \\ []) do
+  def replace_one(%UnorderedBulk{} = bulk, filter, replacement, opts \\ []) do
     _ = modifier_docs(replacement, :replace)
-    %UnorderedBulk{bulk | updates: [{filter, replacement, Keyword.put(opts, :multi, false)} | rest] }
+    get_replace_one(filter, replacement, opts) |> push(bulk)
   end
 
-  def update_one(%UnorderedBulk{updates: rest} = bulk, filter, update, opts \\ []) do
+  def update_one(%UnorderedBulk{} = bulk, filter, update, opts \\ []) do
     _ = modifier_docs(update, :update)
-    %UnorderedBulk{bulk | updates: [{filter, update, Keyword.put(opts, :multi, false)} | rest] }
+    get_update_one(filter, update, opts) |> push(bulk)
   end
 
-  def update_many(%UnorderedBulk{updates: rest} = bulk, filter, replacement, opts \\ []) do
-    _ = modifier_docs(replacement, :update)
-    %UnorderedBulk{bulk | updates: [{filter, replacement, Keyword.put(opts, :multi, true)} | rest] }
+  def update_many(%UnorderedBulk{updates: rest} = bulk, filter, update, opts \\ []) do
+    _ = modifier_docs(update, :update)
+    get_update_many(filter, update, opts) |> push(bulk)
   end
 
-
   defp modifier_docs([{key, _}|_], type), do: key |> key_to_string |> modifier_key(type)
   defp modifier_docs(map, _type) when is_map(map) and map_size(map) == 0,  do: :ok
   defp modifier_docs(map, type) when is_map(map), do: Enum.at(map, 0) |> elem(0) |> key_to_string |> modifier_key(type)
@@ -55,6 +64,20 @@ defmodule Mongo.UnorderedBulk do
   defp key_to_string(key) when is_atom(key), do: Atom.to_string(key)
   defp key_to_string(key) when is_binary(key), do: key
 
+  def stream(enum, top, coll, limit \\ 1000, opts \\ []) when limit > 1 do
+    Stream.chunk_while(enum,
+      {new(coll), limit - 1},
+      fn
+        op, {bulk, 0} -> {:cont, Mongo.BulkWrite.bulk_write(top, push(op, bulk), opts), {new(coll), limit - 1}}
+        op, {bulk, l} -> {:cont, {push(op, bulk), l - 1}}
+      end,
+      fn
+        {bulk, 0} -> {:cont, bulk}
+        {bulk, _} -> {:cont, Mongo.BulkWrite.bulk_write(top, bulk, opts), {new(coll), limit - 1}}
+    end)
+    # todo reduce to one
+  end
+
   def test(top) do
 
     bulk = "bulk"
@@ -107,4 +130,5 @@ defmodule Mongo.UnorderedBulk do
     IO.puts inspect result
   end
 
+
 end