Eventengine/handshake handler

src/Api/PubnubApi/EndPoint/PubSub/SubscribeOperation2.cs

@@ -26,7 +26,7 @@ public class SubscribeOperation2<T>: ISubscribeOperation<T>
         private bool presenceSubscribeEnabled;
         private SubscribeManager2 manager;
         private Dictionary<string, object> queryParam;
-        private EventEngine pnEventEngine;
+        private PubnubEventEngine.EventEngine pnEventEngine;
         private Pubnub PubnubInstance;
         public List<SubscribeCallback> SubscribeListenerList
         {
@@ -104,18 +104,18 @@ public SubscribeOperation2(PNConfiguration pubnubConfig, IJsonPluggableLibrary j
             effectDispatcher.Register(EventType.ReceiveReconnectSuccess, receiveReconnectEffectHandler);
             effectDispatcher.Register(EventType.ReceiveReconnectGiveUp, receiveReconnectEffectHandler);
 
-            pnEventEngine = new EventEngine(effectDispatcher, eventEmitter);
-            pnEventEngine.PubnubUnitTest = unit;
-            pnEventEngine.Setup<T>(config);
-
-            if (pnEventEngine.PubnubUnitTest != null)
-            {
-                pnEventEngine.PubnubUnitTest.EventTypeList = new List<KeyValuePair<string, string>>();
-            }
-            else
-            {
-			    pnEventEngine.InitialState(new State(StateType.Unsubscribed) { EventType = EventType.SubscriptionChanged });
-            }
+       //     pnEventEngine = new EventEngine(effectDispatcher, eventEmitter);
+       //     pnEventEngine.PubnubUnitTest = unit;
+       //     pnEventEngine.Setup<T>(config);
+
+       //     if (pnEventEngine.PubnubUnitTest != null)
+       //     {
+       //         pnEventEngine.PubnubUnitTest.EventTypeList = new List<KeyValuePair<string, string>>();
+       //     }
+       //     else
+       //     {
+			    //pnEventEngine.InitialState(new State(StateType.Unsubscribed) { EventType = EventType.SubscriptionChanged });
+       //     }
         }
 
         private void ReceivingEffect_ReceiveRequested(object sender, ReceiveRequestEventArgs e)

src/Api/PubnubApi/EventEngine/Common/Delay.cs

@@ -0,0 +1,60 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace PubnubApi.EventEngine.Common
+{
+    public class Delay
+    {
+        public bool Cancelled { get; private set; } = false;
+        private readonly TaskCompletionSource<object> taskCompletionSource  = new TaskCompletionSource<object>();
+        private readonly object monitor = new object();
+        private readonly int milliseconds;
+
+        public Delay(int milliseconds)
+        {
+            this.milliseconds = milliseconds;
+        }
+
+        public Task Start()
+        {
+            #if NETFX_CORE || WINDOWS_UWP || UAP || NETSTANDARD10 || NETSTANDARD11 || NETSTANDARD12
+            Task taskAwaiter = Task.Factory.StartNew(AwaiterLoop);
+            taskAwaiter.Wait();
+            #else
+            Thread awaiterThread = new Thread(AwaiterLoop);
+            awaiterThread.Start();
+            #endif
+            return taskCompletionSource.Task;        }
+
+        public void Cancel()
+        {
+            lock (monitor)
+            {
+                Cancelled = true;
+                Monitor.Pulse(monitor);
+            }
+        }
+
+        private void AwaiterLoop()
+        {
+            while(true)
+            {
+                lock (monitor)
+                {
+                    if (Cancelled)
+                    {
+                        taskCompletionSource.SetCanceled();
+                        break;
+                    }
+                    Monitor.Wait(monitor, milliseconds);
+                    if (Cancelled)
+                    {
+                        taskCompletionSource.SetCanceled();
+                        break;
+                    }
+                    taskCompletionSource.SetResult(null);
+                }
+            }
+        }
+    }
+}

src/Api/PubnubApi/EventEngine/Core/EffectDispatcher.cs

@@ -2,7 +2,7 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
 
-namespace PubnubApi.PubnubEventEngine.Core {
+namespace PubnubApi.EventEngine.Core {
 	internal class EffectDispatcher {
 		// assumes 1 instance of handler - capable of managing itself
 		private readonly Dictionary<System.Type, IEffectHandler> effectInvocationHandlerMap =
@@ -18,8 +18,13 @@ public async Task Dispatch<T>(T invocation) where T : IEffectInvocation {
 
 			if (invocation is IEffectCancelInvocation) {
 				await effectInvocationHandlerMap[invocation.GetType()].Cancel();
-			} else {
-				await ((IEffectHandler<T>)effectInvocationHandlerMap[invocation.GetType()]).Run(invocation);
+			} else
+			{
+				var handler = ((IEffectHandler<T>)effectInvocationHandlerMap[invocation.GetType()]);
+				if (handler.IsBackground(invocation))
+					handler.Run(invocation).Start();
+				else
+					await handler.Run(invocation);
 			}
 		}
 

src/Api/PubnubApi/EventEngine/Core/Engine.cs

@@ -1,7 +1,7 @@
 using System.Threading.Tasks;
 using System.Collections.Generic;
 
-namespace PubnubApi.PubnubEventEngine.Core {
+namespace PubnubApi.EventEngine.Core {
 	internal abstract class Engine {
 		public EventQueue eventQueue = new EventQueue();
 

src/Api/PubnubApi/EventEngine/Core/EventEngineInterfaces.cs

@@ -2,7 +2,7 @@
 using System.Threading.Tasks;
 using System.Collections.Generic;
 
-namespace PubnubApi.PubnubEventEngine.Core {
+namespace PubnubApi.EventEngine.Core {
 
 	/// <summary>
 	/// Generic effect handler.
@@ -17,6 +17,7 @@ internal interface IEffectHandler {
 	/// <typeparam name="T">Associated invocation</typeparam>
 	internal interface IEffectHandler<in T> : IEffectHandler where T : IEffectInvocation {
 		Task Run(T invocation);
+		bool IsBackground(T invocation);
 	}
 
 	/// <summary>

src/Api/PubnubApi/EventEngine/Core/EventQueue.cs

@@ -2,7 +2,7 @@
 using System.Linq;
 using System.Threading.Tasks;
 
-namespace PubnubApi.PubnubEventEngine.Core
+namespace PubnubApi.EventEngine.Core
 {
     internal class EventQueue
     {

src/Api/PubnubApi/EventEngine/Core/Utils.cs

@@ -2,7 +2,7 @@
 using System.Threading.Tasks;
 using System.Collections.Generic;
 
-namespace PubnubApi.PubnubEventEngine.Core
+namespace PubnubApi.EventEngine.Core
 {
     internal static class Utils
     {

src/Api/PubnubApi/EventEngine/Subscribe/Common/CommonSubscribeTypes.cs

@@ -0,0 +1,85 @@
+using Newtonsoft.Json;
+
+namespace PubnubApi.EventEngine.Subscribe.Common
+{
+    public class SubscriptionCursor
+    {
+        public long? Timetoken { get; set; }
+        public int? Region { get; set; }
+    }
+
+    public class HandshakeResponse
+    {
+        [JsonProperty("t")]
+        public Timetoken Timetoken { get; set; }
+
+        [JsonProperty("m")]
+        public object[] Messages { get; set; }
+    }
+    public class HandshakeError
+    {
+        [JsonProperty("status")]
+        public int Status { get; set; }
+
+        [JsonProperty("error")]
+        public string ErrorMessage { get; set; }
+    }
+
+    public class Timetoken
+    {
+        [JsonProperty("t")]
+        public long Timestamp { get; set; }
+
+        [JsonProperty("r")]
+        public int Region { get; set; }
+
+    }
+
+	public class ReceivingResponse<T>
+	{
+		[JsonProperty("t")]
+		public Timetoken Timetoken { get; set; }
+
+		[JsonProperty("m")]
+		public Message<T>[] Messages { get; set; }
+	}
+
+	public class Message<T>
+	{
+		[JsonProperty ("a")]
+		public string Shard { get; set;}
+
+		[JsonProperty ("b")]
+		public string SubscriptionMatch { get; set;}
+
+		[JsonProperty("c")]
+		public string Channel { get; set; }
+
+		[JsonProperty("d")]
+		public T Payload { get; set; }
+
+		[JsonProperty("e")]
+		public int MessageType { get; set; }
+
+		[JsonProperty("f")]
+		public string Flags { get; set; }
+
+		[JsonProperty("i")]
+		public string IssuingClientId { get; set; }
+
+		[JsonProperty("k")]
+		public string SubscribeKey { get; set; }
+
+		[JsonProperty("o")]
+		public object OriginatingTimetoken { get; set; }
+
+		[JsonProperty("p")]
+		public object PublishMetadata { get; set; }
+
+		[JsonProperty("s")]
+		public long SequenceNumber { get; set; }
+
+		[JsonProperty("p")]
+		public Timetoken Timetoken { get; set; }
+	}
+}

src/Api/PubnubApi/EventEngine/Subscribe/Context/ReconnectionDelayUtil.cs

@@ -0,0 +1,30 @@
+using System;
+namespace PubnubApi.EventEngine.Subscribe.Context
+{
+	public static class ReconnectionDelayUtil
+	{
+		public static int CalculateDelay(PNReconnectionPolicy policy, int attempts)
+		{
+			Random numGenerator = new Random();
+			int delayValue = 0;
+			int backoff = 5;
+			switch (policy) {
+				case PNReconnectionPolicy.LINEAR:
+					delayValue = attempts * backoff + numGenerator.Next(1000);
+					break;
+				case PNReconnectionPolicy.EXPONENTIAL:
+					delayValue = (int)(Math.Pow(2, attempts - 1) * 1000 + numGenerator.Next(1000));
+					break;
+			}
+			return delayValue;
+
+		}
+
+		public static bool shouldRetry(PNReconnectionPolicy policy, int attempts, int maxAttempts)
+		{
+			if (policy == PNReconnectionPolicy.NONE) return false;
+			return maxAttempts < attempts;
+		}
+	}
+}
+

src/Api/PubnubApi/EventEngine/Subscribe/Effects/EmitMessagesHandler.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Newtonsoft.Json;
+using PubnubApi.EventEngine.Core;
+using PubnubApi.EventEngine.Subscribe.Invocations;
+
+namespace PubnubApi.EventEngine.Subscribe.Effects
+{
+    internal class EmitMessagesHandler : IEffectHandler<Invocations.EmitMessagesInvocation>
+    {
+        private readonly System.Action<Pubnub, PNMessageResult<object>> messageEmitterFunction;
+        private readonly Pubnub pubnubInstance;
+
+        public EmitMessagesHandler(Pubnub pubnubInstance,
+            System.Action<Pubnub, PNMessageResult<object>> messageEmitterFunction)
+        {
+            this.messageEmitterFunction = messageEmitterFunction;
+            this.pubnubInstance = pubnubInstance;
+        }
+
+        public async Task Run(EmitMessagesInvocation invocation)
+        {
+            var processedMessages = invocation.Messages.Messages.Select(m => new PNMessageResult<object>()
+            {
+                Channel = m.Channel,
+                Message = JsonConvert.DeserializeObject(m.Payload),
+                Subscription = m.SubscriptionMatch,
+                Timetoken = m.Timetoken.Timestamp,
+                UserMetadata = m.PublishMetadata,
+                Publisher = m.IssuingClientId
+            });
+
+            foreach (var message in processedMessages)
+            {
+                messageEmitterFunction(pubnubInstance, message);
+            }
+        }
+
+        public bool IsBackground(EmitMessagesInvocation invocation) => false;
+
+        public Task Cancel()
+        {
+            throw new NotImplementedException();
+        }
+    }
+}

src/Api/PubnubApi/EventEngine/Subscribe/Effects/EmitStatusEffectHandler.cs

@@ -0,0 +1,30 @@
+using System;
+using System.Threading.Tasks;
+using PubnubApi.EventEngine.Core;
+using PubnubApi.EventEngine.Subscribe.Invocations;
+
+namespace PubnubApi.EventEngine.Subscribe.Effects
+{
+	public class EmitStatusEffectHandler: Core.IEffectHandler<EmitStatusInvocation>
+	{
+		private readonly Action<Pubnub, PNStatus> statusEmitterFunction;
+		private readonly Pubnub pubnubInstance;
+
+		public EmitStatusEffectHandler(Pubnub pn, Action<Pubnub, PNStatus> statusEmitter)
+		{
+			this.statusEmitterFunction = statusEmitter;
+			this.pubnubInstance = pn;
+		}
+
+		public Task Cancel() => Utils.EmptyTask;
+
+		bool IEffectHandler<EmitStatusInvocation>.IsBackground(EmitStatusInvocation invocation) => false;
+
+		Task IEffectHandler<EmitStatusInvocation>.Run(EmitStatusInvocation invocation)
+		{
+			this.statusEmitterFunction(this.pubnubInstance, invocation.Status);
+			return Utils.EmptyTask;
+		}
+	}
+}
+