happy path contract tests

pubnub · Aug 22, 2023 · e859a53 · e859a53
1 parent 1f7d822
commit e859a53
Show file tree

Hide file tree

Showing 13 changed files with 157 additions and 103 deletions.
diff --git a/src/Api/PubnubApi/EndPoint/PubSub/SubscribeEndpoint.cs b/src/Api/PubnubApi/EndPoint/PubSub/SubscribeEndpoint.cs
@@ -150,28 +150,49 @@ private void Subscribe(string[] channels, string[] channelGroups, Dictionary<str
 
         private void SubscribeEventEngine_OnEffectDispatch(IEffectInvocation obj)
         {
-            unit?.EventTypeList.Add(new KeyValuePair<string, string>("invocation", obj?.Name));
-            LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnEffectDispatch : CurrentState = {subscribeEventEngine.CurrentState.GetType().Name} => Invocation = {obj.GetType().Name}", config.LogVerbosity);
+            try
+            {
+                unit?.EventTypeList.Add(new KeyValuePair<string, string>("invocation", obj?.Name));
+                LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnEffectDispatch : CurrentState = {subscribeEventEngine.CurrentState.GetType().Name} => Invocation = {obj.GetType().Name}", config.LogVerbosity);
+            }
+            catch (Exception ex)
+            {
+                LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnEffectDispatch : CurrentState = {subscribeEventEngine.CurrentState.GetType().Name} => EXCEPTION = {ex}", config.LogVerbosity);
+            }
         }
 
         private void SubscribeEventEngine_OnEventQueued(IEvent @event)
         {
-            unit?.EventTypeList.Add(new KeyValuePair<string, string>("event", @event?.Name));
-            int attempts = 0;
-            if (subscribeEventEngine.CurrentState is HandshakeReconnectingState handshakeReconnectingState)
+            try
             {
-                attempts = handshakeReconnectingState.AttemptedRetries;
+                unit?.EventTypeList.Add(new KeyValuePair<string, string>("event", @event?.Name));
+                int attempts = 0;
+                if (subscribeEventEngine.CurrentState is HandshakeReconnectingState handshakeReconnectingState)
+                {
+                    attempts = handshakeReconnectingState.AttemptedRetries;
+                }
+                else if (subscribeEventEngine.CurrentState is ReceiveReconnectingState receiveReconnectingState)
+                {
+                    attempts = receiveReconnectingState.AttemptedRetries;
+                }
+                LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnEventQueued : CurrentState: {subscribeEventEngine.CurrentState.GetType().Name}; Event = {@event.GetType().Name}; Attempt = {attempts} of {config.ConnectionMaxRetries}", config.LogVerbosity);
             }
-            else if (subscribeEventEngine.CurrentState is ReceiveReconnectingState receiveReconnectingState)
+            catch(Exception ex)
             {
-                attempts = receiveReconnectingState.AttemptedRetries;
+                LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnEventQueued : CurrentState = {subscribeEventEngine.CurrentState.GetType().Name} => EXCEPTION = {ex}", config.LogVerbosity);
             }
-            LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnEventQueued : CurrentState: {subscribeEventEngine.CurrentState.GetType().Name}; Event = {@event.GetType().Name}; Attempt = {attempts} of {config.ConnectionMaxRetries}", config.LogVerbosity);
         }
 
         private void SubscribeEventEngine_OnStateTransition(EventEngine.Core.TransitionResult obj)
         {
-            LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnStateTransition : CurrentState = {subscribeEventEngine.CurrentState.GetType().Name} => Transition State = {obj?.State.GetType().Name}", config.LogVerbosity);
+            try
+            {
+                LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnStateTransition : CurrentState = {subscribeEventEngine.CurrentState.GetType().Name} => Transition State = {obj?.State.GetType().Name}", config.LogVerbosity);
+            }
+            catch(Exception ex)
+            {
+                LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, EE OnStateTransition : CurrentState = {subscribeEventEngine.CurrentState.GetType().Name} => EXCEPTION = {ex}", config.LogVerbosity);
+            }
         }
 
 		private void MessageEmitter<T>(Pubnub pubnubInstance, PNMessageResult<T> messageResult)

diff --git a/src/Api/PubnubApi/EndPoint/PubSub/SubscribeManager2.cs b/src/Api/PubnubApi/EndPoint/PubSub/SubscribeManager2.cs
@@ -310,7 +310,10 @@ internal protected async Task<Tuple<string, PNStatus>> UrlProcessRequest<T>(Uri
                     jsonString = await pubnubHttp.SendRequestAndGetJsonResponse(requestUri, pubnubRequestState, request).ConfigureAwait(false);
                 }
 #endif
-
+                if (pubnubLog != null && config != null)
+                {
+                    LoggingMethod.WriteToLog(pubnubLog, $"DateTime {DateTime.Now.ToString(CultureInfo.InvariantCulture)}, JSON= {jsonString} for request={requestUri}", config.LogVerbosity);
+                }
                 PNStatus errStatus = GetStatusIfError<T>(pubnubRequestState, jsonString);
                 return new Tuple<string, PNStatus>((errStatus == null) ? jsonString : "", errStatus);
             }

diff --git a/src/Api/PubnubApi/EventEngine/Subscribe/Effects/EmitMessagesHandler.cs b/src/Api/PubnubApi/EventEngine/Subscribe/Effects/EmitMessagesHandler.cs
@@ -36,7 +36,7 @@ public EmitMessagesHandler(Pubnub pubnubInstance,
 
         public async override Task Run(EmitMessagesInvocation invocation)
         {
-            var processedMessages = invocation.Messages?.Messages.Select(m =>
+            var processedMessages = invocation.Messages?.Messages?.Select(m =>
             {
                 var msgResult = new PNMessageResult<object>()
                 {

diff --git a/src/Api/PubnubApi/EventEngine/Subscribe/Effects/HandshakeEffectHandler.cs b/src/Api/PubnubApi/EventEngine/Subscribe/Effects/HandshakeEffectHandler.cs
@@ -11,6 +11,7 @@
 using PubnubApi.EventEngine.Subscribe.Events;
 using PubnubApi.EventEngine.Subscribe.Invocations;
 using PubnubApi.EventEngine.Subscribe.Common;
+using System.Diagnostics;
 
 namespace PubnubApi.EventEngine.Subscribe.Effects
 {
@@ -55,7 +56,7 @@ public override async Task Run(HandshakeInvocation invocation)
                     eventQueue.Enqueue(new Events.HandshakeReconnectSuccessEvent() { Cursor = cursor, Status = response.Item2 });
                     break;
                 case { } when response.Item2.Error:
-                    eventQueue.Enqueue(new Events.HandshakeFailureEvent() { Status = response.Item2});
+                    eventQueue.Enqueue(new Events.HandshakeFailureEvent() { Status = response.Item2, Channels = invocation.Channels, ChannelGroups = invocation.ChannelGroups});
                     break;
                 case { }:
                     eventQueue.Enqueue(new Events.HandshakeSuccessEvent() { Cursor = cursor, Status = response.Item2 });
@@ -107,16 +108,23 @@ internal HandshakeReconnectEffectHandler(SubscribeManager2 manager, EventQueue e
 
         public override async Task Run(HandshakeReconnectInvocation invocation)
         {
-            if (!ReconnectionDelayUtil.shouldRetry(invocation.ReconnectionConfiguration, invocation.AttemptedRetries))
+            try
             {
-                eventQueue.Enqueue(new HandshakeReconnectGiveUpEvent() { Status = new PNStatus(PNStatusCategory.PNCancelledCategory) });
+                if (!ReconnectionDelayUtil.shouldRetry(invocation.ReconnectionConfiguration, invocation.AttemptedRetries))
+                {
+                    eventQueue.Enqueue(new HandshakeReconnectGiveUpEvent() { Status = new PNStatus(new Exception(""), PNOperationType.PNSubscribeOperation, PNStatusCategory.PNCancelledCategory, invocation.Channels, invocation.ChannelGroups ) });
+                }
+                else
+                {
+                    retryDelay = new Delay(ReconnectionDelayUtil.CalculateDelay(invocation.ReconnectionConfiguration.ReconnectionPolicy, invocation.AttemptedRetries));
+                    await retryDelay.Start();
+                    if (!retryDelay.Cancelled)
+                        await handshakeEffectHandler.Run(invocation as HandshakeInvocation);
+                }
             }
-            else
+            catch (Exception ex)
             {
-                retryDelay = new Delay(ReconnectionDelayUtil.CalculateDelay(invocation.ReconnectionConfiguration.ReconnectionPolicy, invocation.AttemptedRetries));
-                await retryDelay.Start();
-                if (!retryDelay.Cancelled)
-                    await handshakeEffectHandler.Run(invocation as HandshakeInvocation);
+                Debug.WriteLine(ex);
             }
         }
 

diff --git a/src/Api/PubnubApi/EventEngine/Subscribe/Effects/ReceivingEffectHandler.cs b/src/Api/PubnubApi/EventEngine/Subscribe/Effects/ReceivingEffectHandler.cs
@@ -45,7 +45,7 @@ public override async Task Run(ReceiveMessagesInvocation invocation)
                     eventQueue.Enqueue(new Events.ReceiveReconnectFailureEvent() { AttemptedRetries = (reconnectInvocation.AttemptedRetries + 1) % int.MaxValue, Status = response.Item2});
                     break;
                 case Invocations.ReceiveReconnectInvocation reconnectInvocation:
-                    eventQueue.Enqueue(new Events.ReceiveReconnectSuccessEvent() { Channels = invocation?.Channels, ChannelGroups = invocation?.ChannelGroups, Cursor = cursor, Status = response.Item2 });
+                    eventQueue.Enqueue(new Events.ReceiveReconnectSuccessEvent() { Channels = invocation?.Channels, ChannelGroups = invocation?.ChannelGroups, Cursor = cursor, Status = response.Item2, Messages = response.Item1 });
                     break;
                 case { } when response.Item2.Error:
                     eventQueue.Enqueue(new Events.ReceiveFailureEvent() { Cursor = invocation.Cursor, Status = response.Item2});

diff --git a/src/Api/PubnubApi/EventEngine/Subscribe/Events/SubscriptionEvents.cs b/src/Api/PubnubApi/EventEngine/Subscribe/Events/SubscriptionEvents.cs
@@ -22,28 +22,28 @@ public class SubscriptionRestoredEvent : Core.IEvent {
 	public class HandshakeSuccessEvent : Core.IEvent {
 		public SubscriptionCursor Cursor;
 		public PNStatus Status;
-		public string Name { get; set; } = "HANDSHAKE_SUCCESS";
+		public virtual string Name { get; set; } = "HANDSHAKE_SUCCESS";
 	}
 
 	public class HandshakeFailureEvent : Core.IEvent {
 		public IEnumerable<string> Channels;
 		public IEnumerable<string> ChannelGroups;
 		public PNStatus Status;
 		public int AttemptedRetries;
-		public string Name { get; set; } = "HANDSHAKE_FAILURE";
+		public virtual string Name { get; set; } = "HANDSHAKE_FAILURE";
 	}
 
 	public class HandshakeReconnectSuccessEvent : HandshakeSuccessEvent {
 		public PNStatus Status;
 		public SubscriptionCursor Cursor;
-		public new string Name { get; set; } = "HANDSHAKE_RECONNECT_SUCCESS";
+		public override string Name { get; set; } = "HANDSHAKE_RECONNECT_SUCCESS";
 	}
 
 	public class HandshakeReconnectFailureEvent : HandshakeFailureEvent
 	{
 		public IEnumerable<string> Channels;
 		public IEnumerable<string> ChannelGroups;
-		public new string Name { get; set; } = "HANDSHAKE_RECONNECT_FAILURE";
+		public override string Name { get; set; } = "HANDSHAKE_RECONNECT_FAILURE";
 	}
 
 	public class HandshakeReconnectGiveUpEvent : Core.IEvent {
@@ -57,22 +57,22 @@ public class ReceiveSuccessEvent : Core.IEvent {
 		public ReceivingResponse<object> Messages;
 		public SubscriptionCursor Cursor;
 		public PNStatus Status;
-		public string Name { get; set; } = "RECEIVE_SUCCESS";
+		public virtual string Name { get; set; } = "RECEIVE_SUCCESS";
 	}
 
 	public class ReceiveFailureEvent : Core.IEvent {
 		public PNStatus Status;
 		public int AttemptedRetries;
 		public SubscriptionCursor Cursor;
-		public string Name { get; set; } = "RECEIVE_FAILURE";
+		public virtual string Name { get; set; } = "RECEIVE_FAILURE";
 	}
 
 	public class ReceiveReconnectSuccessEvent : ReceiveSuccessEvent {
-		public new string Name { get; set; } = "RECEIVE_RECONNECT_SUCCESS";
+		public override string Name { get; set; } = "RECEIVE_RECONNECT_SUCCESS";
 	}
 
 	public class ReceiveReconnectFailureEvent : ReceiveFailureEvent {
-		public new string Name { get; set; } = "RECEIVE_RECONNECT_FAILURE";
+		public override string Name { get; set; } = "RECEIVE_RECONNECT_FAILURE";
 	}
 
 	public class ReceiveReconnectGiveUpEvent : Core.IEvent {

diff --git a/src/Api/PubnubApi/EventEngine/Subscribe/Invocations/SubscriptionInvocations.cs b/src/Api/PubnubApi/EventEngine/Subscribe/Invocations/SubscriptionInvocations.cs
@@ -47,7 +47,7 @@ public class HandshakeInvocation : Core.IEffectInvocation {
 		// TODO if we need these, figure out how to pass them.
 		public Dictionary<string, string> InitialSubscribeQueryParams = new Dictionary<string, string>();
 		public Dictionary<string, object> ExternalQueryParams = new Dictionary<string, object>();
-		public string Name { get; set; } = "HANDSHAKE";
+		public virtual string Name { get; set; } = "HANDSHAKE";
 	}
 
 	public class ReceiveMessagesInvocation : Core.IEffectInvocation 
@@ -57,32 +57,40 @@ public class ReceiveMessagesInvocation : Core.IEffectInvocation
 		public SubscriptionCursor  Cursor;
 		public Dictionary<string, string> InitialSubscribeQueryParams = new Dictionary<string, string>();
 		public Dictionary<string, object> ExternalQueryParams = new Dictionary<string, object>();
-		public string Name { get; set; } = "RECEIVE_MESSAGES";
+		public virtual string Name { get; set; } = "RECEIVE_MESSAGES";
 	}
 
 	public class CancelReceiveMessagesInvocation : ReceiveMessagesInvocation, Core.IEffectCancelInvocation 
 	{
-		public new string Name { get; set; } = "CANCEL_RECEIVE_MESSAGES";
+		public override string Name { get; set; } = "CANCEL_RECEIVE_MESSAGES";
 	}
 
 	public class CancelHandshakeInvocation : HandshakeInvocation, Core.IEffectCancelInvocation 
 	{
-		public new string Name { get; set; } = "CANCEL_HANDSHAKE";
+		public override string Name { get; set; } = "CANCEL_HANDSHAKE";
 	}
 
 	public class HandshakeReconnectInvocation: HandshakeInvocation
 	{
 		public ReconnectionConfiguration ReconnectionConfiguration;
 		public int AttemptedRetries;
+		public override string Name { get; set; } = "HANDSHAKE_RECONNECT";
 	}
 
-	public class CancelHandshakeReconnectInvocation: HandshakeReconnectInvocation, Core.IEffectCancelInvocation { }
+	public class CancelHandshakeReconnectInvocation: HandshakeReconnectInvocation, Core.IEffectCancelInvocation 
+	{ 
+		public override string Name { get; set; } = "CANCEL_HANDSHAKE_RECONNECT";
+	}
 
 	public class ReceiveReconnectInvocation: ReceiveMessagesInvocation 
 	{ 
 		public ReconnectionConfiguration ReconnectionConfiguration;
 		public int AttemptedRetries;
+		public override string Name { get; set; } = "RECEIVE_RECONNECT";
 	}
 
-	public class CancelReceiveReconnectInvocation: ReceiveReconnectInvocation, Core.IEffectCancelInvocation { }
+	public class CancelReceiveReconnectInvocation: ReceiveReconnectInvocation, Core.IEffectCancelInvocation 
+	{
+		public override string Name { get; set; } = "CANCEL_RECEIVE_RECONNECT";
+	}
 }
diff --git a/src/Api/PubnubApi/EventEngine/Subscribe/States/HandshakeReconnectingState.cs b/src/Api/PubnubApi/EventEngine/Subscribe/States/HandshakeReconnectingState.cs
@@ -59,7 +59,7 @@ public override TransitionResult Transition(IEvent e)
 					ChannelGroups = this.ChannelGroups,
 					ReconnectionConfiguration = this.ReconnectionConfiguration,
 					AttemptedRetries = (this.AttemptedRetries + 1) % int.MaxValue
-				}.With(new EmitStatusInvocation(handshakeReconnectFailure.Status)),
+				},
 
 				Events.HandshakeReconnectSuccessEvent handshakeReconnectSuccess => new ReceivingState()
 				{

diff --git a/src/Api/PubnubApi/EventEngine/Subscribe/States/ReceiveReconnectingState.cs b/src/Api/PubnubApi/EventEngine/Subscribe/States/ReceiveReconnectingState.cs
@@ -63,7 +63,10 @@ public override TransitionResult Transition(IEvent e)
                     ChannelGroups = receiveReconnectSuccess.ChannelGroups,
                     Cursor = receiveReconnectSuccess.Cursor,
                     ReconnectionConfiguration = this.ReconnectionConfiguration
-                }.With(new EmitStatusInvocation(receiveReconnectSuccess.Status)),
+                }.With(
+                    new EmitMessagesInvocation(receiveReconnectSuccess.Cursor, receiveReconnectSuccess.Messages),
+                    new EmitStatusInvocation(receiveReconnectSuccess.Status)
+                ),
 
                 Events.ReceiveReconnectFailureEvent receiveReconnectFailure => new ReceiveReconnectingState()
                 {

diff --git a/src/Api/PubnubApi/PNConfiguration.cs b/src/Api/PubnubApi/PNConfiguration.cs
@@ -184,7 +184,7 @@ private void ConstructorInit(UserId currentUserId)
             UseRandomInitializationVector = true;
             FileMessagePublishRetryLimit = 5;
             _userId = currentUserId;
-            EnableEventEngine = true;
+            EnableEventEngine = false;
             ConnectionMaxRetries = -1;
         }
         public PNConfiguration SetPresenceTimeoutWithCustomInterval(int timeout, int interval)

diff --git a/src/UnitTests/AcceptanceTests/Features/event-engine/happy-path.feature b/src/UnitTests/AcceptanceTests/Features/event-engine/happy-path.feature
@@ -1,33 +1,33 @@
-@featureSet=eventEngine
+@featureSet=eventEngine @beta
 Feature: Event Engine
   This is a description of the feature
 
   Background:
     Given the demo keyset with event engine enabled
 
-  @contract=simpleSubscribe @beta
+  @contract=simpleSubscribe
   Scenario: Successfully receive messages
     When I subscribe
-    When I publish a message
     Then I receive the message in my subscribe response
     And I observe the following:
-      | type       | name                  |
-      | event      | SUBSCRIPTION_CHANGED  |
-      | invocation | HANDSHAKE             |
-      | event      | HANDSHAKE_SUCCESS     |
-      | invocation | CANCEL_HANDSHAKE      |
-      | invocation | EMIT_STATUS           |
-      | invocation | RECEIVE_EVENTS        |
-      | event      | RECEIVE_SUCCESS       |
-      | invocation | CANCEL_RECEIVE_EVENTS |
-      | invocation | EMIT_STATUS           |
-      | invocation | EMIT_EVENTS           |
+      | type       | name                    |
+      | event      | SUBSCRIPTION_CHANGED    |
+      | invocation | HANDSHAKE               |
+      | event      | HANDSHAKE_SUCCESS       |
+      | invocation | CANCEL_HANDSHAKE        |
+      | invocation | EMIT_STATUS             |
+      | invocation | RECEIVE_MESSAGES        |
+      | event      | RECEIVE_SUCCESS         |
+      | invocation | CANCEL_RECEIVE_MESSAGES |
+      | invocation | EMIT_MESSAGES           |
+      | invocation | EMIT_STATUS             |
+      | invocation | RECEIVE_MESSAGES        |
 
-  @contract=subscribeHandshakeFailure @beta
+  @contract=subscribeHandshakeFailure
   Scenario: Complete handshake failure
     Given a linear reconnection policy with 3 retries
     When I subscribe
-    Then I receive an error
+    Then I receive an error in my subscribe response
     And I observe the following:
       | type       | name                        |
       | event      | SUBSCRIPTION_CHANGED        |
@@ -48,7 +48,7 @@ Feature: Event Engine
       | invocation | CANCEL_HANDSHAKE_RECONNECT  |
       | invocation | EMIT_STATUS                 |
 
-  @contract=subscribeHandshakeRecovery @beta
+  @contract=subscribeHandshakeRecovery
   Scenario: Handshake failure recovery
     Given a linear reconnection policy with 3 retries
     When I subscribe
@@ -66,14 +66,16 @@ Feature: Event Engine
       | event      | HANDSHAKE_RECONNECT_SUCCESS |
       | invocation | CANCEL_HANDSHAKE_RECONNECT  |
       | invocation | EMIT_STATUS                 |
-      | invocation | RECEIVE_EVENTS              |
+      | invocation | RECEIVE_MESSAGES            |
       | event      | RECEIVE_SUCCESS             |
-      | invocation | CANCEL_RECEIVE_EVENTS       |
+      | invocation | CANCEL_RECEIVE_MESSAGES     |
+      | invocation | EMIT_MESSAGES               |
       | invocation | EMIT_STATUS                 |
-      | invocation | EMIT_EVENTS                 |
+      | invocation | RECEIVE_MESSAGES            |
 
-  @contract=subscribeReceivingRecovery @beta
+  @contract=subscribeReceivingRecovery
   Scenario: Receiving failure recovery
+    Given a linear reconnection policy with 3 retries
     When I subscribe
     Then I receive the message in my subscribe response
     And I observe the following:
@@ -83,12 +85,12 @@ Feature: Event Engine
       | event      | HANDSHAKE_SUCCESS         |
       | invocation | CANCEL_HANDSHAKE          |
       | invocation | EMIT_STATUS               |
-      | invocation | RECEIVE_EVENTS            |
+      | invocation | RECEIVE_MESSAGES          |
       | event      | RECEIVE_FAILURE           |
-      | invocation | CANCEL_RECEIVE_EVENTS     |
+      | invocation | CANCEL_RECEIVE_MESSAGES   |
       | invocation | RECEIVE_RECONNECT         |
       | event      | RECEIVE_RECONNECT_SUCCESS |
       | invocation | CANCEL_RECEIVE_RECONNECT  |
+      | invocation | EMIT_MESSAGES             |
       | invocation | EMIT_STATUS               |
-      | invocation | EMIT_EVENTS               |
-      | invocation | RECEIVE_EVENTS            |
+      | invocation | RECEIVE_MESSAGES          |