Merge pull request #237 from Unity-Technologies/stream/ide_fixes

Stream/ide fixes

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

                     Debug.Log("val = " + val);
                 });
             }
+
+            /**
+             * Test Stream.multi
+             */
+            private void test13()
+            {
+                var log = new List<string>();
+                var index = 1;
+
+                var multi = Stream<List<int>>.multi(c =>
+                {
+                    var id = index++;
+                    log.Add($"{id}");
+                    for (var i = 0; i < id + 1; i++)
+                    {
+                        c.add(new List<int>{id, i});
+                    }
+
+                    c.close();
+                });
+
+                void logList(List<int> l)
+                {
+                    log.Add($"{l.first()}-{l.last()}");
+                }
+
+                Future.wait<object>(new List<Future> {multi.forEach(logList), multi.forEach(logList)}).whenComplete(
+                    () =>
+                    {
+                        foreach (var str in log)
+                        {
+                            Debug.Log(str);
+                        }
+                    }
+                );
+            }
-                test12();
+                test13();
                 return new Container();
             }
         }

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

             return result;
         }
 
-        Stream<R> cast<R>() where R : class => new CastStream<S, R>(_source);
+        Stream<R> cast<R>() => new CastStream<S, R>(_source);
     }
 
 
     }
 
     class CastStreamTransformer<SS, ST, TS, TT>
-        : StreamTransformerBase<TS, TT> where TT : class where ST : class {
+        : StreamTransformerBase<TS, TT> {
         public readonly StreamTransformer<SS, ST> _source;
 
         public CastStreamTransformer(StreamTransformer<SS, ST> _source) {

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

             return streamConsumer.addStream(this).then((_) => streamConsumer.close(), (_) => FutureOr.nil);
         }
 
-        public Stream<S> transform<S>(StreamTransformer<T, S> streamTransformer) where S : class {
+        public Stream<S> transform<S>(StreamTransformer<T, S> streamTransformer) {
             return streamTransformer.bind(this);
         }
 
             return future.to<bool>();
         }
 
-        Future forEach(Action<T> action) {
+        public Future forEach(Action<T> action) {
             _Future future = new _Future();
             StreamSubscription<T> subscription = null;
             subscription = listen(
         // }
     }
 
-    public abstract class StreamTransformer<S, T> where T : class {
+    public abstract class StreamTransformer<S, T> {
-            where T : class {
+            {
             return new _StreamSubscriptionTransformer<S, T>(onListen);
         }
 
         }
 
         public static StreamTransformer<TS, TT> castFrom<SS, ST, TS, TT>(
-            StreamTransformer<SS, ST> source) where TT : class where ST : class {
+            StreamTransformer<SS, ST> source) {
-        public abstract StreamTransformer<RS, RT> cast<RS, RT>() where RT : class;
+        public abstract StreamTransformer<RS, RT> cast<RS, RT>();
-    public abstract class StreamTransformerBase<S, T> : StreamTransformer<S, T> where T : class {
+    public abstract class StreamTransformerBase<S, T> : StreamTransformer<S, T> {
         public StreamTransformerBase() {
         }
 

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

         }
     }
 
-    public abstract class StreamController<T> : StreamSink<T> {
+
+    public interface IStreamController<T> {
+        Stream<T> stream { get; }
+        
+        _stream.ControllerCallback onListen { get; set; }
+
+        // void  onListen(void onListenHandler());
+
+        _stream.ControllerCallback onPause { get; set; }
+
+        // void set onPause(void onPauseHandler());
+
+        _stream.ControllerCallback onResume { get; set; }
+
+        // void set onResume(void onResumeHandler());
+
+        _stream.ControllerCancelCallback onCancel { get; set; }
+
+        // void set onCancel(onCancelHandler());
+
+        StreamSink<T> sink { get; }
+
+       bool isClosed { get; }
+
+        bool isPaused { get; }
+
+        /** Whether there is a subscriber on the [Stream]. */
+        bool hasListener { get; }
+
+        // public abstract void add(T evt);
+        //
+        // public abstract void addError(object error, string stackTrace);
+
+        Future close();
+
+        Future addStream(Stream<T> source, bool? cancelOnError = false);
+        
+        void add(T evt);
+        void addError(object error, string stackTrace);
+
+        Future done { get; }
+    } 
+    
+    public abstract class StreamController<T> : StreamSink<T>, IStreamController<T> {
         /** The stream that this controller is controlling. */
         public virtual Stream<T> stream { get; }
 
 
     abstract class _StreamController<T> : _StreamControllerBase<T> {
         /** The controller is in its initial state with no subscription. */
-        const int _STATE_INITIAL = 0;
+        internal const int _STATE_INITIAL = 0;
-        const int _STATE_SUBSCRIBED = 1;
+        internal const int _STATE_SUBSCRIBED = 1;
-        const int _STATE_CANCELED = 2;
+        internal const int _STATE_CANCELED = 2;
-        const int _STATE_SUBSCRIPTION_MASK = 3;
+        internal const int _STATE_SUBSCRIPTION_MASK = 3;
 
         // The following state relate to the controller, not the subscription.
         // If closed, adding more events is not allowed.
-        const int _STATE_CLOSED = 4;
-        const int _STATE_ADDSTREAM = 8;
+        internal const int _STATE_CLOSED = 4;
+        internal const int _STATE_ADDSTREAM = 8;
 
         // @pragma("vm:entry-point")
         object _varData;
-        int _state = _STATE_INITIAL;
+        protected int _state = _STATE_INITIAL;
 
         // TODO(lrn): Could this be stored in the varData field too, if it's not
         // accessed until the call to "close"? Then we need to special case if it's
             }
         }
 
-        Exception _badEventState() {
+        protected Exception _badEventState() {
             if (isClosed) {
                 return new Exception("Cannot add event after closing");
             }

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

 using System;
-using Unity.UIWidgets.async;
 
 namespace Unity.UIWidgets.async {
     class _EventSinkWrapper<T> : EventSink<T> {
         }
     }
 
-    class _StreamSinkTransformer<S, T> : StreamTransformerBase<S, T> where T : class {
+    class _StreamSinkTransformer<S, T> : StreamTransformerBase<S, T> {
         readonly _async._SinkMapper<S, T> _sinkMapper;
 
         public _StreamSinkTransformer(_async._SinkMapper<S, T> _sinkMapper) {
         public delegate void _TransformDoneHandler<T>(EventSink<T> sink);
     }
 
-    class _HandlerEventSink<S, T> : EventSink<S> where T : class {
+    class _HandlerEventSink<S, T> : EventSink<S> {
         readonly _stream._TransformDataHandler<S, T> _handleData;
         readonly _stream._TransformErrorHandler<T> _handleError;
         readonly _stream._TransformDoneHandler<T> _handleDone;
                 _handleData(data, _sink);
             }
             else {
-                _sink.add(data as T);
+                _sink.add((T)((object)data));
             }
         }
 
         }
     }
 
-    class _StreamHandlerTransformer<S, T> : _StreamSinkTransformer<S, T> where T : class {
+    class _StreamHandlerTransformer<S, T> : _StreamSinkTransformer<S, T> {
         internal _StreamHandlerTransformer(
             _stream._TransformDataHandler<S, T> handleData = null,
             _stream._TransformErrorHandler<T> handleError = null,
         }
     }
 
-    class _StreamBindTransformer<S, T> : StreamTransformerBase<S, T> where T : class {
+    class _StreamBindTransformer<S, T> : StreamTransformerBase<S, T> {
         readonly Func<Stream<S>, Stream<T>> _bind;
 
         internal _StreamBindTransformer(Func<Stream<S>, Stream<T>> _bind) {
         public delegate StreamSubscription<T> _SubscriptionTransformer<S, T>(Stream<S> stream, bool cancelOnError);
     }
 
-    class _StreamSubscriptionTransformer<S, T> : StreamTransformerBase<S, T> where T : class {
+    class _StreamSubscriptionTransformer<S, T> : StreamTransformerBase<S, T> {
         readonly _async._SubscriptionTransformer<S, T> _onListen;
 
         internal _StreamSubscriptionTransformer(_async._SubscriptionTransformer<S, T> _onListen) {

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

+using System;
+
+namespace Unity.UIWidgets.async {
+    
+    /**
+     *  Stream.multi is not supported by flutter 1.17.5 yet, but it might be useful for developers. To address this issue, we put all the necessary codes for this feature
+     *  in this single file.
+     *
+     * [TODO] remove this code when we eventually upgrade UIWidgets to above 2.0
+     */
+    public class StreamMultiUtils<T>
+    {
+        public static Stream<T> multi(Action<MultiStreamController<T>> onListen, bool isBroadcast = false) {
+            return new _MultiStream<T>(onListen, isBroadcast);
+        }
+    }
+    
+    public interface MultiStreamController<T> : IStreamController<T> {
+        void addSync(T value);
+
+        void addErrorSync(object error, string trackStack);
+
+        void closeSync();
+    }
+    
+    class _MultiStream<T> : Stream<T> {
+        public override bool isBroadcast {
+            get {
+                return _isBroadcast;
+            }
+        }
+
+        bool _isBroadcast;
+
+        /// The callback called for each listen.
+        public readonly Action<MultiStreamController<T>> _onListen;
+
+        public _MultiStream(Action<MultiStreamController<T>> _onListen, bool isBroadcast) {
+            _isBroadcast = isBroadcast;
+            this._onListen = _onListen;
+        }
+
+        public override StreamSubscription<T> listen(Action<T> onData, Action<object, string> onError = null,
+            Action onDone = null, bool cancelOnError = false) {
+            var controller = new _MultiStreamController<T>();
+            controller.onListen = () => {
+                _onListen(controller);
+            };
+            return controller._subscribe(
+                onData, onError, onDone, cancelOnError);
+        }
+    }
+
+    class _MultiStreamController<T> : _AsyncStreamController<T>, MultiStreamController<T> {
+        public _MultiStreamController() : base(null, null, null, null)
+        {
+        }
+
+        public void addSync(T value) {
+            if (!_mayAddEvent) throw _badEventState();
+            if (hasListener) _subscription._add(value);
+        }
+
+        public void addErrorSync(object error, string trackStack) {
+            if (!_mayAddEvent) throw _badEventState();
+            if (hasListener) {
+                _subscription._addError(error, trackStack ?? "");
+            }
+        }
+
+        public void closeSync() {
+            if (isClosed) return;
+            if (!_mayAddEvent) throw _badEventState();
+            _state |= _StreamController<T>._STATE_CLOSED;
+            if (hasListener) _subscription._close();
+        }
+
+        public override Stream<T> stream {
+            get {
+                throw new Exception("Not available");
+            }
+        }
+    }
+}

com.unity.uiwidgets/Runtime/async/stream_multi.cs.meta

+fileFormatVersion: 2
+guid: 5bf5bf2fe1c7407db0a476f2bbcd01f4
+timeCreated: 1629701152