Skip to content

Commit 929ed27

Browse files
budgetpreneurbudgetpreneur
committed
Merge branch 'ee/integration' of https://github.com/pubnub/c-sharp into ee/integration
2 parents f156472 + 8e1b942 commit 929ed27

File tree

7 files changed

+52
-74
lines changed

7 files changed

+52
-74
lines changed

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

+43-49
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,9 @@
77
using System.Net;
88
using System.Globalization;
99
using PubnubApi.PubnubEventEngine;
10-
10+
using PubnubApi.EventEngine.Subscribe;
11+
using PubnubApi.EventEngine.Subscribe.Events;
12+
using PubnubApi;
1113

1214
namespace PubnubApi.EndPoint
1315
{
@@ -28,13 +30,16 @@ public class SubscribeOperation2<T>: ISubscribeOperation<T>
2830
private Dictionary<string, object> queryParam;
2931
private PubnubEventEngine.EventEngine pnEventEngine;
3032
private Pubnub PubnubInstance;
31-
public List<SubscribeCallback> SubscribeListenerList
33+
private SubscribeEventEngine subscribeEventEngine;
34+
public SubscribeEventEngineFactory subscribeEventEngineFactory;
35+
public string instanceId;
36+
public List<SubscribeCallback> SubscribeListenerList
3237
{
3338
get;
3439
set;
3540
} = new List<SubscribeCallback>();
3641

37-
public SubscribeOperation2(PNConfiguration pubnubConfig, IJsonPluggableLibrary jsonPluggableLibrary, IPubnubUnitTest pubnubUnit, IPubnubLog log, EndPoint.TelemetryManager telemetryManager, EndPoint.TokenManager tokenManager, Pubnub instance)
42+
public SubscribeOperation2(PNConfiguration pubnubConfig, IJsonPluggableLibrary jsonPluggableLibrary, IPubnubUnitTest pubnubUnit, IPubnubLog log, EndPoint.TelemetryManager telemetryManager, EndPoint.TokenManager tokenManager,SubscribeEventEngineFactory subscribeEventEngineFactory, string instanceId, Pubnub instance)
3843
{
3944
PubnubInstance = instance;
4045
config = pubnubConfig;
@@ -43,6 +48,8 @@ public SubscribeOperation2(PNConfiguration pubnubConfig, IJsonPluggableLibrary j
4348
pubnubLog = log;
4449
pubnubTelemetryMgr = telemetryManager;
4550
pubnubTokenMgr = tokenManager;
51+
this.subscribeEventEngineFactory = subscribeEventEngineFactory;
52+
this.instanceId = instanceId;
4653

4754
var eventEmitter = new EventEmitter();
4855
eventEmitter.RegisterJsonListener(JsonCallback);
@@ -772,56 +779,29 @@ public void Execute()
772779

773780
private void Subscribe(string[] channels, string[] channelGroups, Dictionary<string, object> externalQueryParam)
774781
{
775-
if ((channels == null || channels.Length == 0) && (channelGroups == null || channelGroups.Length == 0))
782+
Action<Pubnub, PNStatus> statusListener = null;
783+
Action<Pubnub, PNMessageResult<T>> messageListener = null;
784+
if ((channels == null || channels.Length == 0) && (channelGroups == null || channelGroups.Length == 0))
776785
{
777-
throw new ArgumentException("Either Channel Or Channel Group or Both should be provided.");
778-
}
779-
780-
string channel = (channels != null) ? string.Join(",", channels.OrderBy(x => x).ToArray()) : "";
781-
string channelGroup = (channelGroups != null) ? string.Join(",", channelGroups.OrderBy(x => x).ToArray()) : "";
782-
783-
PNPlatform.Print(config, pubnubLog);
786+
throw new ArgumentException("Either Channel Or Channel Group or Both should be provided.");
787+
}
784788

785-
LoggingMethod.WriteToLog(pubnubLog, string.Format(CultureInfo.InvariantCulture, "DateTime {0}, requested subscribe for channel(s)={1} and channel group(s)={2}", DateTime.Now.ToString(CultureInfo.InvariantCulture), channel, channelGroup), config.LogVerbosity);
786-
787-
Dictionary<string, string> initialSubscribeUrlParams = new Dictionary<string, string>();
788-
if (this.subscribeTimetoken >= 0)
789+
if (this.subscribeEventEngineFactory.hasEventEngine(instanceId))
789790
{
790-
initialSubscribeUrlParams.Add("tt", this.subscribeTimetoken.ToString(CultureInfo.InvariantCulture));
791-
}
792-
if (!string.IsNullOrEmpty(config.FilterExpression) && config.FilterExpression.Trim().Length > 0)
791+
subscribeEventEngine = subscribeEventEngineFactory.getEventEngine(instanceId);
792+
}
793+
else
793794
{
794-
initialSubscribeUrlParams.Add("filter-expr", UriUtil.EncodeUriComponent(config.FilterExpression, PNOperationType.PNSubscribeOperation, false, false, false));
795-
}
796-
795+
if (SubscribeListenerList != null && SubscribeListenerList.Count > 0) {
796+
messageListener = MessageEmitter;
797+
statusListener = StatusEmitter;
798+
}
799+
var subscribeManager = new SubscribeManager2(config, jsonLibrary, unit, pubnubLog, pubnubTelemetryMgr, pubnubTokenMgr, PubnubInstance);
800+
subscribeEventEngine = subscribeEventEngineFactory.initializeEventEngine(instanceId, PubnubInstance, config, subscribeManager, statusListener, messageListener);
797801

798-
#if NETFX_CORE || WINDOWS_UWP || UAP || NETSTANDARD10 || NETSTANDARD11 || NETSTANDARD12
799-
Task.Factory.StartNew(() =>
800-
{
801-
manager = new SubscribeManager2(config, jsonLibrary, unit, pubnubLog, pubnubTelemetryMgr, pubnubTokenMgr, PubnubInstance);
802-
//manager.CurrentPubnubInstance(PubnubInstance);
803-
pnEventEngine.Subscribe(channels.ToList<string>(), channelGroups.ToList<string>());
804-
//manager = new SubscribeManager2(config, jsonLibrary, unit, pubnubLog, pubnubTelemetryMgr, pubnubTokenMgr, PubnubInstance);
805-
//manager.CurrentPubnubInstance(PubnubInstance);
806-
//manager.MultiChannelSubscribeInit<T>(PNOperationType.PNSubscribeOperation, channels, channelGroups, initialSubscribeUrlParams, externalQueryParam);
807-
}, CancellationToken.None, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false);
808-
#else
809-
new Thread(() =>
810-
{
811-
manager = new SubscribeManager2(config, jsonLibrary, unit, pubnubLog, pubnubTelemetryMgr, pubnubTokenMgr, PubnubInstance);
812-
//manager.CurrentPubnubInstance(PubnubInstance);
813-
if (pnEventEngine.CurrentState == null)
814-
{
815-
pnEventEngine.InitialState(new State(StateType.Unsubscribed) { EventType = EventType.SubscriptionChanged});
816-
}
817-
pnEventEngine.Subscribe(channels.ToList<string>(), channelGroups.ToList<string>());
818-
//manager = new SubscribeManager2(config, jsonLibrary, unit, pubnubLog, pubnubTelemetryMgr, pubnubTokenMgr, PubnubInstance);
819-
//manager.CurrentPubnubInstance(PubnubInstance);
820-
//manager.MultiChannelSubscribeInit<T>(PNOperationType.PNSubscribeOperation, channels, channelGroups, initialSubscribeUrlParams, externalQueryParam);
821-
})
822-
{ IsBackground = true }.Start();
823-
#endif
824-
}
802+
}
803+
subscribeEventEngine.eventQueue.Enqueue(new SubscriptionChangedEvent() { Channels = channels, ChannelGroups = channelGroups });
804+
}
825805

826806
internal bool Retry(bool reconnect)
827807
{
@@ -863,5 +843,19 @@ internal void CurrentPubnubInstance(Pubnub instance)
863843
{
864844
PubnubInstance = instance;
865845
}
866-
}
846+
private void MessageEmitter<T>(Pubnub pubnubInstance, PNMessageResult<T> messageResult)
847+
{
848+
foreach (var listener in SubscribeListenerList) {
849+
listener.Message(pubnubInstance, messageResult);
850+
}
851+
}
852+
853+
private void StatusEmitter(Pubnub pubnubInstance, PNStatus status)
854+
{
855+
foreach (var listener in SubscribeListenerList) {
856+
listener.Status(pubnubInstance, status);
857+
}
858+
}
859+
860+
}
867861
}

‎src/Api/PubnubApi/EventEngine/Subscribe/SubscribeEventEngineFactory.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
namespace PubnubApi.EventEngine.Subscribe
66
{
7-
internal class SubscribeEventEngineFactory
7+
public class SubscribeEventEngineFactory
88
{
99
private ConcurrentDictionary<string, SubscribeEventEngine> engineinstances;
1010
internal SubscribeEventEngineFactory()

‎src/Api/PubnubApi/Pubnub.cs

+4-20
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@ public class Pubnub
1616
private ConcurrentDictionary<string, PNConfiguration> pubnubConfig { get; } = new ConcurrentDictionary<string, PNConfiguration>();
1717
private IPubnubUnitTest pubnubUnitTest;
1818
private IPubnubLog pubnubLog;
19-
private SubscribeEventEngineFactory subscribeEventEngineFactory;
2019
private EndPoint.ListenerManager listenerManager;
2120
private readonly EndPoint.TelemetryManager telemetryManager;
2221
private readonly EndPoint.TokenManager tokenManager;
@@ -46,24 +45,9 @@ public ISubscribeOperation<T> Subscribe<T>()
4645
{
4746
if (pubnubConfig[InstanceId].EnableEventEngine)
4847
{
49-
SubscribeEventEngine subscribeEventEngine;
50-
Action<Pubnub, PNStatus> statusListener = null;
51-
Action<Pubnub, PNMessageResult<T>> messageListener = null;
52-
if (this.subscribeEventEngineFactory.hasEventEngine(InstanceId)) {
53-
subscribeEventEngine = subscribeEventEngineFactory.getEventEngine(InstanceId); }
54-
else {
55-
var subscribeManager = new SubscribeManager2(pubnubConfig[InstanceId], JsonPluggableLibrary, pubnubUnitTest, pubnubLog, telemetryManager, tokenManager, this);
56-
if (subscribeCallbackListenerList!= null && subscribeCallbackListenerList.Count > 0) {
57-
messageListener = subscribeCallbackListenerList[0].Message;
58-
statusListener = subscribeCallbackListenerList[0].Status;
59-
}
60-
subscribeEventEngine = subscribeEventEngineFactory.initializeEventEngine(InstanceId, this, pubnubConfig[InstanceId], subscribeManager, statusListener, messageListener);
61-
}
62-
EndPoint.SubscribeOperation2<T> subscribeOperation = new EndPoint.SubscribeOperation2<T>(pubnubConfig.ContainsKey(InstanceId) ? pubnubConfig[InstanceId] : null, JsonPluggableLibrary, pubnubUnitTest, pubnubLog, null, tokenManager, this);
48+
EndPoint.SubscribeOperation2<T> subscribeOperation = new EndPoint.SubscribeOperation2<T>(pubnubConfig.ContainsKey(InstanceId) ? pubnubConfig[InstanceId] : null, JsonPluggableLibrary, pubnubUnitTest, pubnubLog, null, tokenManager, new SubscribeEventEngineFactory(),InstanceId ,this);
6349
subscribeOperation.SubscribeListenerList = subscribeCallbackListenerList;
64-
65-
66-
subscribeEventEngine.eventQueue.Enqueue(new SubscriptionChangedEvent() { Channels = this.GetSubscribedChannels().ToArray(), ChannelGroups = this.GetSubscribedChannelGroups().ToArray() });
50+
6751
//subscribeOperation.CurrentPubnubInstance(this);
6852
savedSubscribeOperation = subscribeOperation;
6953
return subscribeOperation;
@@ -76,6 +60,7 @@ public ISubscribeOperation<T> Subscribe<T>()
7660
return subscribeOperation;
7761
}
7862
}
63+
7964
public EndPoint.UnsubscribeOperation<T> Unsubscribe<T>()
8065
{
8166
EndPoint.UnsubscribeOperation<T> unsubscribeOperation = new EndPoint.UnsubscribeOperation<T>(pubnubConfig.ContainsKey(InstanceId) ? pubnubConfig[InstanceId] : null, JsonPluggableLibrary, pubnubUnitTest, pubnubLog, telemetryManager, tokenManager, this);
@@ -905,7 +890,6 @@ public Pubnub(PNConfiguration config)
905890
{
906891
savedSdkVerion = Version;
907892
InstanceId = Guid.NewGuid().ToString();
908-
subscribeEventEngineFactory = new SubscribeEventEngineFactory();
909893
pubnubConfig.AddOrUpdate(InstanceId, config, (k, o) => config);
910894

911895
if (config != null)
@@ -976,6 +960,6 @@ private void CheckRequiredConfigValues()
976960
}
977961
}
978962

979-
#endregion
963+
#endregion
980964
}
981965
}

‎src/UnitTests/PubnubApi.Tests/EventEngine/HandshakeFailedStateTransition.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
using NUnit.Framework;
1+
using NUnit.Framework;
22
using PubnubApi.EventEngine.Core;
33
using PubnubApi.EventEngine.Subscribe.Common;
44
using PubnubApi.EventEngine.Subscribe.Events;

‎src/UnitTests/PubnubApi.Tests/EventEngine/HandshakeReconnectingStateTransition.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
using NUnit.Framework;
1+
using NUnit.Framework;
22
using PubnubApi.EventEngine.Core;
33
using PubnubApi.EventEngine.Subscribe.Common;
44
using PubnubApi.EventEngine.Subscribe.Context;

‎src/UnitTests/PubnubApi.Tests/EventEngine/ReceivingStateTransition.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
using NUnit.Framework;
1+
using NUnit.Framework;
22
using PubnubApi.EventEngine.Core;
33
using PubnubApi.EventEngine.Subscribe.Common;
44
using PubnubApi.EventEngine.Subscribe.Context;

‎src/UnitTests/PubnubApi.Tests/EventEngine/UnsubscribedStateTransition.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
using NUnit.Framework;
1+
using NUnit.Framework;
22
using PubnubApi.EventEngine.Core;
33
using PubnubApi.EventEngine.Subscribe.Common;
44
using PubnubApi.EventEngine.Subscribe.Context;

0 commit comments

Comments
 (0)