New AsyncReply

Esyur.Stores.MongoDB/MongoDBStore.cs

@@ -202,6 +202,7 @@ namespace Esyur.Stores.MongoDB
                                                     x);
                 });
 
+                
                 bag.Add(av);
             }
 
@@ -222,7 +223,7 @@ namespace Esyur.Stores.MongoDB
             return rt;
         }
 
-        IAsyncReply<object> Parse(BsonValue value)
+        AsyncReply Parse(BsonValue value)
         {
             if (value.BsonType == BsonType.Document)
             {
@@ -850,8 +851,8 @@ namespace Esyur.Stores.MongoDB
                 foreach (var child in children)
                 {
                     var r = Warehouse.Get(child);
-                    if (r is IAsyncReply<T>)
+                    if (r is AsyncReply<T>)
-                        rt.Add((IAsyncReply<T>)r);
+                        rt.Add(r);// (AsyncReply<T>)r);
                 }
 
                 rt.Seal();
@@ -882,8 +883,8 @@ namespace Esyur.Stores.MongoDB
                 foreach (var parent in parents)
                 {
                     var r = Warehouse.Get(parent);
-                    if (r is IAsyncReply<T>)
+                    if (r is AsyncReply<T>)
-                        rt.Add((IAsyncReply<T>)r);
+                        rt.Add(r);// (AsyncReply<T>)r);
                 }
 
 

Esyur/Core/AsyncAwaiter.cs

@@ -14,12 +14,12 @@ namespace Esyur.Core
 
         T result;
 
-        public AsyncAwaiter(AsyncReply<T> reply)
+        public AsyncAwaiter(AsyncReply reply)
         {
             reply.Then(x =>
             {
                 this.IsCompleted = true;
-                this.result = x;
+                this.result = (T)x;
                 this.callback?.Invoke();
             }).Error(x =>
             {

Esyur/Core/AsyncBag.cs

@@ -30,23 +30,27 @@ using System.Threading.Tasks;
 
 namespace Esyur.Core
 {
-    public class AsyncBag<T>: AsyncReply<T[]>
+    public class AsyncBag: AsyncReply
     {
-        //Dictionary<AsyncReply, T> results = new Dictionary<AsyncReply, T>();
 
-        List<IAsyncReply<T>> replies = new List<IAsyncReply<T>>();
+        protected List<AsyncReply> replies = new List<AsyncReply>();
-        List<T> results = new List<T>();
+        List<object> results = new List<object>();
 
         int count = 0;
         bool sealedBag = false;
 
-        /*
+        
-        public AsyncBag<T> Then(Action<T[]> callback)
+        public AsyncBag Then(Action<object[]> callback)
         {
-            base.Then(new Action<object>(o => callback((T[])o)));
+            base.Then(new Action<object>(o => callback((object[])o)));
             return this;
         }
-        */
+
+        public new AsyncAwaiter<object[]> GetAwaiter()
+        {
+            return new AsyncAwaiter<object[]>(this);
+        }
+
 
         public void Seal()
         {
@@ -56,7 +60,7 @@ namespace Esyur.Core
             sealedBag = true;
 
             if (results.Count == 0)
-                Trigger(new T[0]);
+                Trigger(new object[0]);
 
             for (var i = 0; i < results.Count; i++)
             //foreach(var reply in results.Keys)
@@ -66,7 +70,7 @@ namespace Esyur.Core
 
                 k.Then((r) =>
                 {
-                    results[index] = (T)r;
+                    results[index] = r;
                     count++;
                     if (count == results.Count)
                         Trigger(results.ToArray());
@@ -74,17 +78,16 @@ namespace Esyur.Core
             }
         }
 
-        public void Add(IAsyncReply<T> reply)
+        public void Add(AsyncReply reply)
         {
             if (!sealedBag)
             {
-                results.Add(default(T));
+                results.Add(null);
                 replies.Add(reply);
             }
-                //results.Add(reply, default(T));            
         }
 
-        public void AddBag(AsyncBag<T> bag)
+        public void AddBag(AsyncBag bag)
         {
             foreach (var r in bag.replies)
                 Add(r);
@@ -97,10 +100,10 @@ namespace Esyur.Core
 
         }
 
-        public AsyncBag(T[] results)
+        public AsyncBag(object[] results)
+            : base(results)
         {
-            resultReady = true;
+
-            base.result = results;
         }
 
     }

Esyur/Core/AsyncBagGeneric.cs

@@ -0,0 +1,70 @@
+/*
+ 
+Copyright (c) 2017 Ahmed Kh. Zamil
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
+*/
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Esyur.Core
+{
+    public class AsyncBag<T>: AsyncBag
+    {     
+        public AsyncBag<T> Then(Action<T[]> callback)
+        {
+            base.Then(new Action<object>((o) => callback(((object[])o).Select(x=>(T)x).ToArray())));
+            return this;
+        }
+        
+
+        public void Add(AsyncReply<T> reply)
+        {
+            base.Add(reply);
+        }
+
+        public void AddBag(AsyncBag<T> bag)
+        {
+            foreach (var r in bag.replies)
+                Add(r);
+        }
+
+
+        public new AsyncAwaiter<T[]> GetAwaiter()
+        {
+            return new AsyncAwaiter<T[]>(this);
+        }
+
+        public AsyncBag()
+        {
+
+        }
+
+        public AsyncBag(T[] results)
+            : base(results.Select(x=>(object)x).ToArray())
+        {
+
+        }
+    }
+}

Esyur/Core/AsyncReply.cs

@@ -36,8 +36,301 @@ using System.Diagnostics;
 namespace Esyur.Core
 {
     [AsyncMethodBuilder(typeof(AsyncReplyBuilder))]
-    public class AsyncReply : AsyncReply<object>
+    public class AsyncReply 
+    {
+        public bool Debug = false;
+
+        protected List<Action<object>> callbacks = new List<Action<object>>();
+        protected object result;
+
+        protected List<Action<AsyncException>> errorCallbacks = new List<Action<AsyncException>>();
+
+        protected List<Action<ProgressType, int, int>> progressCallbacks = new List<Action<ProgressType, int, int>>();
+
+        protected List<Action<object>> chunkCallbacks = new List<Action<object>>();
+
+        //List<AsyncAwaiter> awaiters = new List<AsyncAwaiter>();
+
+        object asyncLock = new object();
+
+        //public Timer timeout;// = new Timer()
+        protected bool resultReady = false;
+        AsyncException exception;
+        // StackTrace trace;
+        AutoResetEvent mutex = new AutoResetEvent(false);
+
+        public static int MaxId;
+
+        public int Id;
+
+        public bool Ready
+        {
+            get { return resultReady; }
+
+        }
+
+
+        public object Wait()
+        {
+
+            if (resultReady)
+                return result;
+
+            if (Debug)
+                Console.WriteLine($"AsyncReply: {Id} Wait");
+
+            //mutex = new AutoResetEvent(false);
+            mutex.WaitOne();
+
+            if (Debug)
+                Console.WriteLine($"AsyncReply: {Id} Wait ended");
+
+
+            return result;
+        }
+
+
+        public object Result
+        {
+            get { return result; }
+        }
+
+
+        public AsyncReply Then(Action<object> callback)
+        {
+            //lock (callbacksLock)
+            //{
+            lock (asyncLock)
+            {
+                //  trace = new StackTrace();
+
+                if (resultReady)
+                {
+                    if (Debug)
+                        Console.WriteLine($"AsyncReply: {Id} Then ready");
+
+                    callback(result);
+                    return this;
+                }
+
+
+                //timeout = new Timer(x =>
+                //{
+                //    // Get calling method name
+                //    Console.WriteLine(trace.GetFrame(1).GetMethod().Name);
+
+                //    var tr = String.Join("\r\n", trace.GetFrames().Select(f => f.GetMethod().Name));
+                //    timeout.Dispose();
+
+                //    tr = trace.ToString();
+                //    throw new Exception("Request timeout " + Id);
+                //}, null, 15000, 0);
+
+
+                if (Debug)
+                    Console.WriteLine($"AsyncReply: {Id} Then pending");
+
+
+
+                callbacks.Add(callback);
+
+                return this;
+            }
+        }
+
+
+
+        public AsyncReply Error(Action<AsyncException> callback)
+        {
+            // lock (callbacksLock)
+            //  {
+            errorCallbacks.Add(callback);
+
+            if (exception != null)
+                callback(exception);
+
+            return this;
+            //}
+        }
+
+        public AsyncReply Progress(Action<ProgressType, int, int> callback)
+        {
+            //lock (callbacksLock)
+            //{
+            progressCallbacks.Add(callback);
+            return this;
+            //}
+        }
+
+
+        public AsyncReply Chunk(Action<object> callback)
+        {
+            // lock (callbacksLock)
+            // {
+            chunkCallbacks.Add(callback);
+            return this;
+            // }
+        }
+
+        public void Trigger(object result)
+        {
+            lock (asyncLock)
+            {
+                //timeout?.Dispose();
+
+                if (Debug)
+                    Console.WriteLine($"AsyncReply: {Id} Trigger");
+
+                if (resultReady)
+                    return;
+
+                this.result = result;
+
+                resultReady = true;
+
+                //if (mutex != null)
+                mutex.Set();
+
+                foreach (var cb in callbacks)
+                    cb(result);
+
+
+                if (Debug)
+                    Console.WriteLine($"AsyncReply: {Id} Trigger ended");
+
+            }
+        }
+
+        public void TriggerError(Exception exception)
+        {
+            //timeout?.Dispose();
+
+            if (resultReady)
+                return;
+
+            if (exception is AsyncException)
+                this.exception = exception as AsyncException;
+            else
+                this.exception = new AsyncException(ErrorType.Management, 0, exception.Message);
+
+
+            // lock (callbacksLock)
+            // {
+            foreach (var cb in errorCallbacks)
+                cb(this.exception);
+            //  }
+
+            mutex?.Set();
+
+        }
+
+        public void TriggerProgress(ProgressType type, int value, int max)
+        {
+            //timeout?.Dispose();
+
+            if (resultReady)
+                return;
+
+            //lock (callbacksLock)
+            //{
+            foreach (var cb in progressCallbacks)
+                cb(type, value, max);
+
+            //}
+        }
+
+
+        public void TriggerChunk(object value)
+        {
+
+            //timeout?.Dispose();
+
+            if (resultReady)
+                return;
+
+            //lock (callbacksLock)
+            //{
+            foreach (var cb in chunkCallbacks)
+                cb(value);
+
+            //}
+        }
+
+        public AsyncAwaiter<object> GetAwaiter()
+        {
+            return new AsyncAwaiter<object>(this);
+        }
+
+
+
+        public AsyncReply()
+        {
+            //   this.Debug = true;
+            Id = MaxId++;
+        }
+
+        public AsyncReply(object result)
+        {
+            //   this.Debug = true;
+            resultReady = true;
+            this.result = result;
+
+            Id = MaxId++;
+        }
+
+        /*
+    public AsyncReply<T> Then(Action<T> callback)
+        {
+           base.Then(new Action<object>(o => callback((T)o)));
+            return this;
+        }
+
+        public void Trigger(T result)
+        {
+            Trigger((object)result);
+        }
+
+        public Task<bool> MoveNext(CancellationToken cancellationToken)
+        {
+            throw new NotImplementedException();
+        }
+
+        public void Dispose()
+        {
+         }
+
+        public AsyncReply()
         {
             
+        }
+
+        public new Task<T> Task
+        {
+            get
+            {
+                return base.Task.ContinueWith<T>((t) =>
+                {
+
+#if NETSTANDARD
+                    return (T)t.GetType().GetTypeInfo().GetProperty("Result").GetValue(t);
+#else
+                    return (T)t.GetType().GetProperty("Result").GetValue(t);
+#endif
+                });
+            }
+        }
+
+        public T Current => throw new NotImplementedException();
+
+        public AsyncReply(T result)
+            : base(result)
+        {
+
+        }
+
+ */
+
+
+
     }
 }

Esyur/Core/AsyncReplyGeneric.cs

@@ -36,26 +36,62 @@ using System.Diagnostics;
 namespace Esyur.Core
 {
     [AsyncMethodBuilder(typeof(AsyncReplyBuilder<>))]
-    public class AsyncReply<T> : IAsyncReply<T>
+    public class AsyncReply<T> : AsyncReply
     {
 
-        public bool Debug = false;
+        public AsyncReply<T> Then(Action<T> callback)
+        {
+            base.Then((x)=>callback((T)x));
+            return this;
+        }
 
-        protected List<Action<T>> callbacks = new List<Action<T>>();
+        public new AsyncReply<T> Progress(Action<ProgressType, int, int> callback)
-        protected T result;
+        {
+            base.Progress(callback);
+            return this;
+        }
 
-        protected List<Action<AsyncException>> errorCallbacks = new List<Action<AsyncException>>();
 
-        protected List<Action<ProgressType, int, int>> progressCallbacks = new List<Action<ProgressType, int, int>>();
+        public AsyncReply<T> Chunk(Action<T> callback)
+        {
+            chunkCallbacks.Add((x)=>callback((T)x));
+            return this;
+        }
 
-        protected List<Action<T>> chunkCallbacks = new List<Action<T>>();
+        public AsyncReply(T result)
+           : base(result)
+        {
+
+        }
+
+        public AsyncReply()
+            :base()
+        {
+
+        }
+
+        public new AsyncAwaiter<T> GetAwaiter()
+        {
+            return new AsyncAwaiter<T>(this);
+        }
+
+
+        /*
+        protected new List<Action> callbacks = new List<Action>();
+        protected new object result;
+
+        protected new List<Action<AsyncException>> errorCallbacks = new List<Action<AsyncException>>();
+
+        protected new List<Action<ProgressType, int, int>> progressCallbacks = new List<Action<ProgressType, int, int>>();
+
+        protected new List<Action> chunkCallbacks = new List<Action>();
 
         //List<AsyncAwaiter> awaiters = new List<AsyncAwaiter>();
 
         object asyncLock = new object();
 
         //public Timer timeout;// = new Timer()
-        protected bool resultReady = false;
+
         AsyncException exception;
         // StackTrace trace;
         AutoResetEvent mutex = new AutoResetEvent(false);
@@ -323,15 +359,12 @@ namespace Esyur.Core
 
         public T Current => throw new NotImplementedException();
 
-        public AsyncReply(T result)
-            : base(result)
-        {
        
-        }
 
  */
 
 
 
+
     }
 }

Esyur/Data/BinaryList.cs

@@ -705,7 +705,7 @@ namespace Esyur.Data
             return list.ToArray();
         }
 
-        public virtual IAsyncReply<object[]> Done()
+        public virtual AsyncReply<object[]> Done()
         {
             return null;
             // 

Esyur/Data/Codec.cs

@@ -365,7 +365,7 @@ namespace Esyur.Data
         /// <param name="connection">DistributedConnection is required in case a structure in the array holds items at the other end.</param>
         /// <param name="dataType">DataType, in case the data is not prepended with DataType</param>
         /// <returns>Structure</returns>
-        public static IAsyncReply<object> Parse(byte[] data, uint offset, DistributedConnection connection, DataType dataType = DataType.Unspecified)
+        public static AsyncReply Parse(byte[] data, uint offset, DistributedConnection connection, DataType dataType = DataType.Unspecified)
         {
             uint size;
             return Parse(data, offset, out size, connection);
@@ -380,7 +380,7 @@ namespace Esyur.Data
         /// <param name="connection">DistributedConnection is required in case a structure in the array holds items at the other end.</param>
         /// <param name="dataType">DataType, in case the data is not prepended with DataType</param>
         /// <returns>Value</returns>
-        public static IAsyncReply<object> Parse(byte[] data, uint offset, out uint size, DistributedConnection connection, DataType dataType = DataType.Unspecified)
+        public static AsyncReply Parse(byte[] data, uint offset, out uint size, DistributedConnection connection, DataType dataType = DataType.Unspecified)
         {
  
             bool isArray;
@@ -697,7 +697,7 @@ namespace Esyur.Data
             // 
             var result = (ResourceComparisonResult)data[offset++];
 
-            IAsyncReply<IResource> previous = null;
+            AsyncReply previous = null;
 
             if (result == ResourceComparisonResult.Null)
                 previous = new AsyncReply<IResource>(null);
@@ -719,7 +719,7 @@ namespace Esyur.Data
             {
                 result = (ResourceComparisonResult)data[offset++];
 
-                IAsyncReply<IResource> current = null;
+                AsyncReply current = null;
 
                 if (result == ResourceComparisonResult.Null)
                 {

Esyur/Esyur.csproj

@@ -28,6 +28,7 @@
 
   <ItemGroup>
     <Compile Remove="Core\AsyncReplyNon.cs" />
+    <Compile Remove="Core\IAsyncReply.cs" />
     <Compile Remove="Net\UDP\UDPServer.cs" />
   </ItemGroup>
 
@@ -37,6 +38,7 @@
 
   <ItemGroup>
     <None Include="Core\AsyncReplyNon.cs" />
+    <None Include="Core\IAsyncReply.cs" />
     <None Include="Net\UDP\UDPServer.cs" />
   </ItemGroup>
 

Esyur/Net/IIP/DistributedConnection.cs

@@ -114,7 +114,7 @@ namespace Esyur.Net.IIP
         /// Send data to the other end as parameters
         /// </summary>
         /// <param name="values">Values will be converted to bytes then sent.</param>
-        internal SendList SendParams(IAsyncReply<object[]> reply = null)//params object[] values)
+        internal SendList SendParams(AsyncReply<object[]> reply = null)//params object[] values)
         {
             return new SendList(this, reply);
 

Esyur/Net/IIP/DistributedConnectionProtocol.cs

@@ -50,7 +50,7 @@ namespace Esyur.Net.IIP
 
         Dictionary<Guid, ResourceTemplate> templates = new Dictionary<Guid, ResourceTemplate>();
 
-        KeyList<uint, IAsyncReply<object>> requests = new KeyList<uint, IAsyncReply<object>>();
+        KeyList<uint, AsyncReply> requests = new KeyList<uint, AsyncReply>();
 
         volatile uint callbackCounter = 0;
 
@@ -1196,10 +1196,10 @@ namespace Esyur.Net.IIP
                                         //await t;
                                         //SendParams((byte)0x90, callback, Codec.Compose(res, this));
                                     }
-                                    else if (rt is IAsyncReply<object>)// Codec.ImplementsInterface(rt.GetType(), typeof(IAsyncReply<>)))// rt.GetType().GetTypeInfo().IsGenericType
+                                    else if (rt is AsyncReply)// Codec.ImplementsInterface(rt.GetType(), typeof(IAsyncReply<>)))// rt.GetType().GetTypeInfo().IsGenericType
                                           //&& rt.GetType().GetGenericTypeDefinition() == typeof(IAsyncReply<>))
                                     {
-                                        (rt as IAsyncReply<object>).Then(res =>
+                                        (rt as AsyncReply).Then(res =>
                                         {
                                             SendReply(IIPPacket.IIPPacketAction.InvokeFunctionArrayArguments, callback)
                                                         .AddUInt8Array(Codec.Compose(res, this))
@@ -1345,9 +1345,9 @@ namespace Esyur.Net.IIP
                                          });
 
                                      }
-                                    else if (rt is IAsyncReply<object>)
+                                    else if (rt is AsyncReply)
                                      {
-                                         (rt as IAsyncReply<object>).Then(res =>
+                                         (rt as AsyncReply).Then(res =>
                                 {
                                              SendReply(IIPPacket.IIPPacketAction.InvokeFunctionNamedArguments, callback)
                                                          .AddUInt8Array(Codec.Compose(res, this))

Esyur/Net/SendList.cs

@@ -9,15 +9,15 @@ namespace Esyur.Net
     public class SendList : BinaryList
     {
         NetworkConnection connection;
-        IAsyncReply<object[]> reply;
+        AsyncReply<object[]> reply;
 
-        public SendList(NetworkConnection connection, IAsyncReply<object[]> reply)
+        public SendList(NetworkConnection connection, AsyncReply<object[]> reply)
         {
             this.reply = reply;
             this.connection = connection;
         }
 
-        public override IAsyncReply<object[]> Done()
+        public override AsyncReply<object[]> Done()
         {
             connection.Send(this.ToArray());
             return reply;

Esyur/Resource/Warehouse.cs

@@ -60,6 +60,7 @@ namespace Esyur.Resource
 
         private static Regex urlRegex = new Regex(@"^(?:([\S]*)://([^/]*)/?)");
 
+        private static object resourcesLock = new object();
 
         static KeyList<string, Func<IStore>> getSupportedProtocols()
         {
@@ -596,6 +597,7 @@ namespace Esyur.Resource
                 return false;
 
             if (resources.ContainsKey(resource.Instance.Id))
+                lock(resourcesLock)
                     resources.Remove(resource.Instance.Id);
             else
                 return false;
@@ -604,12 +606,17 @@ namespace Esyur.Resource
             {
                 stores.Remove(resource as IStore);
 
+                WeakReference<IResource>[] toBeRemoved;
+
+                lock (resourcesLock)
+                {
                     // remove all objects associated with the store
-                var toBeRemoved = resources.Values.Where(x =>
+                     toBeRemoved = resources.Values.Where(x =>
                     {
                         IResource r;
                         return x.TryGetTarget(out r) && r.Instance.Store == resource;
                     }).ToArray();
+                }
 
                 foreach (var o in toBeRemoved)
                 {