将用户加入新对话并开始与 signalR 交换消息

Question

我在 Angular 上有以下 signalR 服务来与我的 signalR 服务器通信：

export class SignalService {
    private hubConnection: signalR.HubConnection;
    private messageSubject = new Subject<ConversationSignalMessage>();
    public messageReceived$ = this.messageSubject.asObservable();
    
    constructor(private authService: AuthService) {
        this.hubConnection = new signalR.HubConnectionBuilder()
        .withUrl(`https://localhost:5001/sampleHub?userId=${user.id}`, { withCredentials: false})
        .build();

        this.hubConnection.on('MessageReceived', (chatId: string, message: Message) => {
            if (message.userSender.id == user.id)
                return;

            this.messageSubject.next(new ConversationSignalMessage(chatId, message));
        });
    }

    startConnection(): Promise<any> {
        return this.hubConnection.start();
    }

    joinChat(chatId: string, participant: Participant): void {
        this.hubConnection.invoke('Join', chatId, participant)
        .catch(err => console.error(`Error joining chat: ${err}`));
    }

    sendMessage(chatId: string, message: Message): void {
        this.hubConnection.invoke('SendMessage', chatId, message)
        .catch(err => console.error(`Error sending message: ${err}`));
    }

    createConversation(users: string[], chatId: string, message: Message): void {
        this.hubConnection.invoke('CreateConversation', users, chatId, message)
        .catch(err => console.error(`Error creating conversation and joining users: ${err}`));
    }

    getMessageStream(): Observable<ConversationSignalMessage> {
        return this.messageSubject.asObservable();
    }

    stopConnection(): Promise<any> {
        return this.hubConnection.stop();
    }
}

这是我的 signalR c# 集线器：

private static readonly Dictionary<string, string> _connectedUsers = new();
private readonly Dictionary<Guid, List<string>> _conversationParticipants = new();
private readonly Dictionary<Guid, List<Message>> _conversationMessages = new();

public async Task SendMessage(Guid chatId, Message message)
{
    Console.WriteLine($"{conversationId}: {message.Text}");
    await Clients.Groups(chatId.ToString()).SendAsync("MessageReceived", chatId, message).ConfigureAwait(false);
}

public async Task Join(Guid chatId, Participant participant)
{
    Console.WriteLine($"{participant.UserName} joined {chatId}.");
    await Groups.AddToGroupAsync(Context.ConnectionId, chatId.ToString()).ConfigureAwait(false);
}

public async Task LeaveConversation(Guid chatId) => await Groups.RemoveFromGroupAsync(Context.ConnectionId, chatId.ToString()).ConfigureAwait(false);

public override async Task OnConnectedAsync()
{
    string userId = Context.GetHttpContext().Request.Query["userId"];
    _connectedUsers[Context.ConnectionId] = userId;
    await base.OnConnectedAsync().ConfigureAwait(false);
    Console.WriteLine($"Client `{userId}` connected.");
}

public override async Task OnDisconnectedAsync(Exception exception)
{
    if (_connectedUsers.TryGetValue(Context.ConnectionId, out string userId))
    {
        _ = _connectedUsers.Remove(Context.ConnectionId);
        Console.WriteLine($"Client `{userId}` disconnected.");
    }

    await base.OnDisconnectedAsync(exception).ConfigureAwait(false);
}

public async Task CreateConversation(List<string> users, Guid chatId, Message message)
{
    string creatorConnectionId = Context.ConnectionId;

    _conversationParticipants[conversationId] = userIds;

    if (!_conversationMessages.ContainsKey(conversationId))
    {
        _conversationMessages[conversationId] = new List<Message>();
    }

    _conversationMessages[conversationId].Add(message);

    foreach (var userId in users)
    {
        if (_connectedUsers.ContainsValue(userId.ToString()))
        {
            await Groups.AddToGroupAsync(creatorConnectionId, chatId.ToString()).ConfigureAwait(false);
            Console.WriteLine($"{userId} joined {chatId}.");
        }
    }

    // Frontend not listening this call.
    await Clients.Groups(chatId.ToString()).SendAsync("ConversationCreated", chatId).ConfigureAwait(false);

   await Clients.Groups(chatId.ToString()).SendAsync("MessageReceived", chatId, message).ConfigureAwait(false);
}

此刻我可以在我和另一个用户之间开始新的对话。我可以发送消息，但其他用户在刷新整个页面之前不会收到消息（基本上是从数据库加载聊天）。刷新后，将调用 Join 方法，现在我可以发送消息，他将收到该消息（通过 SendMessage），这工作正常。

我想做的是，一旦我在我之间创建一个新的对话，另一个用户会自动创建一个新的对话频道（集线器......也许？）并创建连接（类似于“加入”的作用）并且最后，一旦两个用户都连接并收听新的对话频道，我想开始在他们之间共享消息（类似于“SendMessage”的作用）。

目前我从 CreateConversation 创建了代码，我可以看到 Console.WriteLine($"{userId} join {chatId}.");我的控制台上显示消息，但在我发送消息后，其他用户从未收到该消息。我做错了什么或者我可以做什么来完成我在这里尝试的事情。

Answer 1

更改以下代码

foreach (var userId in users)
{
    if (_connectedUsers.ContainsValue(userId.ToString()))
    {
        await Groups.AddToGroupAsync(creatorConnectionId, chatId.ToString()).ConfigureAwait(false);
        Console.WriteLine($"{userId} joined {chatId}.");
    }
}

至

foreach (var userId in users)
{
    if (_connectedUsers.TryGetValue(userId.ToString(), out string connectionId))
    {
        await Groups.AddToGroupAsync(connectionId, chatId.ToString()).ConfigureAwait(false);
        Console.WriteLine($"{userId} joined {chatId}.");
    }
}

这里有逻辑问题。每次执行该方法时，都会添加一个 ConnectionId 到组中。