fix for broadcaststream

Samples/UIWidgetsSamples_2019_4/Assets/Editor/Tests/Stream/TestMain.cs

                 });
             }
             /**
-             * Test OnDone/OnData/Stream.fromIterable
+             * Test ErrorHandler
              */
             private void test2()
             {
             }
             
             /**
-             * Test ErrorHandler
+             * Test OnDone/OnData/Stream.fromIterable
              */
             private void test3()
             {
                 });
             }
 
+            /**
+             * Test Stream.take
+             */
             private void test5()
             {
                 Stream<int> numbers = Stream<int>.periodic(new TimeSpan(0, 0, 0, 1), t => t).take(3);
                     Debug.Log("periodic finished");
                 });
             }
+
+            /**
+             * Test Stream.asBroadcastStream
+             */
+            private void test6()
+            {
+                Stream<int> numbers = Stream<int>.periodic(new TimeSpan(0, 0, 0, 1), t => t).asBroadcastStream().take(10);
+
+                var subscription1 = numbers.listen((data) =>
+                {
+                    Debug.Log("Sub1: " + data);
+                });
+                
+                var subscription2 = numbers.listen((data) =>
+                {
+                    Debug.Log("Sub2: " + data);
+                    if (data == 3)
+                    {
+                        subscription1.cancel();
+                    }
+                });
+            }
+            
+            /**
+             * Test listen( ..., cancelOnError = true)
+             */
+            private void test7()
+            {
+                Stream<int> numbers = Stream<int>.periodic(new TimeSpan(0, 0, 0, 1), t =>
+                {
+                    if (t == 2)
+                    {
+                        throw new Exception("LaLaLa");
+                    }
+                    
+                    return t;
+                }).take(5);
+                void sumStream(Stream<int> stream, Action<int> onDone)
+                {
+                    var sum = 0;
+                    stream.listen(val =>
+                        {
+                            sum += val;
+                            Debug.Log("sum stream = " + sum);
+                        }, 
+                        onDone: () =>
+                        {
+                            onDone.Invoke(sum);
+                        },
+                        onError: (e, stack) =>
+                        {
+                            Debug.Log("error at " + stack);
+                        },
+                        cancelOnError: true);
+                }
+
+                sumStream(numbers, val =>
+                {
+                    Debug.Log("sum = " + (int)val);
+                });
+            }
-                test5();
+                test7();
                 return new Container();
             }
         }

com.unity.uiwidgets/Runtime/async/stream_impl.cs

             _onListenHandler = a => Zone.current
                 .registerUnaryCallback(
                     b => {
-                        onListenHandler((StreamSubscription<T>) b);
+                        onListenHandler?.Invoke((StreamSubscription<T>) b);
                         return default;
                     }
                 )(a);
-                        onCancelHandler((StreamSubscription<T>) c);
+                        onCancelHandler?.Invoke((StreamSubscription<T>) c);
                         return default;
                     })(d);
             _zone = Zone.current;