some reactive shit

DTLib.csproj

@@ -51,6 +51,10 @@
     <Compile Include="Network\OldNetwork.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="Hasher.cs" />
+    <Compile Include="Reactive\ReactiveListener.cs" />
+    <Compile Include="Reactive\ReactiveSender.cs" />
+    <Compile Include="Reactive\ReactiveStream.cs" />
+    <Compile Include="Reactive\ReactiveProvider.cs" />
     <Compile Include="SecureRandom.cs" />
     <Compile Include="FrameworkFix.cs" />
     <Compile Include="TImer.cs" />

Reactive/ReactiveListener.cs

@@ -4,30 +4,46 @@ using System.Threading.Tasks;
 
 namespace DTLib.Reactive
 {
-    public class ReactiveListener<T> : ReactiveWorker<T>
+    public class ReactiveListener<T> : ReactiveProvider<T>
     {
         public ReactiveListener() { }
         public ReactiveListener(ReactiveStream<T> stream) : base(stream) { }
-        public ReactiveListener(IEnumerable<ReactiveStream<T>> streams) : base(streams) { }
 
-        public Action<object, T> ElementAddedHandler;
+        public EventHandlerAsync<T> ElementAddedHandler;
-        public void SetHandler(Action<object, T> handler) => ReactiveWorkerMutex.Execute(() => ElementAddedHandler=handler);
+        public void SetHandler(EventHandlerAsync<T> handler)
-        public async Task ElementAdded(object s, T e) => await Task.Run(() => ElementAddedHandler?.Invoke(s, e));
+        {
+            lock (Stream) ElementAddedHandler = handler;
+        }
+        public async Task ElementAdded(T e) => await Task.Run(() => ElementAddedHandler?.Invoke(e));
 
-        public override void Join(ReactiveStream<T> stream) =>
+        public override void Join(ReactiveStream<T> stream)
-            ReactiveWorkerMutex.Execute(() =>
+        {
-            {
+            base.Join(stream);
-                Streams.Add(stream);
+            lock (Stream) stream.ElementAddedEvent += ElementAdded;
-                stream.ElementAdded+=ElementAdded;
+        }
-            });
 
-        public override void Leave(ReactiveStream<T> stream) =>
+        public override void Leave(ReactiveStream<T> stream)
-            ReactiveWorkerMutex.Execute(() =>
+        {
-            {
+            base.Leave(stream);
-                Streams.Remove(stream);
+            lock (Stream) stream.ElementAddedEvent -= ElementAdded;
-                stream.ElementAdded-=ElementAdded;
+        }
-            });
 
-        //public T GetElement()
+        public T GetFirstElement() => Stream[0];
+        public T GetLastElement() => Stream[Stream.Length - 1];
+
+        public T FindOne(Func<T, bool> condition) =>
+            /*foreach (T el in Stream)
+if (condition(el))
+return el;*/
+            default;
+
+        public List<T> FindAll(Func<T, bool> condition)
+        {
+            List<T> elements = new();
+            /*foreach (T el in Stream)
+                if (condition(el))
+                    elements.Add(el);*/
+            return elements;
+        }
     }
 }

Reactive/ReactiveProvider.cs

@@ -1,28 +1,28 @@
-using System;
+namespace DTLib.Reactive
-using System.Collections.Generic;
-
-namespace DTLib.Reactive
 {
-    public class ReactiveProvider<T> : ReactiveWorker<T>
+    public abstract class ReactiveProvider<T>
     {
+        protected ReactiveStream<T> Stream
+        {
+            get
+            { lock (_stream) return _stream; }
+            set
+            { lock (_stream) _stream = value; }
+        }
+        protected ReactiveStream<T> _stream;
 
         public ReactiveProvider() { }
-        public ReactiveProvider(ReactiveStream<T> stream) : base(stream) { }
-        public ReactiveProvider(IEnumerable<ReactiveStream<T>> streams) : base(streams) { }
 
-        event Action<T> AnnounceEvent;
+        public ReactiveProvider(ReactiveStream<T> stream) => Join(stream);
-        public void Announce(T e) => ReactiveWorkerMutex.Execute(() => AnnounceEvent.Invoke(e));
 
-        public override void Join(ReactiveStream<T> stream) => ReactiveWorkerMutex.Execute(() =>
+        public virtual void Join(ReactiveStream<T> stream)
-            {
+        {
-                Streams.Add(stream);
+            lock (Stream) Stream = stream;
-                AnnounceEvent+=stream.Add;
+        }
-            });
 
-        public override void Leave(ReactiveStream<T> stream) => ReactiveWorkerMutex.Execute(() =>
+        public virtual void Leave(ReactiveStream<T> stream)
-            {
+        {
-                Streams.Remove(stream);
+            lock (Stream) Stream = null;
-                AnnounceEvent-=stream.Add;
+        }
-            });
     }
 }

Reactive/ReactiveSender.cs

@@ -0,0 +1,14 @@
+namespace DTLib.Reactive
+{
+    public class ReactiveSender<T> : ReactiveProvider<T>
+    {
+
+        public ReactiveSender() { }
+        public ReactiveSender(ReactiveStream<T> stream) : base(stream) { }
+
+        public void Send(T e)
+        {
+            lock (Stream) Stream.Add(e);
+        }
+    }
+}

Reactive/ReactiveStream.cs

@@ -1,25 +1,38 @@
-using System;
+using System.Collections.Generic;
-using System.Collections.Generic;
 
 namespace DTLib.Reactive
 {
     public class ReactiveStream<T>
     {
-        List<(long time, T value)> Storage = new();
-        public event EventHandlerAsync<T> ElementAdded;
-        SafeMutex StorageMutex = new();
-        public int Length => StorageMutex.Execute(() => Storage.Count);
-
         public ReactiveStream() { }
 
-        public void Add(T elem)
+        List<T> _storage = new();
+        List<T> Storage
         {
-            StorageMutex.Execute(() => Storage.Add((DateTime.Now.Ticks, elem)));
+            get
-            ElementAdded?.Invoke(this, elem);
+            { lock (Storage) return _storage; }
         }
 
-        public void Get(int index) => StorageMutex.Execute(() => Storage[index]);
+        public int Length
+        {
+            get
+            { lock (Storage) return Storage.Count; }
+        }
 
-        public void Clear() => StorageMutex.Execute(() => Storage.Clear());
+        public T this[int index]
+        {
+            get
+            { lock (Storage) return Storage[index]; }
+        }
+
+        internal event EventHandlerAsync<T> ElementAddedEvent;
+
+        internal void Add(T elem)
+        {
+            lock (Storage) Storage.Add(elem);
+            ElementAddedEvent?.Invoke(elem);
+        }
+
+        internal void Clear() { lock (Storage) Storage.Clear(); }
     }
-}
+}

Reactive/ReactiveWorker.cs

@@ -1,20 +0,0 @@
-using System.Collections.Generic;
-
-namespace DTLib.Reactive
-{
-    public abstract class ReactiveWorker<T>
-    {
-        protected List<ReactiveStream<T>> Streams = new();
-        protected SafeMutex ReactiveWorkerMutex = new();
-
-        public ReactiveWorker() { }
-
-        public ReactiveWorker(ReactiveStream<T> stream) => Join(stream);
-
-        public ReactiveWorker(IEnumerable<ReactiveStream<T>> streams) =>
-            ReactiveWorkerMutex.Execute(() => { foreach(ReactiveStream<T> stream in streams) Join(stream); });
-
-        public abstract void Join(ReactiveStream<T> stream);
-        public abstract void Leave(ReactiveStream<T> stream);
-    }
-}