Eventengine/deserialization generic

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

@@ -703,7 +703,7 @@ private void Subscribe(string[] channels, string[] channelGroups, Dictionary<str
 				subscribeEventEngine = subscribeEventEngineFactory.initializeEventEngine(instanceId, PubnubInstance, config, subscribeManager, statusListener, messageListener);
 
 			}
-			subscribeEventEngine.Subscribe(channels, channelGroups);
+			subscribeEventEngine.Subscribe<T>(channels, channelGroups);
 		}
 
         internal bool Retry(bool reconnect)

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

@@ -1,4 +1,6 @@
 using System;
+using System.Collections;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Newtonsoft.Json;
@@ -11,28 +13,51 @@ public class EmitMessagesHandler : EffectHandler<Invocations.EmitMessagesInvocat
     {
         private readonly System.Action<Pubnub, PNMessageResult<object>> messageEmitterFunction;
         private readonly Pubnub pubnubInstance;
+        private readonly Dictionary<string, Type> channelTypeMap;
+        private readonly Dictionary<string, Type> channelGroupTypeMap;
 
         public EmitMessagesHandler(Pubnub pubnubInstance,
-            System.Action<Pubnub, PNMessageResult<object>> messageEmitterFunction)
+            System.Action<Pubnub, PNMessageResult<object>> messageEmitterFunction, Dictionary<string, Type> channelTypeMap = null, Dictionary<string, Type> channelGroupTypeMap = null)
         {
             this.messageEmitterFunction = messageEmitterFunction;
             this.pubnubInstance = pubnubInstance;
         }
 
         public async override Task Run(EmitMessagesInvocation invocation)
         {
-            var processedMessages = invocation.Messages?.Messages.Select(m => new PNMessageResult<object>()
+            var processedMessages = invocation.Messages?.Messages.Select(m =>
             {
-                Channel = m.Channel,
-                Message = JsonConvert.DeserializeObject(m.Payload),
-                Subscription = m.SubscriptionMatch,
-                Timetoken = m.Timetoken.Timestamp,
-                UserMetadata = m.PublishMetadata,
-                Publisher = m.IssuingClientId
-            });
+                var msgResult = new PNMessageResult<object>()
+                {
+                    Channel = m.Channel,
+                    Subscription = m.SubscriptionMatch,
+                    Timetoken = m.Timetoken.Timestamp,
+                    UserMetadata = m.PublishMetadata,
+                    Publisher = m.IssuingClientId
+                };
+
+                Type msgType, groupType;
+                if (!(channelTypeMap is null) && channelTypeMap.TryGetValue(m.Channel, out msgType))
+                {
+                    msgResult.Message = JsonConvert.DeserializeObject(m.Payload, msgType);
+                } else if (!(channelGroupTypeMap is null) && channelGroupTypeMap.TryGetValue(m.SubscriptionMatch, out groupType))
+                {
+                    msgResult.Message = JsonConvert.DeserializeObject(m.Payload, groupType);
+                }
+                else
+                {
+                    msgResult.Message = m.Payload;
+                }
 
-            processedMessages?.ToList().ForEach(message => messageEmitterFunction(pubnubInstance, message));
+                return msgResult;
+            });
 
+            if (processedMessages is null) return;
+
+            foreach (var message in processedMessages)
+            {
+                messageEmitterFunction(pubnubInstance, message);
+            }
         }
 
         public override bool IsBackground(EmitMessagesInvocation invocation) => false;

src/Api/PubnubApi/EventEngine/Subscribe/SubscribeEventEngine.cs

@@ -5,11 +5,15 @@
 using PubnubApi.EventEngine.Subscribe.Effects;
 using PubnubApi.EventEngine.Subscribe.Invocations;
 using System;
+using System.Collections.Generic;
 
 namespace PubnubApi.EventEngine.Subscribe {
 	public class SubscribeEventEngine : Engine {
 		private SubscribeManager2 subscribeManager;
 
+		private readonly Dictionary<string, Type> channelTypeMap = new Dictionary<string, Type>();
+		private readonly Dictionary<string, Type> channelGroupTypeMap = new Dictionary<string, Type>();
+
 		internal SubscribeEventEngine(Pubnub pubnubInstance,
 			PNConfiguration pubnubConfiguration,
 			SubscribeManager2 subscribeManager,
@@ -31,18 +35,32 @@ internal SubscribeEventEngine(Pubnub pubnubInstance,
 			dispatcher.Register<Invocations.ReceiveReconnectInvocation, Effects.ReceivingEffectHandler>(receiveHandler);
 			dispatcher.Register<Invocations.CancelReceiveReconnectInvocation, Effects.ReceivingEffectHandler>(receiveHandler);
 
-			var emitMessageHandler = new Effects.EmitMessagesHandler(pubnubInstance, messageListener);
+			var emitMessageHandler = new Effects.EmitMessagesHandler(pubnubInstance, messageListener, channelTypeMap, channelGroupTypeMap);
 			dispatcher.Register<Invocations.EmitMessagesInvocation, Effects.EmitMessagesHandler>(emitMessageHandler);
 
 			var emitStatusHandler = new Effects.EmitStatusEffectHandler(pubnubInstance, statusListener);
 			dispatcher.Register<Invocations.EmitStatusInvocation, Effects.EmitStatusEffectHandler>(emitStatusHandler);
 
 			currentState = new UnsubscribedState() { ReconnectionConfiguration = new Context.ReconnectionConfiguration(pubnubConfiguration.ReconnectionPolicy, pubnubConfiguration.ConnectionMaxRetries) };
 		}
-		public void Subscribe(string[] channels, string[] channelGroups)
+		public void Subscribe<T>(string[] channels, string[] channelGroups)
 		{
+			foreach (var c in channels)
+			{
+				channelTypeMap[c] = typeof(T);
+			}
+			foreach (var c in channelGroups)
+			{
+				channelGroupTypeMap[c] = typeof(T);
+			}
 			this.EventQueue.Enqueue(new SubscriptionChangedEvent() { Channels = channels, ChannelGroups = channelGroups });
 		}
+
+		public void Subscribe(string[] channels, string[] channelGroups)
+		{
+			Subscribe<string>(channels, channelGroups);
+		}
+
 		public void UnsubscribeAll()
 		{
 			this.EventQueue.Enqueue(new UnsubscribeAllEvent());