Decouple notifications websocket handling from chat operations

osu.Game.Tests/Chat/TestSceneChannelManager.cs

@@ -75,8 +75,6 @@ public void SetUpSteps()
                     return false;
                 };
             });
-
-            AddUntilStep("wait for notifications client", () => channelManager.NotificationsConnected);
         }
 
         [Test]

osu.Game/Online/API/APIAccess.cs

@@ -21,7 +21,7 @@
 using osu.Game.Localisation;
 using osu.Game.Online.API.Requests;
 using osu.Game.Online.API.Requests.Responses;
-using osu.Game.Online.Notifications;
+using osu.Game.Online.Chat;
 using osu.Game.Online.Notifications.WebSocket;
 using osu.Game.Users;
 
@@ -55,6 +55,8 @@ public partial class APIAccess : Component, IAPIProvider
         public IBindable<UserActivity> Activity => activity;
         public IBindable<UserStatistics> Statistics => statistics;
 
+        public INotificationsClient NotificationsClient { get; }
+
         public Language Language => game.CurrentLanguage.Value;
 
         private Bindable<APIUser> localUser { get; } = new Bindable<APIUser>(createGuestUser());
@@ -82,6 +84,7 @@ public APIAccess(OsuGameBase game, OsuConfigManager config, EndpointConfiguratio
 
             APIEndpointUrl = endpointConfiguration.APIEndpointUrl;
             WebsiteRootUrl = endpointConfiguration.WebsiteRootUrl;
+            NotificationsClient = new WebSocketNotificationsClientConnector(this);
 
             authentication = new OAuth(endpointConfiguration.APIClientID, endpointConfiguration.APIClientSecret, APIEndpointUrl);
             log = Logger.GetLogger(LoggingTarget.Network);
@@ -324,8 +327,7 @@ public void Login(string username, string password)
         public IHubClientConnector GetHubConnector(string clientName, string endpoint, bool preferMessagePack) =>
             new HubClientConnector(clientName, endpoint, this, versionHash, preferMessagePack);
 
-        public NotificationsClientConnector GetNotificationsConnector() =>
-            new WebSocketNotificationsClientConnector(this);
+        public IChatClient GetChatClient() => new WebSocketChatClient(this);
 
         public RegistrationRequest.RegistrationRequestErrors CreateAccount(string email, string username, string password)
         {

osu.Game/Online/API/DummyAPIAccess.cs

@@ -8,7 +8,8 @@
 using osu.Framework.Graphics;
 using osu.Game.Localisation;
 using osu.Game.Online.API.Requests.Responses;
-using osu.Game.Online.Notifications;
+using osu.Game.Online.Chat;
+using osu.Game.Online.Notifications.WebSocket;
 using osu.Game.Tests;
 using osu.Game.Users;
 
@@ -30,6 +31,9 @@ public partial class DummyAPIAccess : Component, IAPIProvider
 
         public Bindable<UserStatistics?> Statistics { get; } = new Bindable<UserStatistics?>();
 
+        public DummyNotificationsClient NotificationsClient { get; } = new DummyNotificationsClient();
+        INotificationsClient IAPIProvider.NotificationsClient => NotificationsClient;
+
         public Language Language => Language.en;
 
         public string AccessToken => "token";
@@ -144,7 +148,7 @@ public void UpdateStatistics(UserStatistics newStatistics)
 
         public IHubClientConnector? GetHubConnector(string clientName, string endpoint, bool preferMessagePack) => null;
 
-        public NotificationsClientConnector GetNotificationsConnector() => new PollingNotificationsClientConnector(this);
+        public IChatClient GetChatClient() => new PollingChatClientConnector(this);
 
         public RegistrationRequest.RegistrationRequestErrors? CreateAccount(string email, string username, string password)
         {

osu.Game/Online/API/IAPIProvider.cs

@@ -6,7 +6,8 @@
 using osu.Framework.Bindables;
 using osu.Game.Localisation;
 using osu.Game.Online.API.Requests.Responses;
-using osu.Game.Online.Notifications;
+using osu.Game.Online.Chat;
+using osu.Game.Online.Notifications.WebSocket;
 using osu.Game.Users;
 
 namespace osu.Game.Online.API
@@ -130,9 +131,14 @@ public interface IAPIProvider
         IHubClientConnector? GetHubConnector(string clientName, string endpoint, bool preferMessagePack = true);
 
         /// <summary>
-        /// Constructs a new <see cref="NotificationsClientConnector"/>.
+        /// Accesses the <see cref="INotificationsClient"/> used to receive asynchronous notifications from web.
         /// </summary>
-        NotificationsClientConnector GetNotificationsConnector();
+        INotificationsClient NotificationsClient { get; }
+
+        /// <summary>
+        /// Creates a <see cref="IChatClient"/> instance to use in order to chat.
+        /// </summary>
+        IChatClient GetChatClient();
 
         /// <summary>
         /// Create a new user account. This is a blocking operation.

osu.Game/Online/Chat/ChannelManager.cs

@@ -16,7 +16,6 @@
 using osu.Game.Online.API;
 using osu.Game.Online.API.Requests;
 using osu.Game.Online.API.Requests.Responses;
-using osu.Game.Online.Notifications;
 using osu.Game.Overlays.Chat.Listing;
 
 namespace osu.Game.Online.Chat
@@ -64,13 +63,8 @@ public partial class ChannelManager : CompositeComponent, IChannelPostTarget
         /// </summary>
         public IBindableList<Channel> AvailableChannels => availableChannels;
 
-        /// <summary>
-        /// Whether the client responsible for channel notifications is connected.
-        /// </summary>
-        public bool NotificationsConnected => connector.IsConnected.Value;
-
         private readonly IAPIProvider api;
-        private readonly NotificationsClientConnector connector;
+        private readonly IChatClient chatClient;
 
         [Resolved]
         private UserLookupCache users { get; set; }
@@ -85,23 +79,19 @@ public ChannelManager(IAPIProvider api)
         {
             this.api = api;
 
-            connector = api.GetNotificationsConnector();
+            chatClient = api.GetChatClient();
 
             CurrentChannel.ValueChanged += currentChannelChanged;
         }
 
         [BackgroundDependencyLoader]
         private void load()
         {
-            connector.ChannelJoined += ch => Schedule(() => joinChannel(ch));
-
-            connector.ChannelParted += ch => Schedule(() => leaveChannel(getChannel(ch), false));
-
-            connector.NewMessages += msgs => Schedule(() => addMessages(msgs));
-
-            connector.PresenceReceived += () => Schedule(initializeChannels);
-
-            connector.Start();
+            chatClient.ChannelJoined += ch => Schedule(() => joinChannel(ch));
+            chatClient.ChannelParted += ch => Schedule(() => leaveChannel(getChannel(ch), false));
+            chatClient.NewMessages += msgs => Schedule(() => addMessages(msgs));
+            chatClient.PresenceReceived += () => Schedule(initializeChannels);
+            chatClient.RequestPresence();
 
             apiState.BindTo(api.State);
             apiState.BindValueChanged(_ => SendAck(), true);
@@ -655,7 +645,7 @@ public void MarkChannelAsRead(Channel channel)
         protected override void Dispose(bool isDisposing)
         {
             base.Dispose(isDisposing);
-            connector?.Dispose();
+            chatClient?.Dispose();
         }
     }
 

osu.Game/Online/Chat/IChatClient.cs

@@ -0,0 +1,39 @@
+// Copyright (c) ppy Pty Ltd <contact@ppy.sh>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.Collections.Generic;
+
+namespace osu.Game.Online.Chat
+{
+    /// <summary>
+    /// Interface for consuming online chat.
+    /// </summary>
+    public interface IChatClient : IDisposable
+    {
+        /// <summary>
+        /// Fired when a <see cref="Channel"/> has been joined.
+        /// </summary>
+        event Action<Channel>? ChannelJoined;
+
+        /// <summary>
+        /// Fired when a <see cref="Channel"/> has been parted.
+        /// </summary>
+        event Action<Channel>? ChannelParted;
+
+        /// <summary>
+        /// Fired when new <see cref="Message"/>s have arrived from the server.
+        /// </summary>
+        event Action<List<Message>>? NewMessages;
+
+        /// <summary>
+        /// Requests presence information from the server.
+        /// </summary>
+        void RequestPresence();
+
+        /// <summary>
+        /// Fired when the initial user presence information has been received.
+        /// </summary>
+        event Action? PresenceReceived;
+    }
+}

osu.Game/Online/Chat/WebSocketChatClient.cs

@@ -0,0 +1,144 @@
+// Copyright (c) ppy Pty Ltd <contact@ppy.sh>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using Newtonsoft.Json;
+using osu.Framework.Bindables;
+using osu.Framework.Extensions;
+using osu.Framework.Logging;
+using osu.Game.Online.API;
+using osu.Game.Online.API.Requests;
+using osu.Game.Online.Notifications.WebSocket;
+
+namespace osu.Game.Online.Chat
+{
+    public class WebSocketChatClient : IChatClient
+    {
+        public event Action<Channel>? ChannelJoined;
+        public event Action<Channel>? ChannelParted;
+        public event Action<List<Message>>? NewMessages;
+        public event Action? PresenceReceived;
+
+        private readonly IAPIProvider api;
+        private readonly INotificationsClient client;
+        private readonly ConcurrentDictionary<long, Channel> channelsMap = new ConcurrentDictionary<long, Channel>();
+
+        public WebSocketChatClient(IAPIProvider api)
+        {
+            this.api = api;
+            client = api.NotificationsClient;
+            client.IsConnected.BindValueChanged(start, true);
+        }
+
+        private void start(ValueChangedEvent<bool> connected)
+        {
+            if (!connected.NewValue)
+                return;
+
+            client.MessageReceived += onMessageReceived;
+            client.SendAsync(new StartChatRequest()).WaitSafely();
+            RequestPresence();
+        }
+
+        public void RequestPresence()
+        {
+            var fetchReq = new GetUpdatesRequest(0);
+
+            fetchReq.Success += updates =>
+            {
+                if (updates?.Presence != null)
+                {
+                    foreach (var channel in updates.Presence)
+                        joinChannel(channel);
+
+                    handleMessages(updates.Messages);
+                }
+
+                PresenceReceived?.Invoke();
+            };
+
+            api.Queue(fetchReq);
+        }
+
+        private void onMessageReceived(SocketMessage message)
+        {
+            switch (message.Event)
+            {
+                case @"chat.channel.join":
+                    Debug.Assert(message.Data != null);
+
+                    Channel? joinedChannel = JsonConvert.DeserializeObject<Channel>(message.Data.ToString());
+                    Debug.Assert(joinedChannel != null);
+
+                    joinChannel(joinedChannel);
+                    break;
+
+                case @"chat.channel.part":
+                    Debug.Assert(message.Data != null);
+
+                    Channel? partedChannel = JsonConvert.DeserializeObject<Channel>(message.Data.ToString());
+                    Debug.Assert(partedChannel != null);
+
+                    partChannel(partedChannel);
+                    break;
+
+                case @"chat.message.new":
+                    Debug.Assert(message.Data != null);
+
+                    NewChatMessageData? messageData = JsonConvert.DeserializeObject<NewChatMessageData>(message.Data.ToString());
+                    Debug.Assert(messageData != null);
+
+                    foreach (var msg in messageData.Messages)
+                        postToChannel(msg);
+
+                    break;
+            }
+        }
+
+        private void postToChannel(Message message)
+        {
+            if (channelsMap.TryGetValue(message.ChannelId, out Channel? channel))
+            {
+                joinChannel(channel);
+                NewMessages?.Invoke(new List<Message> { message });
+                return;
+            }
+
+            var req = new GetChannelRequest(message.ChannelId);
+
+            req.Success += response =>
+            {
+                joinChannel(channelsMap[message.ChannelId] = response.Channel);
+                NewMessages?.Invoke(new List<Message> { message });
+            };
+            req.Failure += ex => Logger.Error(ex, "Failed to join channel");
+
+            api.Queue(req);
+        }
+
+        private void joinChannel(Channel ch)
+        {
+            ch.Joined.Value = true;
+            ChannelJoined?.Invoke(ch);
+        }
+
+        private void partChannel(Channel channel) => ChannelParted?.Invoke(channel);
+
+        private void handleMessages(List<Message>? messages)
+        {
+            if (messages == null)
+                return;
+
+            NewMessages?.Invoke(messages);
+        }
+
+        public void Dispose()
+        {
+            client.IsConnected.ValueChanged -= start;
+            client.MessageReceived -= onMessageReceived;
+        }
+    }
+}