merge: Integrate v2.1 progress into public repository

This pull request integrates the changes planned for the v2.1 release of
OpenDC into the public Github repository in order to sync the progress
of both repositories.

2 files changed, 82 insertions, 0 deletions

diff --git a/opendc-web/opendc-web-ui/src/redux/sagas/index.js b/opendc-web/opendc-web-ui/src/redux/sagas/index.js
new file mode 100644
index 00000000..0fabdb6d
--- /dev/null
+++ b/opendc-web/opendc-web-ui/src/redux/sagas/index.js
@@ -0,0 +1,7 @@
+import { fork } from 'redux-saga/effects'
+import { watchServer, updateServer } from './topology'
+
+export default function* rootSaga() {
+    yield fork(watchServer)
+    yield fork(updateServer)
+}
diff --git a/opendc-web/opendc-web-ui/src/redux/sagas/topology.js b/opendc-web/opendc-web-ui/src/redux/sagas/topology.js
new file mode 100644
index 00000000..f40cff28
--- /dev/null
+++ b/opendc-web/opendc-web-ui/src/redux/sagas/topology.js
@@ -0,0 +1,75 @@
+import { QueryObserver, MutationObserver } from 'react-query'
+import { normalize, denormalize } from 'normalizr'
+import { select, put, take, race, getContext, call } from 'redux-saga/effects'
+import { eventChannel } from 'redux-saga'
+import { Topology } from '../../util/topology-schema'
+import { storeTopology, OPEN_TOPOLOGY } from '../actions/topologies'
+
+/**
+ * Update the topology on the server.
+ */
+export function* updateServer() {
+    const queryClient = yield getContext('queryClient')
+    const mutationObserver = new MutationObserver(queryClient, { mutationKey: 'updateTopology' })
+
+    while (true) {
+        yield take(
+            (action) =>
+                action.type.startsWith('EDIT') || action.type.startsWith('ADD') || action.type.startsWith('DELETE')
+        )
+        const topology = yield select((state) => state.topology)
+
+        if (!topology.root) {
+            continue
+        }
+
+        const denormalizedTopology = denormalize(topology.root, Topology, topology)
+        yield call([mutationObserver, mutationObserver.mutate], denormalizedTopology)
+    }
+}
+
+/**
+ * Watch the topology on the server for changes.
+ */
+export function* watchServer() {
+    let { id } = yield take(OPEN_TOPOLOGY)
+    while (true) {
+        const channel = yield queryObserver(id)
+
+        while (true) {
+            const [action, response] = yield race([take(OPEN_TOPOLOGY), take(channel)])
+
+            if (action) {
+                id = action.id
+                break
+            }
+
+            const { isFetched, data } = response
+            // Only update the topology on the client-side when a new topology was fetched
+            if (isFetched) {
+                const { result: topologyId, entities } = normalize(data, Topology)
+                yield put(storeTopology(entities.topologies[topologyId], entities))
+            }
+        }
+    }
+}
+
+/**
+ * Observe changes for the topology with the specified identifier.
+ */
+function* queryObserver(id) {
+    const queryClient = yield getContext('queryClient')
+    const observer = new QueryObserver(queryClient, { queryKey: ['topologies', id] })
+
+    return eventChannel((emitter) => {
+        const unsubscribe = observer.subscribe((result) => {
+            emitter(result)
+        })
+
+        // Update result to make sure we did not miss any query updates
+        // between creating the observer and subscribing to it.
+        observer.updateResult()
+
+        return unsubscribe
+    })
+}