added support for transaction to bucket/gridfs and bulk operations

Author: zookzook

lib/mongo.ex

@@ -664,6 +664,35 @@ defmodule Mongo do
     issue_command(topology_pid, [ping: 1], :read, [batch_size: 1])
   end
 
+  @doc """
+  Explicitly creates a collection or view.
+  """
+  @spec create(GenServer.server, collection, Keyword.t) :: :ok | {:error, Mongo.Error.t}
+  def create(topology_pid, coll, opts \\ []) do
+
+    cmd = [
+            create:              coll,
+            capped:              opts[:capped],
+            autoIndexId:         opts[:auto_index_id],
+            size:                opts[:size],
+            max:                 opts[:max],
+            storageEngine:       opts[:storage_engine],
+            validator:           opts[:validator],
+            validationLevel:     opts[:validation_level],
+            validationAction:    opts[:validation_action],
+            indexOptionDefaults: opts[:index_option_defaults],
+            viewOn:              opts[:view_on],
+            pipeline:            opts[:pipeline],
+            collation:           opts[:collation],
+            writeConcern:        write_concern(opts),
+          ] |> filter_nils()
+
+    with {:ok, _doc} <- issue_command(topology_pid, cmd, :write, opts) do
+      :ok
+    end
+
+  end
+
   @doc """
   Insert a single document into the collection.
 

lib/mongo/grid_fs/bucket.ex

@@ -41,10 +41,14 @@ defmodule Mongo.GridFs.Bucket do
 
     Keyword.merge(@defaults, options)
     |> Enum.reduce(%Bucket{topology_pid: topology_pid, opts: options}, fn {k, v}, bucket -> Map.put(bucket, k, v) end)
-    |> check_indexes
+    |> check_indexes()
 
   end
 
+  def add_session(%Bucket{opts: opts} = bucket, session_opts) do
+    %Bucket{bucket | opts: opts ++ session_opts}
+  end
+
   @doc """
   Returns the collection name for the files collection, default is fs.files.
   """
@@ -127,19 +131,41 @@ defmodule Mongo.GridFs.Bucket do
   #
   defp check_indexes(bucket) do
 
-    case files_collection_empty?(bucket)do
-      true ->
-        _ = create_files_index({bucket, false})
-        _ = create_chunks_index({bucket, false})
-
+    case files_collection_empty?(bucket) do
+      true  -> create_indexes(bucket)
       false ->
-        bucket
-        |> check_files_index()
-        |> create_files_index()
-        |> check_chunks_index()
-        |> create_chunks_index()
+        with :ok <- check_and_create_files_index(bucket),
+             :ok <- check_and_chunks_files_index(bucket) do
+          bucket
+        end
+    end
+
+  end
+
+  ##
+  # create the collection and indexes for files and chunks
+  #
+  defp create_indexes(%Bucket{topology_pid: topology_pid, opts: opts} = bucket) do
+    with :ok <- Mongo.create(topology_pid, files_collection_name(bucket), opts),
+         :ok <- Mongo.create(topology_pid, chunks_collection_name(bucket), opts),
+         :ok <- create_files_index(bucket),
+         :ok <- create_chunks_index(bucket) do
+      bucket
     end
+  end
 
+  def check_and_create_files_index(bucket) do
+    case check_files_index(bucket) do
+      true  -> create_files_index(bucket)
+      false -> :ok
+    end
+  end
+
+  def check_and_chunks_files_index(bucket) do
+    case check_chunks_index(bucket) do
+      true  -> create_chunks_index(bucket)
+      false -> :ok
+    end
   end
 
   ##
@@ -150,33 +176,33 @@ defmodule Mongo.GridFs.Bucket do
   # db.fs.files.findOne({}, { _id : 1 })
   #
   defp files_collection_empty?(%Bucket{topology_pid: topology_pid, opts: opts} = bucket) do
-
+    coll_name = files_collection_name(bucket)
     topology_pid
-    |> Mongo.find_one(files_collection_name(bucket), %{}, Keyword.merge(opts, projection: %{_id: 1}))
+    |> Mongo.show_collections()
+    |> Enum.find(fn name -> name == coll_name end)
     |> is_nil()
-
   end
 
   ##
   # Checks the indexes for the fs.files collection
   #
   defp check_files_index(%Bucket{topology_pid: topology_pid, opts: opts} = bucket) do
-    {bucket, index_member?(topology_pid, files_collection_name(bucket), @files_index_name, opts)}
+    index_member?(topology_pid, files_collection_name(bucket), @files_index_name, opts)
   end
 
   ##
   # Checks the indexes for the fs.chunks collection
   #
   defp check_chunks_index(%Bucket{topology_pid: topology_pid, opts: opts} = bucket) do
-    {bucket, index_member?(topology_pid, chunks_collection_name(bucket), @chunks_index_name, opts)}
+    index_member?(topology_pid, chunks_collection_name(bucket), @chunks_index_name, opts)
   end
 
   # returns true if the collection contains a index with the given name
   def index_member?(topology_pid, coll, index, opts) do
     topology_pid
     |> Mongo.list_indexes(coll, opts)
     |> Enum.map(fn
-      %{"name" => name} -> name
+     %{"name" => name} -> name
         _               -> ""
        end)
     |> Enum.member?(index)
@@ -185,34 +211,23 @@ defmodule Mongo.GridFs.Bucket do
   ##
   # Creates the indexes for the fs.chunks collection
   #
-  defp create_chunks_index({%Bucket{topology_pid: topology_pid, opts: opts} = bucket, false} ) do
-
-    cmd      = [createIndexes: chunks_collection_name(bucket), indexes: [[key: [files_id: 1, n: 1], name: @chunks_index_name, unique: true]]]
-    {:ok, _} = Mongo.issue_command(topology_pid, cmd, :write, opts)
-
-    bucket
+  defp create_chunks_index(%Bucket{topology_pid: topology_pid, opts: opts} = bucket) do
+    cmd = [createIndexes: chunks_collection_name(bucket), indexes: [[key: [files_id: 1, n: 1], name: @chunks_index_name, unique: true]]]
+    with {:ok, _} <- Mongo.issue_command(topology_pid, cmd, :write, opts) do
+      :ok
+    end
   end
 
-  ##
-  # index exists, nothing to do
-  #
-  defp create_chunks_index({bucket, true}), do: bucket
-
   ##
   # Creates the indexes for the fs.files collection
   #
-  defp create_files_index({%Bucket{topology_pid: topology_pid, opts: opts} = bucket, false}) do
-
-    cmd      = [createIndexes: files_collection_name(bucket), indexes: [[key: [filename: 1, uploadDate: 1], name: @files_index_name]]]
-    {:ok, _} = Mongo.issue_command(topology_pid, cmd, :write, opts)
-
-    bucket
+  defp create_files_index(%Bucket{topology_pid: topology_pid, opts: opts} = bucket) do
+    cmd = [createIndexes: files_collection_name(bucket), indexes: [[key: [filename: 1, uploadDate: 1], name: @files_index_name]]]
+    with {:ok, _} <- Mongo.issue_command(topology_pid, cmd, :write, opts) do
+      :ok
+    end
   end
 
-  ##
-  # index exists, nothing to do
-  #
-  defp create_files_index({bucket, true}), do: bucket
 
   defimpl Inspect, for: Bucket do
 

lib/mongo/session.ex

@@ -140,10 +140,10 @@ defmodule Mongo.Session do
 
     ## warte max 120ms, ansonsten kill
     try do
+      fun.(opts)
     rescue
       reason -> {:error, reason}
     end
-    fun.(opts)
 
   end
 
@@ -280,7 +280,12 @@ defmodule Mongo.Session do
       maxTimeMS: Keyword.get(opts, :max_commit_time_ms)
       ] |> filter_nils()
 
-    Mongo.exec_command(conn, cmd, database: "admin")
+    _doc = Mongo.exec_command(conn, cmd, database: "admin")
+
+    # {:ok, %{"$clusterTime" => %{"clusterTime" => #BSON.Timestamp<1567853627:8>,
+    # "signature" => %{"hash" => #BSON.Binary<0000000000000000000000000000000000000000>, "keyId" => 0}},
+    # "ok" => 1.0, "operationTime" => #BSON.Timestamp<1567853627:6>}}
+    :ok
   end
 
   defp run_abort_command(%{conn: conn, server_session: %ServerSession{session_id: id, txn_num: txn_num}, opts: opts}) do
@@ -295,7 +300,15 @@ defmodule Mongo.Session do
       writeConcern: write_concern(opts)
     ] |> filter_nils()
 
-    Mongo.exec_command(conn, cmd, database: "admin")
+    _doc = Mongo.exec_command(conn, cmd, database: "admin")
+
+    #
+    # doc:
+    #    %{"$clusterTime" => %{"clusterTime" => #BSON.Timestamp<1567853164:4>,
+    # "signature" => %{"hash" => #BSON.Binary<0000000000000000000000000000000000000000>, "keyId" => 0}},
+    #"ok" => 1.0, "operationTime" => #BSON.Timestamp<1567853164:4>}
+
+    :ok
   end
 
 

lib/mongo/write_concern.ex

@@ -9,10 +9,20 @@ defmodule Mongo.WriteConcern do
       w: Keyword.get(opts, :w),
       j: Keyword.get(opts, :j),
       wtimeout: Keyword.get(opts, :wtimeout)
-     } |> filter_nils()
+     } |> filter_nils() |> filter_empty()
 
   end
 
+  def filter_empty(map) when is_map(map) and map == %{} do
+    nil
+  end
+  def filter_empty(map) when is_map(map) do
+    map
+  end
+
+  def acknowledged?(nil) do
+    true
+  end
   def acknowledged?(write_concern) when is_map(write_concern) do
     case Map.get(write_concern, :w) do
       0 -> false

test/mongo/grid_fs/upload_test.exs

@@ -3,6 +3,7 @@ defmodule Mongo.GridFs.UploadTest do
 
   alias Mongo.GridFs.Bucket
   alias Mongo.GridFs.Upload
+  alias Mongo.Session
 
   setup_all do
     assert {:ok, pid} = Mongo.TestConnection.connect
@@ -90,4 +91,47 @@ defmodule Mongo.GridFs.UploadTest do
     assert x == chksum
   end
 
+  @tag :mongo_4_2
+  test "upload a text file, check download, length, meta-data and checksum transaction", c do
+
+    src_filename  = "./test/data/test.txt"
+    chksum        = calc_checksum(src_filename)
+    bucket        = Bucket.new(c.pid)
+
+    {:ok, upload_stream } = Session.with_transaction(c.pid, fn opt ->
+      bucket = Bucket.add_session(bucket, opt)
+      upload_stream = Upload.open_upload_stream(bucket, "my-example-file.txt", %{tag: "checked", chk_sum: chksum})
+      File.stream!(src_filename, [], 512) |> Stream.into(upload_stream) |> Stream.run()
+      {:ok, upload_stream}
+    end, w: 1)
+
+    file_id = upload_stream.id
+
+    assert file_id != nil
+
+    %{"metadata" => %{"tag" => "checked", "chk_sum" => x}} = Mongo.find_one(c.pid, Bucket.files_collection_name(bucket), %{_id: file_id})
+    assert x == chksum
+  end
+
+  @tag :mongo_4_2
+  test "upload a text file, check download, length, meta-data and checksum abort transaction", c do
+
+    src_filename  = "./test/data/test.txt"
+    chksum        = calc_checksum(src_filename)
+    bucket        = Bucket.new(c.pid)
+
+    {:error, upload_stream} = Session.with_transaction(c.pid, fn opt ->
+      bucket = Bucket.add_session(bucket, opt)
+      upload_stream = Upload.open_upload_stream(bucket, "my-example-file.txt", %{tag: "checked", chk_sum: chksum})
+      File.stream!(src_filename, [], 512) |> Stream.into(upload_stream) |> Stream.run()
+      {:error, upload_stream}
+    end, w: 1)
+
+    file_id = upload_stream.id
+
+    assert file_id != nil
+
+    assert nil == Mongo.find_one(c.pid, Bucket.files_collection_name(Bucket.new(c.pid)), %{_id: file_id})
+  end
+
 end