From 63d9ffae2ce4bd7d9e2863c18cbaee736d68dcd7 Mon Sep 17 00:00:00 2001
From: suoerkerokero <superkerokero@outlook.com>
Date: Sun, 31 May 2020 15:16:31 +0900
Subject: [PATCH 1/2] Allow AsyncMessageBroker to recieve data from
 subscribers.

---
 .../UniRx/Scripts/Notifiers/MessageBroker.cs  | 245 ++++++++++--------
 1 file changed, 139 insertions(+), 106 deletions(-)

diff --git a/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs b/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs
index a951831d..f0497a7c 100644
--- a/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs
+++ b/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs
@@ -2,102 +2,93 @@
 using System.Collections.Generic;
 using UniRx.InternalUtil;
 
-namespace UniRx
-{
-    public interface IMessagePublisher
-    {
+namespace UniRx {
+    public interface IMessagePublisher {
         /// <summary>
         /// Send Message to all receiver.
         /// </summary>
-        void Publish<T>(T message);
+        void Publish<T> (T message);
     }
 
-    public interface IMessageReceiver
-    {
+    public interface IMessageReceiver {
         /// <summary>
         /// Subscribe typed message.
         /// </summary>
-        IObservable<T> Receive<T>();
+        IObservable<T> Receive<T> ();
     }
 
-    public interface IMessageBroker : IMessagePublisher, IMessageReceiver
-    {
-    }
+    public interface IMessageBroker : IMessagePublisher, IMessageReceiver { }
 
-    public interface IAsyncMessagePublisher
-    {
+    public interface IAsyncMessagePublisher {
         /// <summary>
         /// Send Message to all receiver and await complete.
         /// </summary>
-        IObservable<Unit> PublishAsync<T>(T message);
+        IObservable<Unit> PublishAsync<T> (T message);
+
+        /// <summary>
+        /// Send Message to all recievers and await data from recievers
+        /// </summary>
+        IObservable<T[]> PublishAsync<T, T1> (T1 message);
     }
 
-    public interface IAsyncMessageReceiver
-    {
+    public interface IAsyncMessageReceiver {
         /// <summary>
         /// Subscribe typed message.
         /// </summary>
-        IDisposable Subscribe<T>(Func<T, IObservable<Unit>> asyncMessageReceiver);
-    }
+        IDisposable Subscribe<T> (Func<T, IObservable<Unit>> asyncMessageReceiver);
 
-    public interface IAsyncMessageBroker : IAsyncMessagePublisher, IAsyncMessageReceiver
-    {
+        /// <summary>
+        /// Subscribe typed message with payload.
+        /// </summary>
+        IDisposable Subscribe<T, T1> (Func<T, IObservable<T1>> asyncMessageReceiver);
     }
 
+    public interface IAsyncMessageBroker : IAsyncMessagePublisher, IAsyncMessageReceiver { }
+
     /// <summary>
     /// In-Memory PubSub filtered by Type.
     /// </summary>
-    public class MessageBroker : IMessageBroker, IDisposable
-    {
+    public class MessageBroker : IMessageBroker, IDisposable {
         /// <summary>
         /// MessageBroker in Global scope.
         /// </summary>
-        public static readonly IMessageBroker Default = new MessageBroker();
+        public static readonly IMessageBroker Default = new MessageBroker ();
 
         bool isDisposed = false;
-        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object>();
+        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object> ();
 
-        public void Publish<T>(T message)
-        {
+        public void Publish<T> (T message) {
             object notifier;
-            lock (notifiers)
-            {
-                if (isDisposed) return;
+            lock (notifiers) {
+                    if (isDisposed) return;
 
-                if (!notifiers.TryGetValue(typeof(T), out notifier))
-                {
-                    return;
+                    if (!notifiers.TryGetValue (typeof (T), out notifier)) {
+                        return;
+                    }
                 }
-            }
-            ((ISubject<T>)notifier).OnNext(message);
+                ((ISubject<T>) notifier).OnNext (message);
         }
 
-        public IObservable<T> Receive<T>()
-        {
+        public IObservable<T> Receive<T> () {
             object notifier;
-            lock (notifiers)
-            {
-                if (isDisposed) throw new ObjectDisposedException("MessageBroker");
+            lock (notifiers) {
+                if (isDisposed) throw new ObjectDisposedException ("MessageBroker");
 
-                if (!notifiers.TryGetValue(typeof(T), out notifier))
-                {
-                    ISubject<T> n = new Subject<T>().Synchronize();
+                if (!notifiers.TryGetValue (typeof (T), out notifier)) {
+                    ISubject<T> n = new Subject<T> ().Synchronize ();
                     notifier = n;
-                    notifiers.Add(typeof(T), notifier);
+                    notifiers.Add (typeof (T), notifier);
                 }
             }
 
-            return ((IObservable<T>)notifier).AsObservable();
+            return ((IObservable<T>) notifier).AsObservable ();
         }
 
-        public void Dispose()
-        {
-            lock (notifiers)
-            {
-                if (!isDisposed)
-                {
+        public void Dispose () {
+            lock (notifiers) {
+                if (!isDisposed) {
                     isDisposed = true;
-                    notifiers.Clear();
+                    notifiers.Clear ();
                 }
             }
         }
@@ -106,101 +97,143 @@ public void Dispose()
     /// <summary>
     /// In-Memory PubSub filtered by Type.
     /// </summary>
-    public class AsyncMessageBroker : IAsyncMessageBroker, IDisposable
-    {
+    public class AsyncMessageBroker : IAsyncMessageBroker, IDisposable {
         /// <summary>
         /// AsyncMessageBroker in Global scope.
         /// </summary>
-        public static readonly IAsyncMessageBroker Default = new AsyncMessageBroker();
+        public static readonly IAsyncMessageBroker Default = new AsyncMessageBroker ();
 
         bool isDisposed = false;
-        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object>();
+        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object> ();
 
-        public IObservable<Unit> PublishAsync<T>(T message)
-        {
+        public IObservable<Unit> PublishAsync<T> (T message) {
             UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>> notifier;
-            lock (notifiers)
-            {
-                if (isDisposed) throw new ObjectDisposedException("AsyncMessageBroker");
+            lock (notifiers) {
+                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
 
                 object _notifier;
-                if (notifiers.TryGetValue(typeof(T), out _notifier))
-                {
-                    notifier = (UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>>)_notifier;
-                }
-                else
-                {
-                    return Observable.ReturnUnit();
+                if (notifiers.TryGetValue (typeof (T), out _notifier)) {
+                    notifier = (UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>>) _notifier;
+                } else {
+                    return Observable.ReturnUnit ();
                 }
             }
 
             var data = notifier.Data;
             var awaiter = new IObservable<Unit>[data.Length];
-            for (int i = 0; i < data.Length; i++)
-            {
-                awaiter[i] = data[i].Invoke(message);
+            for (int i = 0; i < data.Length; i++) {
+                awaiter[i] = data[i].Invoke (message);
+            }
+            return Observable.WhenAll (awaiter);
+        }
+
+        public IObservable<T[]> PublishAsync<T, T1> (T1 message) {
+            UniRx.InternalUtil.ImmutableList<Func<T1, IObservable<T>>> notifier;
+            lock (notifiers) {
+                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
+
+                object _notifier;
+                if (notifiers.TryGetValue (typeof (T1), out _notifier)) {
+                    notifier = (UniRx.InternalUtil.ImmutableList<Func<T1, IObservable<T>>>) _notifier;
+                } else {
+                    return null;
+                }
+            }
+
+            var data = notifier.Data;
+            var awaiter = new IObservable<T>[data.Length];
+            for (int i = 0; i < data.Length; i++) {
+                awaiter[i] = data[i].Invoke (message);
             }
-            return Observable.WhenAll(awaiter);
+            return Observable.WhenAll (awaiter);
         }
 
-        public IDisposable Subscribe<T>(Func<T, IObservable<Unit>> asyncMessageReceiver)
-        {
-            lock (notifiers)
-            {
-                if (isDisposed) throw new ObjectDisposedException("AsyncMessageBroker");
+        public IDisposable Subscribe<T> (Func<T, IObservable<Unit>> asyncMessageReceiver) {
+            lock (notifiers) {
+                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
 
                 object _notifier;
-                if (!notifiers.TryGetValue(typeof(T), out _notifier))
-                {
+                if (!notifiers.TryGetValue (typeof (T), out _notifier)) {
                     var notifier = UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>>.Empty;
-                    notifier = notifier.Add(asyncMessageReceiver);
-                    notifiers.Add(typeof(T), notifier);
+                    notifier = notifier.Add (asyncMessageReceiver);
+                    notifiers.Add (typeof (T), notifier);
+                } else {
+                    var notifier = (ImmutableList<Func<T, IObservable<Unit>>>) _notifier;
+                    notifier = notifier.Add (asyncMessageReceiver);
+                    notifiers[typeof (T)] = notifier;
                 }
-                else
-                {
-                    var notifier = (ImmutableList<Func<T, IObservable<Unit>>>)_notifier;
-                    notifier = notifier.Add(asyncMessageReceiver);
-                    notifiers[typeof(T)] = notifier;
+            }
+
+            return new Subscription<T> (this, asyncMessageReceiver);
+        }
+
+        public IDisposable Subscribe<T, T1> (Func<T, IObservable<T1>> asyncMessageReceiver) {
+            lock (notifiers) {
+                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
+
+                object _notifier;
+                if (!notifiers.TryGetValue (typeof (T), out _notifier)) {
+                    var notifier = UniRx.InternalUtil.ImmutableList<Func<T, IObservable<T1>>>.Empty;
+                    notifier = notifier.Add (asyncMessageReceiver);
+                    notifiers.Add (typeof (T), notifier);
+                } else {
+                    var notifier = (ImmutableList<Func<T, IObservable<T1>>>) _notifier;
+                    notifier = notifier.Add (asyncMessageReceiver);
+                    notifiers[typeof (T)] = notifier;
                 }
             }
 
-            return new Subscription<T>(this, asyncMessageReceiver);
+            return new Subscription<T, T1> (this, asyncMessageReceiver);
         }
 
-        public void Dispose()
-        {
-            lock (notifiers)
-            {
-                if (!isDisposed)
-                {
+        public void Dispose () {
+            lock (notifiers) {
+                if (!isDisposed) {
                     isDisposed = true;
-                    notifiers.Clear();
+                    notifiers.Clear ();
                 }
             }
         }
 
-        class Subscription<T> : IDisposable
-        {
+        class Subscription<T> : IDisposable {
             readonly AsyncMessageBroker parent;
             readonly Func<T, IObservable<Unit>> asyncMessageReceiver;
 
-            public Subscription(AsyncMessageBroker parent, Func<T, IObservable<Unit>> asyncMessageReceiver)
-            {
+            public Subscription (AsyncMessageBroker parent, Func<T, IObservable<Unit>> asyncMessageReceiver) {
+                this.parent = parent;
+                this.asyncMessageReceiver = asyncMessageReceiver;
+            }
+
+            public void Dispose () {
+                lock (parent.notifiers) {
+                    object _notifier;
+                    if (parent.notifiers.TryGetValue (typeof (T), out _notifier)) {
+                        var notifier = (ImmutableList<Func<T, IObservable<Unit>>>) _notifier;
+                        notifier = notifier.Remove (asyncMessageReceiver);
+
+                        parent.notifiers[typeof (T)] = notifier;
+                    }
+                }
+            }
+        }
+
+        class Subscription<T, T1> : IDisposable {
+            readonly AsyncMessageBroker parent;
+            readonly Func<T, IObservable<T1>> asyncMessageReceiver;
+
+            public Subscription (AsyncMessageBroker parent, Func<T, IObservable<T1>> asyncMessageReceiver) {
                 this.parent = parent;
                 this.asyncMessageReceiver = asyncMessageReceiver;
             }
 
-            public void Dispose()
-            {
-                lock (parent.notifiers)
-                {
+            public void Dispose () {
+                lock (parent.notifiers) {
                     object _notifier;
-                    if (parent.notifiers.TryGetValue(typeof(T), out _notifier))
-                    {
-                        var notifier = (ImmutableList<Func<T, IObservable<Unit>>>)_notifier;
-                        notifier = notifier.Remove(asyncMessageReceiver);
+                    if (parent.notifiers.TryGetValue (typeof (T), out _notifier)) {
+                        var notifier = (ImmutableList<Func<T, IObservable<T1>>>) _notifier;
+                        notifier = notifier.Remove (asyncMessageReceiver);
 
-                        parent.notifiers[typeof(T)] = notifier;
+                        parent.notifiers[typeof (T)] = notifier;
                     }
                 }
             }

From 06d40ca753f97f018e8d02703e173649ef97bfe2 Mon Sep 17 00:00:00 2001
From: suoerkerokero <superkerokero@outlook.com>
Date: Sun, 31 May 2020 15:49:58 +0900
Subject: [PATCH 2/2] Fix format

---
 .../UniRx/Scripts/Notifiers/MessageBroker.cs  | 252 +++++++++++-------
 1 file changed, 154 insertions(+), 98 deletions(-)

diff --git a/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs b/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs
index f0497a7c..91341d77 100644
--- a/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs
+++ b/Assets/Plugins/UniRx/Scripts/Notifiers/MessageBroker.cs
@@ -2,93 +2,112 @@
 using System.Collections.Generic;
 using UniRx.InternalUtil;
 
-namespace UniRx {
-    public interface IMessagePublisher {
+namespace UniRx
+{
+    public interface IMessagePublisher
+    {
         /// <summary>
         /// Send Message to all receiver.
         /// </summary>
-        void Publish<T> (T message);
+        void Publish<T>(T message);
     }
 
-    public interface IMessageReceiver {
+    public interface IMessageReceiver
+    {
         /// <summary>
         /// Subscribe typed message.
         /// </summary>
-        IObservable<T> Receive<T> ();
+        IObservable<T> Receive<T>();
     }
 
-    public interface IMessageBroker : IMessagePublisher, IMessageReceiver { }
+    public interface IMessageBroker : IMessagePublisher, IMessageReceiver
+    {
+    }
 
-    public interface IAsyncMessagePublisher {
+    public interface IAsyncMessagePublisher
+    {
         /// <summary>
         /// Send Message to all receiver and await complete.
         /// </summary>
-        IObservable<Unit> PublishAsync<T> (T message);
+        IObservable<Unit> PublishAsync<T>(T message);
 
         /// <summary>
         /// Send Message to all recievers and await data from recievers
         /// </summary>
-        IObservable<T[]> PublishAsync<T, T1> (T1 message);
+        IObservable<T[]> PublishAsync<T, T1>(T1 message);
     }
 
-    public interface IAsyncMessageReceiver {
+    public interface IAsyncMessageReceiver
+    {
         /// <summary>
         /// Subscribe typed message.
         /// </summary>
-        IDisposable Subscribe<T> (Func<T, IObservable<Unit>> asyncMessageReceiver);
+        IDisposable Subscribe<T>(Func<T, IObservable<Unit>> asyncMessageReceiver);
 
         /// <summary>
         /// Subscribe typed message with payload.
         /// </summary>
-        IDisposable Subscribe<T, T1> (Func<T, IObservable<T1>> asyncMessageReceiver);
+        IDisposable Subscribe<T, T1>(Func<T, IObservable<T1>> asyncMessageReceiver);
     }
 
-    public interface IAsyncMessageBroker : IAsyncMessagePublisher, IAsyncMessageReceiver { }
+    public interface IAsyncMessageBroker : IAsyncMessagePublisher, IAsyncMessageReceiver
+    {
+    }
 
     /// <summary>
     /// In-Memory PubSub filtered by Type.
     /// </summary>
-    public class MessageBroker : IMessageBroker, IDisposable {
+    public class MessageBroker : IMessageBroker, IDisposable
+    {
         /// <summary>
         /// MessageBroker in Global scope.
         /// </summary>
-        public static readonly IMessageBroker Default = new MessageBroker ();
+        public static readonly IMessageBroker Default = new MessageBroker();
 
         bool isDisposed = false;
-        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object> ();
+        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object>();
 
-        public void Publish<T> (T message) {
+        public void Publish<T>(T message)
+        {
             object notifier;
-            lock (notifiers) {
-                    if (isDisposed) return;
+            lock (notifiers)
+            {
+                if (isDisposed) return;
 
-                    if (!notifiers.TryGetValue (typeof (T), out notifier)) {
-                        return;
-                    }
+                if (!notifiers.TryGetValue(typeof(T), out notifier))
+                {
+                    return;
                 }
-                ((ISubject<T>) notifier).OnNext (message);
+            }
+            ((ISubject<T>)notifier).OnNext(message);
         }
 
-        public IObservable<T> Receive<T> () {
+        public IObservable<T> Receive<T>()
+        {
             object notifier;
-            lock (notifiers) {
-                if (isDisposed) throw new ObjectDisposedException ("MessageBroker");
+            lock (notifiers)
+            {
+                if (isDisposed) throw new ObjectDisposedException("MessageBroker");
 
-                if (!notifiers.TryGetValue (typeof (T), out notifier)) {
-                    ISubject<T> n = new Subject<T> ().Synchronize ();
+                if (!notifiers.TryGetValue(typeof(T), out notifier))
+                {
+                    ISubject<T> n = new Subject<T>().Synchronize();
                     notifier = n;
-                    notifiers.Add (typeof (T), notifier);
+                    notifiers.Add(typeof(T), notifier);
                 }
             }
 
-            return ((IObservable<T>) notifier).AsObservable ();
+            return ((IObservable<T>)notifier).AsObservable();
         }
 
-        public void Dispose () {
-            lock (notifiers) {
-                if (!isDisposed) {
+        public void Dispose()
+        {
+            lock (notifiers)
+            {
+                if (!isDisposed)
+                {
                     isDisposed = true;
-                    notifiers.Clear ();
+                    notifiers.Clear();
                 }
             }
         }
@@ -97,143 +116,180 @@ public void Dispose () {
     /// <summary>
     /// In-Memory PubSub filtered by Type.
     /// </summary>
-    public class AsyncMessageBroker : IAsyncMessageBroker, IDisposable {
+    public class AsyncMessageBroker : IAsyncMessageBroker, IDisposable
+    {
         /// <summary>
         /// AsyncMessageBroker in Global scope.
         /// </summary>
-        public static readonly IAsyncMessageBroker Default = new AsyncMessageBroker ();
+        public static readonly IAsyncMessageBroker Default = new AsyncMessageBroker();
 
         bool isDisposed = false;
-        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object> ();
+        readonly Dictionary<Type, object> notifiers = new Dictionary<Type, object>();
 
-        public IObservable<Unit> PublishAsync<T> (T message) {
+        public IObservable<Unit> PublishAsync<T>(T message)
+        {
             UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>> notifier;
-            lock (notifiers) {
-                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
+            lock (notifiers)
+            {
+                if (isDisposed) throw new ObjectDisposedException("AsyncMessageBroker");
 
                 object _notifier;
-                if (notifiers.TryGetValue (typeof (T), out _notifier)) {
-                    notifier = (UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>>) _notifier;
-                } else {
-                    return Observable.ReturnUnit ();
+                if (notifiers.TryGetValue(typeof(T), out _notifier))
+                {
+                    notifier = (UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>>)_notifier;
+                }
+                else
+                {
+                    return Observable.ReturnUnit();
                 }
             }
 
             var data = notifier.Data;
             var awaiter = new IObservable<Unit>[data.Length];
-            for (int i = 0; i < data.Length; i++) {
-                awaiter[i] = data[i].Invoke (message);
+            for (int i = 0; i < data.Length; i++)
+            {
+                awaiter[i] = data[i].Invoke(message);
             }
-            return Observable.WhenAll (awaiter);
+            return Observable.WhenAll(awaiter);
         }
 
-        public IObservable<T[]> PublishAsync<T, T1> (T1 message) {
+        public IObservable<T[]> PublishAsync<T, T1>(T1 message)
+        {
             UniRx.InternalUtil.ImmutableList<Func<T1, IObservable<T>>> notifier;
-            lock (notifiers) {
-                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
+            lock (notifiers)
+            {
+                if (isDisposed) throw new ObjectDisposedException("AsyncMessageBroker");
 
                 object _notifier;
-                if (notifiers.TryGetValue (typeof (T1), out _notifier)) {
-                    notifier = (UniRx.InternalUtil.ImmutableList<Func<T1, IObservable<T>>>) _notifier;
-                } else {
+                if (notifiers.TryGetValue(typeof(T1), out _notifier))
+                {
+                    notifier = (UniRx.InternalUtil.ImmutableList<Func<T1, IObservable<T>>>)_notifier;
+                }
+                else
+                {
                     return null;
                 }
             }
 
             var data = notifier.Data;
             var awaiter = new IObservable<T>[data.Length];
-            for (int i = 0; i < data.Length; i++) {
-                awaiter[i] = data[i].Invoke (message);
+            for (int i = 0; i < data.Length; i++)
+            {
+                awaiter[i] = data[i].Invoke(message);
             }
-            return Observable.WhenAll (awaiter);
+            return Observable.WhenAll(awaiter);
         }
 
-        public IDisposable Subscribe<T> (Func<T, IObservable<Unit>> asyncMessageReceiver) {
-            lock (notifiers) {
-                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
+
+        public IDisposable Subscribe<T>(Func<T, IObservable<Unit>> asyncMessageReceiver)
+        {
+            lock (notifiers)
+            {
+                if (isDisposed) throw new ObjectDisposedException("AsyncMessageBroker");
 
                 object _notifier;
-                if (!notifiers.TryGetValue (typeof (T), out _notifier)) {
+                if (!notifiers.TryGetValue(typeof(T), out _notifier))
+                {
                     var notifier = UniRx.InternalUtil.ImmutableList<Func<T, IObservable<Unit>>>.Empty;
-                    notifier = notifier.Add (asyncMessageReceiver);
-                    notifiers.Add (typeof (T), notifier);
-                } else {
-                    var notifier = (ImmutableList<Func<T, IObservable<Unit>>>) _notifier;
-                    notifier = notifier.Add (asyncMessageReceiver);
-                    notifiers[typeof (T)] = notifier;
+                    notifier = notifier.Add(asyncMessageReceiver);
+                    notifiers.Add(typeof(T), notifier);
+                }
+                else
+                {
+                    var notifier = (ImmutableList<Func<T, IObservable<Unit>>>)_notifier;
+                    notifier = notifier.Add(asyncMessageReceiver);
+                    notifiers[typeof(T)] = notifier;
                 }
             }
 
-            return new Subscription<T> (this, asyncMessageReceiver);
+            return new Subscription<T>(this, asyncMessageReceiver);
         }
 
-        public IDisposable Subscribe<T, T1> (Func<T, IObservable<T1>> asyncMessageReceiver) {
-            lock (notifiers) {
-                if (isDisposed) throw new ObjectDisposedException ("AsyncMessageBroker");
+        public IDisposable Subscribe<T, T1>(Func<T, IObservable<T1>> asyncMessageReceiver)
+        {
+            lock (notifiers)
+            {
+                if (isDisposed) throw new ObjectDisposedException("AsyncMessageBroker");
 
                 object _notifier;
-                if (!notifiers.TryGetValue (typeof (T), out _notifier)) {
+                if (!notifiers.TryGetValue(typeof(T), out _notifier))
+                {
                     var notifier = UniRx.InternalUtil.ImmutableList<Func<T, IObservable<T1>>>.Empty;
-                    notifier = notifier.Add (asyncMessageReceiver);
-                    notifiers.Add (typeof (T), notifier);
-                } else {
-                    var notifier = (ImmutableList<Func<T, IObservable<T1>>>) _notifier;
-                    notifier = notifier.Add (asyncMessageReceiver);
-                    notifiers[typeof (T)] = notifier;
+                    notifier = notifier.Add(asyncMessageReceiver);
+                    notifiers.Add(typeof(T), notifier);
+                }
+                else
+                {
+                    var notifier = (ImmutableList<Func<T, IObservable<T1>>>)_notifier;
+                    notifier = notifier.Add(asyncMessageReceiver);
+                    notifiers[typeof(T)] = notifier;
                 }
             }
 
-            return new Subscription<T, T1> (this, asyncMessageReceiver);
+            return new Subscription<T, T1>(this, asyncMessageReceiver);
         }
 
-        public void Dispose () {
-            lock (notifiers) {
-                if (!isDisposed) {
+        public void Dispose()
+        {
+            lock (notifiers)
+            {
+                if (!isDisposed)
+                {
                     isDisposed = true;
-                    notifiers.Clear ();
+                    notifiers.Clear();
                 }
             }
         }
 
-        class Subscription<T> : IDisposable {
+        class Subscription<T> : IDisposable
+        {
             readonly AsyncMessageBroker parent;
             readonly Func<T, IObservable<Unit>> asyncMessageReceiver;
 
-            public Subscription (AsyncMessageBroker parent, Func<T, IObservable<Unit>> asyncMessageReceiver) {
+            public Subscription(AsyncMessageBroker parent, Func<T, IObservable<Unit>> asyncMessageReceiver)
+            {
                 this.parent = parent;
                 this.asyncMessageReceiver = asyncMessageReceiver;
             }
 
-            public void Dispose () {
-                lock (parent.notifiers) {
+            public void Dispose()
+            {
+                lock (parent.notifiers)
+                {
                     object _notifier;
-                    if (parent.notifiers.TryGetValue (typeof (T), out _notifier)) {
-                        var notifier = (ImmutableList<Func<T, IObservable<Unit>>>) _notifier;
-                        notifier = notifier.Remove (asyncMessageReceiver);
+                    if (parent.notifiers.TryGetValue(typeof(T), out _notifier))
+                    {
+                        var notifier = (ImmutableList<Func<T, IObservable<Unit>>>)_notifier;
+                        notifier = notifier.Remove(asyncMessageReceiver);
 
-                        parent.notifiers[typeof (T)] = notifier;
+                        parent.notifiers[typeof(T)] = notifier;
                     }
                 }
             }
         }
 
-        class Subscription<T, T1> : IDisposable {
+        class Subscription<T, T1> : IDisposable
+        {
             readonly AsyncMessageBroker parent;
             readonly Func<T, IObservable<T1>> asyncMessageReceiver;
 
-            public Subscription (AsyncMessageBroker parent, Func<T, IObservable<T1>> asyncMessageReceiver) {
+            public Subscription(AsyncMessageBroker parent, Func<T, IObservable<T1>> asyncMessageReceiver)
+            {
                 this.parent = parent;
                 this.asyncMessageReceiver = asyncMessageReceiver;
             }
 
-            public void Dispose () {
-                lock (parent.notifiers) {
+            public void Dispose()
+            {
+                lock (parent.notifiers)
+                {
                     object _notifier;
-                    if (parent.notifiers.TryGetValue (typeof (T), out _notifier)) {
-                        var notifier = (ImmutableList<Func<T, IObservable<T1>>>) _notifier;
-                        notifier = notifier.Remove (asyncMessageReceiver);
+                    if (parent.notifiers.TryGetValue(typeof(T), out _notifier))
+                    {
+                        var notifier = (ImmutableList<Func<T, IObservable<T1>>>)_notifier;
+                        notifier = notifier.Remove(asyncMessageReceiver);
 
-                        parent.notifiers[typeof (T)] = notifier;
+                        parent.notifiers[typeof(T)] = notifier;
                     }
                 }
             }