Add Latest operator to Ref and Atom Rx support

LanguageExt.Rx/Atom.Extensions.cs

@@ -19,6 +19,19 @@ public static IObservable<A> OnChange<A>(this Atom<A> atom) =>
             add => atom.Change += new AtomChangedEvent<A>(add),
             remove => atom.Change -= new AtomChangedEvent<A>(remove));
 
+    /// <summary>
+    /// Creates an observable that watches for changes to the `Atom`, but
+    /// also gets primed with the initial state on subscription, so there's
+    /// always at least one value that flows downstream.
+    /// </summary>
+    /// <param name="atom">Atom to observe</param>
+    /// <typeparam name="A">Value type</typeparam>
+    /// <returns>Observable atom</returns>
+    [Pure]
+    public static IObservable<A> Latest<A>(this Atom<A> atom) =>
+        Observable.Return(atom.Value)
+                  .Merge(atom.OnChange());
+
     /// <summary>
     /// Observe changes to the `Atom`
     /// </summary>
@@ -31,6 +44,19 @@ public static IObservable<A> OnChange<M, A>(this Atom<M, A> atom) =>
             add => atom.Change += new AtomChangedEvent<A>(add),
             remove => atom.Change -= new AtomChangedEvent<A>(remove));
 
+    /// <summary>
+    /// Creates an observable that watches for changes to the `Atom`, but
+    /// also gets primed with the initial state on subscription, so there's
+    /// always at least one value that flows downstream.
+    /// </summary>
+    /// <param name="atom">Atom to observe</param>
+    /// <typeparam name="A">Value type</typeparam>
+    /// <returns>Observable atom</returns>
+    [Pure]
+    public static IObservable<A> Latest<M, A>(this Atom<M, A> atom) =>
+        Observable.Return(atom.Value)
+                  .Merge(atom.OnChange());
+
     /// <summary>
     /// Observe changes to the `Ref`
     /// </summary>
@@ -43,6 +69,19 @@ public static IObservable<A> OnChange<A>(this Ref<A> atom) =>
             add => atom.Change += new AtomChangedEvent<A>(add),
             remove => atom.Change -= new AtomChangedEvent<A>(remove));
 
+    /// <summary>
+    /// Creates an observable that watches for changes to the `Atom`, but
+    /// also gets primed with the initial state on subscription, so there's
+    /// always at least one value that flows downstream.
+    /// </summary>
+    /// <param name="atom">Atom to observe</param>
+    /// <typeparam name="A">Value type</typeparam>
+    /// <returns>Observable atom</returns>
+    [Pure]
+    public static IObservable<A> Latest<A>(this Ref<A> atom) =>
+        Observable.Return(atom.Value)
+                  .Merge(atom.OnChange());
+
     /// <summary>
     /// Observe changes to the `AtomHashMap`
     /// </summary>