Add observe and observe_deep functions to Array, Map, Text, XmlElemen…

lib/doc.ex

@@ -54,38 +54,52 @@ defmodule Yex.Doc do
   """
   defmacro run_in_worker_process(doc, do: block) do
     quote do
-      case unquote(doc).worker_pid do
-        pid when pid == self() ->
-          unquote(block)
-
-        nil ->
-          raise RuntimeError, "Document has no worker process assigned"
-
-        worker_pid ->
-          wrapped_fun = fn ->
-            try do
-              unquote(block)
-            rescue
-              e ->
-                {Yex.Doc, :reraise, e, __STACKTRACE__}
-            end
-          end
+      Yex.Doc.run_in_worker_process_fn(unquote(doc), fn ->
+        unquote(block)
+      end)
+    end
+  end
+
+  @spec run_in_worker_process_fn(
+          atom() | %{:worker_pid => any(), optional(any()) => any()},
+          any()
+        ) :: any()
+  def run_in_worker_process_fn(doc, fun) do
+    case doc.worker_pid do
+      pid when pid == self() ->
+        result = fun.()
+        handle_on_update_handler()
+        result
 
-          # When a Doc or SharedType operation is performed from a process different from the
-          # process that created the Doc, a message like {Yex.Doc, :run, fun} is sent to the
-          # creator process via GenServer.call and the processing is delegated to that process.
-          # If you encounter a GenServer.call timeout, this delegation mechanism may be the cause.
-          case GenServer.call(
-                 worker_pid,
-                 {Yex.Doc, :run, wrapped_fun}
-               ) do
-            {Yex.Doc, :reraise, e, stacktrace} ->
-              reraise e, stacktrace
-
-            result ->
-              result
+      nil ->
+        raise RuntimeError, "Document has no worker process assigned"
+
+      worker_pid ->
+        wrapped_fun = fn ->
+          try do
+            result = fun.()
+            handle_on_update_handler()
+            result
+          rescue
+            e ->
+              {Yex.Doc, :reraise, e, __STACKTRACE__}
           end
-      end
+        end
+
+        # When a Doc or SharedType operation is performed from a process different from the
+        # process that created the Doc, a message like {Yex.Doc, :run, fun} is sent to the
+        # creator process via GenServer.call and the processing is delegated to that process.
+        # If you encounter a GenServer.call timeout, this delegation mechanism may be the cause.
+        case GenServer.call(
+               worker_pid,
+               {Yex.Doc, :run, wrapped_fun}
+             ) do
+          {Yex.Doc, :reraise, e, stacktrace} ->
+            reraise e, stacktrace
+
+          result ->
+            result
+        end
     end
   end
 
@@ -285,6 +299,82 @@ defmodule Yex.Doc do
     end
   end
 
+  def on_subdocs(%__MODULE__{} = doc, handler) do
+    result =
+      run_in_worker_process(doc,
+        do: Yex.Nif.doc_monitor_subdocs(doc, self(), {Yex.CallbackHandler, handler})
+      )
+
+    case result do
+      {:ok, sub} ->
+        {:ok, Yex.Subscription.register(sub)}
+
+      error ->
+        error
+    end
+  end
+
+  @spec on_update(Yex.Doc.t(), handler :: (update :: binary(), origin :: term() -> nil)) :: any()
+  def on_update(%__MODULE__{} = doc, handler) do
+    on_update_v1(doc, handler)
+  end
+
+  @spec on_update_v1(Yex.Doc.t(), handler :: (update :: binary(), origin :: term() -> nil)) ::
+          any()
+  def on_update_v1(%__MODULE__{} = doc, handler) do
+    result =
+      run_in_worker_process(doc,
+        do: Yex.Nif.doc_monitor_update_v1(doc, self(), {Yex.CallbackHandler, handler})
+      )
+
+    case result do
+      {:ok, sub} ->
+        {:ok, Yex.Subscription.register(sub)}
+
+      error ->
+        error
+    end
+  end
+
+  @spec on_update_v2(Yex.Doc.t(), handler :: (update :: binary(), origin :: term() -> nil)) ::
+          any()
+  def on_update_v2(%__MODULE__{} = doc, handler) do
+    result =
+      run_in_worker_process(doc,
+        do: Yex.Nif.doc_monitor_update_v2(doc, self(), {Yex.CallbackHandler, handler})
+      )
+
+    case result do
+      {:ok, sub} ->
+        {:ok, Yex.Subscription.register(sub)}
+
+      error ->
+        error
+    end
+  end
+
+  defp handle_on_update_handler() do
+    receive do
+      {_, arg1, {Yex.CallbackHandler, handler}} ->
+        handler.(arg1)
+        handle_on_update_handler()
+
+      {:subdocs, event, origin, {Yex.CallbackHandler, handler}} ->
+        handler.(event, origin)
+        handle_on_update_handler()
+
+      {_, arg1, arg2, {Yex.CallbackHandler, handler}} ->
+        handler.(arg1, arg2)
+        handle_on_update_handler()
+
+      {_, _ref, event, origin, {Yex.ObserveCallbackHandler, handler}} ->
+        handler.(event, origin)
+        handle_on_update_handler()
+    after
+      0 -> :ok
+    end
+  end
+
   defp cur_txn(%__MODULE__{reference: ref}) do
     Process.get(ref, nil)
   end

lib/shared_type/array.ex

@@ -388,6 +388,28 @@ defmodule Yex.Array do
     Enum.member?(to_list(array), val)
   end
 
+  @spec observe(
+          Yex.Array.t(),
+          handler :: (update :: Yex.ArrayEvent.t(), origin :: term() -> nil)
+        ) :: reference()
+  def observe(%__MODULE__{doc: doc} = array, handler) do
+    Yex.SharedType.observe(array,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
+  @spec observe_deep(
+          Yex.Array.t(),
+          handler :: (update :: list(Yex.event_type()), origin :: term() -> nil)
+        ) :: reference()
+  def observe_deep(%__MODULE__{doc: doc} = array, handler) do
+    Yex.SharedType.observe_deep(array,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
   defp cur_txn(%{doc: %Yex.Doc{reference: doc_ref}}) do
     Process.get(doc_ref, nil)
   end

lib/shared_type/event.ex

@@ -109,3 +109,20 @@ defmodule Yex.XmlTextEvent do
           delta: Yex.Text.delta()
         }
 end
+
+defmodule Yex.WeakEvent do
+  @moduledoc """
+
+  Event when Weak type changes
+
+  @see Yex.SharedType.observe/1
+  @see Yex.SharedType.observe_deep/1
+  """
+  defstruct [
+    :path
+  ]
+
+  @type t :: %__MODULE__{
+          path: list(number() | String.t())
+        }
+end

lib/shared_type/map.ex

@@ -325,6 +325,28 @@ defmodule Yex.Map do
     )
   end
 
+  @spec observe(
+          t,
+          handler :: (update :: Yex.MapEvent.t(), origin :: term() -> nil)
+        ) :: reference()
+  def observe(%__MODULE__{doc: doc} = map, handler) do
+    Yex.SharedType.observe(map,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
+  @spec observe_deep(
+          t,
+          handler :: (update :: list(Yex.event_type()), origin :: term() -> nil)
+        ) :: reference()
+  def observe_deep(%__MODULE__{doc: doc} = map, handler) do
+    Yex.SharedType.observe_deep(map,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
   @doc false
   # Gets the current transaction reference from the process dictionary for the given document
   defp cur_txn(%{doc: %Yex.Doc{reference: doc_ref}}) do

lib/shared_type/shared_type.ex

@@ -39,7 +39,7 @@ defmodule Yex.SharedType do
   @spec observe(t, keyword()) :: reference()
   def observe(%{doc: doc} = shared_type, opt \\ []) do
     ref = make_ref()
-    notify_pid = self()
+    notify_pid = Keyword.get(opt, :notify_pid, self())
 
     sub =
       Doc.run_in_worker_process(doc,
@@ -87,7 +87,7 @@ defmodule Yex.SharedType do
   @spec observe_deep(t, keyword()) :: reference()
   def observe_deep(%{doc: doc} = shared_type, opt \\ []) do
     ref = make_ref()
-    notify_pid = self()
+    notify_pid = Keyword.get(opt, :notify_pid, self())
 
     sub =
       Doc.run_in_worker_process(doc,

lib/shared_type/text.ex

@@ -202,6 +202,28 @@ defmodule Yex.Text do
     )
   end
 
+  @spec observe(
+          t,
+          handler :: (update :: Yex.TextEvent.t(), origin :: term() -> nil)
+        ) :: reference()
+  def observe(%__MODULE__{doc: doc} = text, handler) do
+    Yex.SharedType.observe(text,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
+  @spec observe_deep(
+          t,
+          handler :: (update :: list(Yex.event_type()), origin :: term() -> nil)
+        ) :: reference()
+  def observe_deep(%__MODULE__{doc: doc} = text, handler) do
+    Yex.SharedType.observe_deep(text,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
   defp cur_txn(%{doc: %Yex.Doc{reference: doc_ref}}) do
     Process.get(doc_ref, nil)
   end

lib/shared_type/xml_element.ex

@@ -351,6 +351,28 @@ defmodule Yex.XmlElement do
     )
   end
 
+  @spec observe(
+          t,
+          handler :: (update :: Yex.XmlEvent.t(), origin :: term() -> nil)
+        ) :: reference()
+  def observe(%__MODULE__{doc: doc} = xml_element, handler) do
+    Yex.SharedType.observe(xml_element,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
+  @spec observe_deep(
+          t,
+          handler :: (update :: list(Yex.event_type()), origin :: term() -> nil)
+        ) :: reference()
+  def observe_deep(%__MODULE__{doc: doc} = xml_element, handler) do
+    Yex.SharedType.observe_deep(xml_element,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
   defp cur_txn(%{doc: %Yex.Doc{reference: doc_ref}}) do
     Process.get(doc_ref, nil)
   end

lib/shared_type/xml_fragment.ex

@@ -280,6 +280,28 @@ defmodule Yex.XmlFragment do
     )
   end
 
+  @spec observe(
+          t,
+          handler :: (update :: Yex.XmlEvent.t(), origin :: term() -> nil)
+        ) :: reference()
+  def observe(%__MODULE__{doc: doc} = xml_fragment, handler) do
+    Yex.SharedType.observe(xml_fragment,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
+  @spec observe_deep(
+          t,
+          handler :: (update :: list(Yex.event_type()), origin :: term() -> nil)
+        ) :: reference()
+  def observe_deep(%__MODULE__{doc: doc} = xml_fragment, handler) do
+    Yex.SharedType.observe_deep(xml_fragment,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
   @doc false
   # Gets the current transaction reference from the process dictionary for the given document
   defp cur_txn(%{doc: %Yex.Doc{reference: doc_ref}}) do

lib/shared_type/xml_text.ex

@@ -139,6 +139,28 @@ defmodule Yex.XmlText do
     )
   end
 
+  @spec observe(
+          t,
+          handler :: (update :: Yex.XmlTextEvent.t(), origin :: term() -> nil)
+        ) :: reference()
+  def observe(%__MODULE__{doc: doc} = xml_text, handler) do
+    Yex.SharedType.observe(xml_text,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
+  @spec observe_deep(
+          t,
+          handler :: (update :: list(Yex.event_type()), origin :: term() -> nil)
+        ) :: reference()
+  def observe_deep(%__MODULE__{doc: doc} = xml_text, handler) do
+    Yex.SharedType.observe_deep(xml_text,
+      metadata: {Yex.ObserveCallbackHandler, handler},
+      notify_pid: doc.worker_pid
+    )
+  end
+
   @doc false
   # Gets the current transaction reference from the process dictionary for the given document
   defp cur_txn(%{doc: %Yex.Doc{reference: doc_ref}}) do

lib/y_ex.ex

@@ -15,6 +15,13 @@ defmodule Yex do
           any_type
           | Yex.PrelimType.t()
 
+  @type event_type ::
+          Yex.XmlTextEvent.t()
+          | Yex.XmlEvent.t()
+          | Yex.TextEvent.t()
+          | Yex.MapEvent.t()
+          | Yex.ArrayEvent.t()
+          | Yex.WeakEvent.t()
   @doc """
   Computes the state vector and encodes it into an Binary.
   A state vector describes the state of the local client.