Merge pull request #71 from neurogears/issue-69

Redirect initialization and dispose errors
open-ephys · Apr 9, 2024 · db83496 · db83496
2 parents 7ed4ada + c31b7e9
commit db83496
Show file tree

Hide file tree

Showing 3 changed files with 60 additions and 16 deletions.
diff --git a/OpenEphys.Onix/OpenEphys.Onix/ContextTask.cs b/OpenEphys.Onix/OpenEphys.Onix/ContextTask.cs
@@ -93,23 +93,23 @@ public void Reset()
 
         // NB: This is where actions that reconfigure the hub state, or otherwise
         // change the device table should be executed
-        internal void ConfigureHost(Func<ContextTask, IDisposable> action)
+        internal void ConfigureHost(Func<ContextTask, IDisposable> configure)
         {
-            configureHost += action;
+            configureHost += configure;
         }
 
         // NB: This is where actions that calibrate port voltage or otherwise
         // check link lock state should be executed
-        internal void ConfigureLink(Func<ContextTask, IDisposable> action)
+        internal void ConfigureLink(Func<ContextTask, IDisposable> configure)
         {
-            configureLink += action;
+            configureLink += configure;
         }
 
         // NB: Actions queued using this method should assume that the device table
         // is finalized and cannot be changed
-        internal void ConfigureDevice(Func<ContextTask, IDisposable> selector)
+        internal void ConfigureDevice(Func<ContextTask, IDisposable> configure)
         {
-            configureDevice += selector;
+            configureDevice += configure;
         }
 
         private IDisposable ConfigureContext()

diff --git a/OpenEphys.Onix/OpenEphys.Onix/DeviceManager.cs b/OpenEphys.Onix/OpenEphys.Onix/DeviceManager.cs
@@ -24,8 +24,9 @@ internal static IDisposable RegisterDevice(string name, DeviceInfo deviceInfo)
                 var subject = disposable.Subject;
                 if (subject.IsCompleted)
                 {
-                    throw new InvalidOperationException(
-                        "A device with the same name has already been configured."
+                    throw new ArgumentException(
+                        $"A device with the same name '{name}' has already been configured.",
+                        nameof(name)
                     );
                 }
 
@@ -39,8 +40,11 @@ internal static IDisposable RegisterDevice(string name, DeviceInfo deviceInfo)
                     var info = entry.Value.Subject.GetResult();
                     if (info.Context == deviceInfo.Context && info.DeviceAddress == deviceInfo.DeviceAddress)
                     {
-                        throw new InvalidOperationException(
-                            "A device with the same address has already been configured in this context."
+                        throw new ArgumentException(
+                            $"The specified device '{deviceInfo.DeviceType.Name}' could not be registered " +
+                            $"because another device '{info.DeviceType.Name}' with the same address " +
+                            $"{info.DeviceAddress} has already been configured in this context.",
+                            nameof(deviceInfo)
                         );
                     }
                 }

diff --git a/OpenEphys.Onix/OpenEphys.Onix/ObservableExtensions.cs b/OpenEphys.Onix/OpenEphys.Onix/ObservableExtensions.cs
@@ -1,23 +1,63 @@
 using System;
+using System.Reactive;
+using System.Reactive.Disposables;
 using System.Reactive.Linq;
 
 namespace OpenEphys.Onix
 {
     internal static class ObservableExtensions
     {
-        public static IObservable<ContextTask> ConfigureHost(this IObservable<ContextTask> source, Func<ContextTask, IDisposable> action)
+        public static IObservable<ContextTask> ConfigureHost(this IObservable<ContextTask> source, Func<ContextTask, IDisposable> configure)
         {
-            return source.Do(context => context.ConfigureHost(action));
+            return source.ConfigureContext((context, action) => context.ConfigureHost(action), configure);
         }
 
-        public static IObservable<ContextTask> ConfigureLink(this IObservable<ContextTask> source, Func<ContextTask, IDisposable> action)
+        public static IObservable<ContextTask> ConfigureLink(this IObservable<ContextTask> source, Func<ContextTask, IDisposable> configure)
         {
-            return source.Do(context => context.ConfigureLink(action));
+            return source.ConfigureContext((context, action) => context.ConfigureLink(action), configure);
         }
 
-        public static IObservable<ContextTask> ConfigureDevice(this IObservable<ContextTask> source, Func<ContextTask, IDisposable> selector)
+        public static IObservable<ContextTask> ConfigureDevice(this IObservable<ContextTask> source, Func<ContextTask, IDisposable> configure)
         {
-            return source.Do(context => context.ConfigureDevice(selector));
+            return source.ConfigureContext((context, action) => context.ConfigureDevice(action), configure);
+        }
+
+        static IObservable<ContextTask> ConfigureContext(
+            this IObservable<ContextTask> source,
+            Action<ContextTask, Func<ContextTask, IDisposable>> configureContext,
+            Func<ContextTask, IDisposable> configure)
+        {
+            return Observable.Create<ContextTask>(observer =>
+            {
+                var contextObserver = Observer.Create<ContextTask>(
+                    context =>
+                    {
+                        configureContext(context, ctx =>
+                        {
+                            try
+                            {
+                                var disposable = configure(ctx);
+                                return Disposable.Create(() =>
+                                {
+                                    try { disposable.Dispose(); }
+                                    catch (Exception ex)
+                                    {
+                                        observer.OnError(ex);
+                                    }
+                                });
+                            }
+                            catch (Exception ex)
+                            {
+                                observer.OnError(ex);
+                                throw;
+                            }
+                        });
+                        observer.OnNext(context);
+                    },
+                    observer.OnError,
+                    observer.OnCompleted);
+                return source.SubscribeSafe(contextObserver);
+            });
         }
     }
 }