Add more attempts & logs for tests

Author: sierpinskid

Assets/Plugins/StreamChat/Core/StreamChatClient.cs

@@ -4,6 +4,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using StreamChat.Core.Configs;
+using StreamChat.Core.Exceptions;
 using StreamChat.Core.Helpers;
 using StreamChat.Core.InternalDTO.Events;
 using StreamChat.Core.InternalDTO.Models;
@@ -53,17 +54,17 @@ namespace StreamChat.Core
     //StreamTodo: Handle restoring state after lost connection
 
     public delegate void ChannelInviteHandler(IStreamChannel channel, IStreamUser invitee);
-    
+
     /// <summary>
     /// Member added to the channel handler
     /// </summary>
     public delegate void ChannelMemberAddedHandler(IStreamChannel channel, IStreamChannelMember member);
-    
+
     /// <summary>
     /// Member removed from the channel handler
     /// </summary>
     public delegate void ChannelMemberRemovedHandler(IStreamChannel channel, IStreamChannelMember member);
-    
+
     public sealed class StreamChatClient : IStreamChatClient
     {
         public event ConnectionMadeHandler Connected;
@@ -79,7 +80,7 @@ public sealed class StreamChatClient : IStreamChatClient
         public event ChannelInviteHandler ChannelInviteReceived;
         public event ChannelInviteHandler ChannelInviteAccepted;
         public event ChannelInviteHandler ChannelInviteRejected;
-        
+
         public event ChannelMemberAddedHandler AddedToChannelAsMember;
         public event ChannelMemberRemovedHandler RemovedFromChannelAsMember;
 
@@ -115,7 +116,7 @@ public static IStreamChatClient CreateDefaultClient(IStreamClientConfig config =
             {
                 config = StreamClientConfig.Default;
             }
-            
+
             var logs = StreamDependenciesFactory.CreateLogger(config.LogLevel.ToLogLevel());
             var websocketClient
                 = StreamDependenciesFactory.CreateWebsocketClient(logs, config.LogLevel.IsDebugEnabled());
@@ -206,6 +207,7 @@ var ownUserDto
         }
 
         //StreamTodo: test scenario: ConnectUserAsync and immediately call DisconnectUserAsync
+        //StreamTodo: this should cancel token that would be globally passed to all async tasks so the moment we disconnect all async tasks are cancelled
         public Task DisconnectUserAsync()
         {
             TryCancelWaitingForUserConnection();
@@ -216,7 +218,8 @@ public Task DisconnectUserAsync()
 
         public Task<IStreamChannel> GetOrCreateChannelWithIdAsync(ChannelType channelType, string channelId,
             string name = null, IDictionary<string, object> optionalCustomData = null)
-            => InternalGetOrCreateChannelWithIdAsync(channelType, channelId, name, presence: true, state: true, watch: true, optionalCustomData);
+            => InternalGetOrCreateChannelWithIdAsync(channelType, channelId, name, presence: true, state: true,
+                watch: true, optionalCustomData);
 
         public async Task<IStreamChannel> GetOrCreateChannelWithMembersAsync(ChannelType channelType,
             IEnumerable<IStreamUser> members, IDictionary<string, object> optionalCustomData = null)
@@ -572,10 +575,11 @@ void IStreamChatClientEventsListener.Destroy()
         void IStreamChatClientEventsListener.Update() => InternalLowLevelClient.Update(_timeService.DeltaTime);
 
         internal StreamChatLowLevelClient InternalLowLevelClient { get; }
-        
+
         // We probably don't want to expose the presence, state, watch params to the public API
-        internal async Task<IStreamChannel> InternalGetOrCreateChannelWithIdAsync(ChannelType channelType, string channelId, 
-            string name = null, bool presence = true, bool state = true, bool watch = true, 
+        internal async Task<IStreamChannel> InternalGetOrCreateChannelWithIdAsync(ChannelType channelType,
+            string channelId,
+            string name = null, bool presence = true, bool state = true, bool watch = true,
             IDictionary<string, object> optionalCustomData = null)
         {
             StreamAsserts.AssertChannelTypeIsValid(channelType);
@@ -607,13 +611,13 @@ internal IStreamLocalUserData UpdateLocalUser(OwnUserInternalDTO ownUserInternal
         {
             _localUserData = _cache.TryCreateOrUpdate(ownUserInternalDto);
 
-            if(LocalUserData == null)
+            if (LocalUserData == null)
             {
                 _logs.Error("Local User Data is null");
                 return _localUserData;
             }
 
-            if(LocalUserData.ChannelMutes != null)
+            if (LocalUserData.ChannelMutes != null)
             {
                 //StreamTodo: Can we not rely on whoever called TryCreateOrUpdate to update this but make it more reliable? Better to react to some event
                 // This could be solved if ChannelMutes would be an observable collection
@@ -705,6 +709,47 @@ private void TryCancelWaitingForUserConnection()
                 _connectUserTaskSource.TrySetCanceled();
             }
         }
+        
+        private async Task InternalGetOrCreateChannelAsync(ChannelType channelType, string channelId)
+        {
+#if STREAM_TESTS_ENABLED
+            const int maxAttempts = 10;
+#else
+            const int maxAttempts = 1;
+#endif
+
+            for (int i = 1; i <= maxAttempts; i++)
+            {
+                try
+                {
+                    await GetOrCreateChannelWithIdAsync(channelType, channelId);
+                }
+                catch (StreamApiException streamException)
+                {
+                    if (!streamException.IsRateLimitExceededError() || i == maxAttempts)
+                    {
+                        throw;
+                    }
+
+                    if (ConnectionState != ConnectionState.Connected)
+                    {
+                        break;
+                    }
+
+                    var delay = 4 * i;
+#if STREAM_TESTS_ENABLED
+                    _logs.Warning($"InternalGetOrCreateChannelAsync attempt failed due to rate limit. Wait {delay} seconds and try again");
+#endif
+                    //StreamTodo: pass CancellationToken
+                    await Task.Delay(delay * 1000);
+
+                    if (ConnectionState != ConnectionState.Connected)
+                    {
+                        break;
+                    }
+                }
+            }
+        }
 
         #region Events
 
@@ -716,13 +761,14 @@ private void OnConnected(HealthCheckEventInternalDTO dto)
 
                 // This can sometimes be null. I think it's when the client lost network and believes he's reconnecting
                 // but the healthcheck timeout didn't pass on server and from the server perspective the client never disconnected
-                if(localUserDto != null)
+                if (localUserDto != null)
                 {
                     UpdateLocalUser(localUserDto);
                 }
                 else
                 {
-                    _logs.Warning("OnConnected localUserDto was NULL and current LocalUserData is " + (LocalUserData != null) + " value " + LocalUserData);
+                    _logs.Warning("OnConnected localUserDto was NULL and current LocalUserData is " +
+                                  (LocalUserData != null) + " value " + LocalUserData);
                 }
 
                 Connected?.Invoke(LocalUserData);
@@ -904,13 +950,26 @@ private void OnMarkReadNotification(NotificationMarkReadEventInternalDTO eventDt
 
         private void OnAddedToChannelNotification(NotificationAddedToChannelEventInternalDTO eventDto)
         {
-            //StreamTodo: sometimes when I run all tests the eventDto.Channel.Type is null. Inspect how different is this DTO from the channel kept in cache. If its incomplete we shouldn't update the cached value
-            if (eventDto.Channel.Type == null && eventDto.ChannelType != null)
-            {
-                eventDto.Channel.Type = eventDto.ChannelType;
-            }
+#if STREAM_TESTS_ENABLED
+            var sb = new StringBuilder();
+            sb.AppendLine("OnAddedToChannelNotification");
+            sb.AppendLine($"{nameof(eventDto.ChannelType)}: {eventDto.ChannelType}");
+            sb.AppendLine($"{nameof(eventDto.Channel.Type)}: {eventDto.Channel.Type}");
+            sb.AppendLine($"{nameof(eventDto.Channel.Id)}: {eventDto.Channel.Id}");
+            sb.AppendLine($"{nameof(eventDto.Channel.Cid)}: {eventDto.Channel.Cid}");
+#endif
 
             var channel = _cache.TryCreateOrUpdate(eventDto.Channel, out var wasCreated);
+
+#if STREAM_TESTS_ENABLED
+            sb.Length = 0;
+            sb.AppendLine("Channel returned from cache:");
+            sb.AppendLine($"{nameof(channel.Type)}: {channel.Type}");
+            sb.AppendLine($"{nameof(channel.Id)}: {channel.Id}");
+            sb.AppendLine($"{nameof(channel.Cid)}: {channel.Cid}");
+            _logs.Info(sb.ToString());
+#endif
+
             var member = _cache.TryCreateOrUpdate(eventDto.Member);
             _cache.TryCreateOrUpdate(eventDto.Member.User);
 
@@ -919,14 +978,15 @@ private void OnAddedToChannelNotification(NotificationAddedToChannelEventInterna
                 AddedToChannelAsMember?.Invoke(channel, member);
                 return;
             }
-            
+
             // Watch channel, otherwise WS events won't be received
-            GetOrCreateChannelWithIdAsync(channel.Type, channel.Id).ContinueWith(t =>
+            InternalGetOrCreateChannelAsync(channel.Type, channel.Id).ContinueWith(t =>
             {
                 if (t.IsFaulted)
                 {
                     _logs.Error($"Failed to watch channel with type: {channel.Type} & id: {channel.Id} " +
-                                $"before triggering the {nameof(AddedToChannelAsMember)} event. Inspect the following exception.");
+                                $"before triggering the {nameof(AddedToChannelAsMember)} event. Inspect the following exception: " +
+                                t.Exception);
                     _logs.Exception(t.Exception);
                     return;
                 }
@@ -938,23 +998,43 @@ private void OnAddedToChannelNotification(NotificationAddedToChannelEventInterna
         private void OnRemovedFromChannelNotification(
             NotificationRemovedFromChannelEventInternalDTO eventDto)
         {
+#if STREAM_TESTS_ENABLED
+            var sb = new StringBuilder();
+            sb.AppendLine("OnRemovedFromChannelNotification BEFORE CACHE");
+            sb.AppendLine($"{nameof(eventDto.ChannelType)}: {eventDto.ChannelType}");
+            sb.AppendLine($"{nameof(eventDto.Channel.Type)}: {eventDto.Channel.Type}");
+            sb.AppendLine($"{nameof(eventDto.Channel.Id)}: {eventDto.Channel.Id}");
+            sb.AppendLine($"{nameof(eventDto.Channel.Cid)}: {eventDto.Channel.Cid}");
+            _logs.Info(sb.ToString());
+#endif
             var channel = _cache.TryCreateOrUpdate(eventDto.Channel, out var wasCreated);
+
+#if STREAM_TESTS_ENABLED
+            sb.Length = 0;
+            sb.AppendLine("Channel returned FROM CACHE:");
+            sb.AppendLine($"{nameof(channel.Type)}: {channel.Type}");
+            sb.AppendLine($"{nameof(channel.Id)}: {channel.Id}");
+            sb.AppendLine($"{nameof(channel.Cid)}: {channel.Cid}");
+            _logs.Info(sb.ToString());
+#endif
+
             var member = _cache.TryCreateOrUpdate(eventDto.Member);
             _cache.TryCreateOrUpdate(eventDto.Member.User);
-            
+
             if (!wasCreated)
             {
                 RemovedFromChannelAsMember?.Invoke(channel, member);
                 return;
             }
-            
+
             // Watch channel, otherwise WS events won't be received
-            GetOrCreateChannelWithIdAsync(channel.Type, channel.Id).ContinueWith(t =>
+            InternalGetOrCreateChannelAsync(channel.Type, channel.Id).ContinueWith(t =>
             {
                 if (t.IsFaulted)
                 {
                     _logs.Error($"Failed to watch channel with type: {channel.Type} & id: {channel.Id} " +
-                                $"before triggering the {nameof(RemovedFromChannelAsMember)} event. Inspect the following exception.");
+                                $"before triggering the {nameof(RemovedFromChannelAsMember)} event. Inspect the following exception: " +
+                                t.Exception);
                     _logs.Exception(t.Exception);
                     return;
                 }
@@ -967,20 +1047,21 @@ private void OnInvitedNotification(NotificationInvitedEventInternalDTO eventDto)
         {
             var channel = _cache.TryCreateOrUpdate(eventDto.Channel, out var wasCreated);
             var user = _cache.TryCreateOrUpdate(eventDto.User);
-            
+
             if (!wasCreated)
             {
                 ChannelInviteReceived?.Invoke(channel, user);
                 return;
             }
-            
+
             // Watch channel, otherwise WS events won't be received
-            GetOrCreateChannelWithIdAsync(channel.Type, channel.Id).ContinueWith(t =>
+            InternalGetOrCreateChannelAsync(channel.Type, channel.Id).ContinueWith(t =>
             {
                 if (t.IsFaulted)
                 {
                     _logs.Error($"Failed to watch channel with type: {channel.Type} & id: {channel.Id} " +
-                                $"before triggering the {nameof(ChannelInviteReceived)} event. Inspect the following exception.");
+                                $"before triggering the {nameof(ChannelInviteReceived)} event. Inspect the following exception: " +
+                                t.Exception);
                     _logs.Exception(t.Exception);
                     return;
                 }
@@ -993,20 +1074,21 @@ private void OnInviteAcceptedNotification(NotificationInviteAcceptedEventInterna
         {
             var channel = _cache.TryCreateOrUpdate(eventDto.Channel, out var wasCreated);
             var user = _cache.TryCreateOrUpdate(eventDto.User);
-            
+
             if (!wasCreated)
             {
                 ChannelInviteAccepted?.Invoke(channel, user);
                 return;
             }
-            
+
             // Watch channel, otherwise WS events won't be received
-            GetOrCreateChannelWithIdAsync(channel.Type, channel.Id).ContinueWith(t =>
+            InternalGetOrCreateChannelAsync(channel.Type, channel.Id).ContinueWith(t =>
             {
                 if (t.IsFaulted)
                 {
                     _logs.Error($"Failed to watch channel with type: {channel.Type} & id: {channel.Id} " +
-                                $"before triggering the {nameof(ChannelInviteAccepted)} event. Inspect the following exception.");
+                                $"before triggering the {nameof(ChannelInviteAccepted)} event. Inspect the following exception: " +
+                                t.Exception);
                     _logs.Exception(t.Exception);
                     return;
                 }
@@ -1019,20 +1101,21 @@ private void OnInviteRejectedNotification(NotificationInviteRejectedEventInterna
         {
             var channel = _cache.TryCreateOrUpdate(eventDto.Channel, out var wasCreated);
             var user = _cache.TryCreateOrUpdate(eventDto.User);
-            
+
             if (!wasCreated)
             {
                 ChannelInviteRejected?.Invoke(channel, user);
                 return;
             }
-            
+
             // Watch channel, otherwise WS events won't be received
-            GetOrCreateChannelWithIdAsync(channel.Type, channel.Id).ContinueWith(t =>
+            InternalGetOrCreateChannelAsync(channel.Type, channel.Id).ContinueWith(t =>
             {
                 if (t.IsFaulted)
                 {
                     _logs.Error($"Failed to watch channel with type: {channel.Type} & id: {channel.Id} " +
-                                $"before triggering the {nameof(ChannelInviteRejected)} event. Inspect the following exception.");
+                                $"before triggering the {nameof(ChannelInviteRejected)} event. Inspect the following exception: " +
+                                t.Exception);
                     _logs.Exception(t.Exception);
                     return;
                 }

Assets/Plugins/StreamChat/Tests/LowLevelClient/Integration/BaseIntegrationTests.cs

@@ -105,12 +105,21 @@ protected static async Task<T> Try<T>(Func<Task<T>> task, Predicate<T> successCo
                 catch (StreamApiException streamApiException)
                 {
                     // Check for "Too many requests" error
-                    if (streamApiException.StatusCode == 429)
+                    if (streamApiException.IsRateLimitExceededError())
                     {
-                        const int tooManyRequestsDelay = 4;
+                        var tooManyRequestsDelay = 4 * attempt;
                         Debug.Log($"Wait {tooManyRequestsDelay} seconds due to \"too many requests\" error");
                         await Task.Delay(tooManyRequestsDelay * 1000);
+                        continue;
+                    }
+
+                    // upstream request timeout - often received when running tests via docker
+                    if (streamApiException.Code == 504)
+                    {
+                        continue;
                     }
+                    
+                    Debug.LogException(streamApiException);
                 }
 
                 if (successCondition(response))