clemp6r · March 3, 2016 14:25 · Mar 3, 2016 · Mar 3, 2016
diff --git a/ReactiveStore.kt b/ReactiveStore.kt
@@ -43,6 +43,7 @@ abstract class ReactiveStore<T> {
                             onNewValue(it)
                         },
                         {
+                            // TODO wrap errors into values without stopping the observable
                             bus.onError(it)
                             bus = PublishSubject.create<T>()
                         }

diff --git a/ReactiveStore.kt b/ReactiveStore.kt
@@ -0,0 +1,62 @@
+package com.wizbii.wizbiiandroid.services
+
+import rx.Observable
+import rx.subjects.PublishSubject
+
+/**
+ * POC of a reactive data store.
+ */
+abstract class ReactiveStore<T> {
+
+    /**
+     * Local version of the data.
+     */
+    private var value: T? = null
+
+    /**
+     * Bus for notifying new versions of the data to all subscribers.
+     */
+    private var bus = PublishSubject.create<T>()
+
+    /**
+     * Observable for fetching the remote version of the data.
+     */
+    protected abstract fun getRemoteData(): Observable<T>
+
+    /**
+     * Updates the remote data and emits the updated value.
+     */
+    protected abstract fun updateRemoteData(parameters: Any): Observable<T>
+
+    /**
+     * Infinite stream of value updates.
+     */
+    fun values(): Observable<T> {
+        return Observable.create {
+            if (value != null) {
+                it.onNext(value)
+                bus.subscribe(it)
+            } else {
+                bus.subscribe(it)
+                getRemoteData().subscribe(
+                        {
+                            onNewValue(it)
+                        },
+                        {
+                            bus.onError(it)
+                            bus = PublishSubject.create<T>()
+                        }
+                )
+            }
+        }
+    }
+
+    private fun onNewValue(it: T) {
+        value = it
+        bus.onNext(it)
+    }
+
+    fun updateValue(parameters: Any): Observable<T> {
+        return updateRemoteData(parameters).doOnNext { onNewValue(it) }
+    }
+}