Fix merge behavior

lempiji · Jun 14, 2020 · a04e02e · a04e02e
1 parent d997dc9
commit a04e02e
Show file tree

Hide file tree

Showing 3 changed files with 395 additions and 25 deletions.
diff --git a/examples/rx-fswatch/source/app.d b/examples/rx-fswatch/source/app.d
@@ -11,17 +11,25 @@ void main()
         auto shutdown = false;
 
         auto thread = new Thread({
-            auto watch = FileWatch("./data", true);
-
-            while (!shutdown)
+            try
             {
-                auto events = watch.getEvents();
-                if (events.length > 0)
+                auto watch = FileWatch("./data", true);
+
+                while (!shutdown)
                 {
-                    .put(observer, events);
-                }
+                    auto events = watch.getEvents();
+                    if (events.length > 0)
+                    {
+                        .put(observer, events);
+                    }
 
-                Thread.sleep(100.msecs);
+                    Thread.sleep(100.msecs);
+                }
+                observer.completed();
+            }
+            catch (Exception e)
+            {
+                observer.failure(e);
             }
         });
         thread.start();
@@ -30,7 +38,9 @@ void main()
     });
 
     auto flatten = watcher.map!(events => from(events)).merge();
-    auto changes = flatten.groupBy!(event => event.path).map!(o => o.debounce(2.seconds)).merge();
+    auto changes = flatten.groupBy!(event => event.path)
+        .map!(o => o.debounce(2.seconds))
+        .merge();
 
     // start FileWatch
     auto disposable = changes.doSubscribe((FileChangeEvent event) {