add change stream example

Author: zookzook

examples/change_stream/.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

examples/change_stream/README.md

@@ -0,0 +1,38 @@
+# Change streams
+
+This project shows a change stream example. You need to set up a replica set, because change streams are only available for replica sets and sharded clusters. For more information see 
+
+* https://www.mongodb.com/blog/post/an-introduction-to-change-streams
+* https://docs.mongodb.com/manual/changeStreams/
+
+If you creating a new replica set then you need to create the database `db-1` first, before starting the example. Otherwise you will get some errors, because the database which we will observe does not exsists.
+
+The `ChangeStream` module uses a GenServer for observing changes. It spawns a process to consume the documents returned by the change stream cursor:
+
+    pid = spawn(fn -> Enum.each(get_cursor(state), fn doc -> new_doc(doc) end) end) 
+
+While running this process you will receive some message:
+
+* token: you get a token after a while. You can use this token to reconnect to the change stream without getting old change documents again. 
+* documents: If data changes, you get a document which describes these changes
+
+Let's start the program with `iex -S mix`:
+
+        iex(2)> 
+        16:10:05.018 [info]  Connecting change stream
+         
+        16:10:05.022 [info]  Receiving new token nil
+         
+        iex(3)> Mongo.insert_one(:mongo, "http_errors", %{url: "https://elixir-lang.org"})  
+        {:ok,
+         %Mongo.InsertOneResult{
+           acknowledged: true,
+           inserted_id: #BSON.ObjectId<5d595c42306a5f0d87ab24e7>
+         }}
+        iex(4)> 
+        16:10:10.509 [info]  Receiving new token %{"_data" => #BSON.Binary<825d595c420000000146645f696400645d595c42306a5f0d87ab24e7005a1004fefbdf8754024c339cd73f510a91db2b04>}
+         
+        16:10:10.509 [info]  Receiving new document %{"coll" => "http_errors", "db" => "db-1"}
+        
+        16:10:10.509 [info]  Got http error for url https://elixir-lang.org
+        

examples/change_stream/config/config.exs

@@ -0,0 +1,30 @@
+# This file is responsible for configuring your application
+# and its dependencies with the aid of the Mix.Config module.
+use Mix.Config
+
+# This configuration is loaded before any dependency and is restricted
+# to this project. If another project depends on this project, this
+# file won't be loaded nor affect the parent project. For this reason,
+# if you want to provide default values for your application for
+# third-party users, it should be done in your "mix.exs" file.
+
+# You can configure your application as:
+#
+#     config :change_stream, key: :value
+#
+# and access this configuration in your application as:
+#
+#     Application.get_env(:change_stream, :key)
+#
+# You can also configure a third-party app:
+#
+#     config :logger, level: :info
+#
+
+# It is also possible to import configuration files, relative to this
+# directory. For example, you can emulate configuration per environment
+# by uncommenting the line below and defining dev.exs, test.exs and such.
+# Configuration from the imported file will override the ones defined
+# here (which is why it is important to import them last).
+#
+#     import_config "#{Mix.env()}.exs"

examples/change_stream/lib/change_stream.ex

@@ -0,0 +1,66 @@
+defmodule ChangeStream do
+  use GenServer
+
+  require Logger
+
+  @collection "http_errors"
+  @me __MODULE__
+
+  def start_link() do
+    GenServer.start_link(__MODULE__, nil, name: @me)
+  end
+
+  def new_token(token) do
+    GenServer.cast(@me, {:token, token})
+  end
+
+  def new_doc(doc) do
+    GenServer.cast(@me, {:doc, doc})
+  end
+
+  def init(_) do
+    state = %{last_resume_token: nil}
+    Process.send_after(self(), :connect, 3000)
+    {:ok, state}
+  end
+
+  def handle_info({:DOWN, _, :process, _pid, reason}, state) do
+    Logger.info("#Cursor process is down: #{inspect reason}")
+    Process.send_after(self(), :connect, 3000)
+    {:noreply, state}
+  end
+
+  def handle_info(:connect, state) do
+    Logger.info("Connecting change stream")
+    # Span a new process
+    pid = spawn(fn -> Enum.each(get_cursor(state), fn doc -> new_doc(doc) end) end)
+
+    # Monitor the process
+    Process.monitor(pid)
+
+    {:noreply, state}
+  end
+
+  def handle_cast({:doc, doc}, state) do
+    Logger.info("Receiving new document #{inspect doc["ns"]}")
+    process_doc(doc)
+    {:noreply, state}
+  end
+
+  def handle_cast({:token, token}, state) do
+    Logger.info("Receiving new token #{inspect token}")
+    {:noreply, %{state | last_resume_token: token}}
+  end
+
+  defp process_doc(%{"fullDocument" => %{"url" => url}, "ns" => %{"coll" => "http_errors", "db" => "db-1"}}) do
+    Logger.info("Got http error for url #{url}")
+  end
+
+  defp get_cursor(%{last_resume_token: nil}) do
+    Mongo.watch_collection(:mongo, @collection, [], fn token -> new_token(token) end, max_time: 2_000)
+  end
+  defp get_cursor(%{last_resume_token: token}) do
+    Mongo.watch_collection(:mongo, @collection, [], fn token -> new_token(token) end, max_time: 2_000, resume_after: token)
+  end
+
+end

examples/change_stream/lib/change_stream/application.ex

@@ -0,0 +1,19 @@
+defmodule ChangeStream.Application do
+
+  @moduledoc false
+
+  use Application
+  import Supervisor.Spec
+
+  def start(_type, _args) do
+
+    children = [
+      # this should be a replicat set!
+      worker(Mongo, [[name: :mongo, url: "mongodb://localhost:27027/db-1", pool_size: 3]]),
+      worker(ChangeStream, [])
+    ]
+
+    opts = [strategy: :one_for_one, name: ChangeStream.Supervisor]
+    Supervisor.start_link(children, opts)
+  end
+end

examples/change_stream/mix.exs

@@ -0,0 +1,28 @@
+defmodule ChangeStream.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :change_stream,
+      version: "0.1.0",
+      elixir: "~> 1.8",
+      start_permanent: Mix.env() == :prod,
+      deps: deps()
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger],
+      mod: {ChangeStream.Application, []}
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      {:mongodb_driver, "~> 0.5"}
+    ]
+  end
+end

examples/change_stream/mix.lock

@@ -0,0 +1,6 @@
+%{
+  "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm"},
+  "db_connection": {:hex, :db_connection, "2.0.6", "bde2f85d047969c5b5800cb8f4b3ed6316c8cb11487afedac4aa5f93fd39abfa", [:mix], [{:connection, "~> 1.0.2", [hex: :connection, repo: "hexpm", optional: false]}], "hexpm"},
+  "decimal": {:hex, :decimal, "1.8.0", "ca462e0d885f09a1c5a342dbd7c1dcf27ea63548c65a65e67334f4b61803822e", [:mix], [], "hexpm"},
+  "mongodb_driver": {:hex, :mongodb_driver, "0.5.7", "f29cab9a011f685210c472888ee3fde3573459323f839b67c4d942e067d9eda1", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.0.6", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5", [hex: :decimal, repo: "hexpm", optional: false]}], "hexpm"},
+}

examples/change_stream/test/change_stream_test.exs

@@ -0,0 +1,8 @@
+defmodule ChangeStreamTest do
+  use ExUnit.Case
+  doctest ChangeStream
+
+  test "greets the world" do
+    assert ChangeStream.hello() == :world
+  end
+end

examples/change_stream/test/test_helper.exs

@@ -0,0 +1 @@
+ExUnit.start()