添加项目文件。

ConnectorService/ConnectorService.csproj

@@ -0,0 +1,20 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Swashbuckle.AspNetCore" Version="6.6.2" />
+  </ItemGroup>
+
+  <ItemGroup>
+	  <ProjectReference Include="..\IM.Commons\IM.Commons.csproj" />
+    <ProjectReference Include="..\IM.ASPNETCore\IM.ASPNETCore.csproj" />
+    <ProjectReference Include="..\IM.InitCommon\IM.InitCommon.csproj" />
+    <ProjectReference Include="..\IM.Protocols\IM.Protocols.csproj" />
+  </ItemGroup>
+
+</Project>

ConnectorService/ConnectorService.http

@@ -0,0 +1,6 @@
+@ConnectorService_HostAddress = http://localhost:5100
+
+GET {{ConnectorService_HostAddress}}/weatherforecast/
+Accept: application/json
+
+###

ConnectorService/Consumers/MessageConsumer.cs

@@ -0,0 +1,24 @@
+using ConnectorService.Dtos;
+using ConnectorService.Hubs;
+using IM.Commons.IntegrationEvents;
+using MassTransit;
+using Microsoft.AspNetCore.SignalR;
+
+namespace ConnectorService.Consumers
+{
+    public class MessageConsumer : IConsumer<MsgCreatedEvent>
+    {
+        private readonly IHubContext<ChatHub> hub;
+
+        public MessageConsumer(IHubContext<ChatHub> hub)
+        {
+            this.hub = hub;
+        }
+
+        public async Task Consume(ConsumeContext<MsgCreatedEvent> context)
+        {
+            var @event = context.Message;
+            await hub.Clients.Group(@event.StreamKey).SendAsync("ReceiveNewMessage", @event.ToHubResponse());
+        }
+    }
+}

ConnectorService/Dockerfile

@@ -0,0 +1,30 @@
+FROM mcr.microsoft.com/dotnet/sdk:8.0 AS build
+WORKDIR /src
+
+COPY IM_API_NEW.sln ./
+
+COPY ConnectorService/ConnectorService.csproj ConnectorService/
+COPY IM.Commons/IM.Commons.csproj IM.Commons/
+COPY IM.InitCommon/IM.InitCommon.csproj IM.InitCommon/
+COPY IM.Protocols/IM.Protocols.csproj IM.Protocols/
+
+RUN dotnet restore ConnectorService/ConnectorService.csproj
+
+COPY . .
+
+RUN dotnet publish ConnectorService/ConnectorService.csproj \
+    -c Release \
+    -o /app/publish \
+    --no-restore
+
+FROM mcr.microsoft.com/dotnet/aspnet:8.0 AS runtime
+WORKDIR /app
+
+ENV ASPNETCORE_ENVIRONMENT=Production
+ENV ASPNETCORE_URLS=http://+:8080
+
+EXPOSE 8080
+
+COPY --from=build /app/publish .
+
+ENTRYPOINT ["dotnet", "ConnectorService.dll"]

ConnectorService/Dtos/MessageHubResponse.cs

@@ -0,0 +1,81 @@
+using IM.Commons.IntegrationEvents;
+
+namespace ConnectorService.Dtos
+{
+    public record MessageHubResponse
+    {
+        public Guid Id { get; init; }
+
+        /// <summary>
+        /// 客户端去重/回执使用的本地 ID
+        /// </summary>
+        public Guid ClientId { get; init; }
+
+        public string ChatType { get; init; } = string.Empty;
+        public string MsgType { get; init; } = string.Empty;
+        public Guid SenderId { get; init; }
+        public Guid TargetId { get; init; }
+        public string State { get; init; } = string.Empty;
+        public string StreamKey { get; init; } = string.Empty;
+        public long SequenceId { get; init; }
+
+        /// <summary>
+        /// 服务器推送到达时间 (毫秒级时间戳，强烈建议加上)
+        /// </summary>
+        public long PushTimestamp { get; init; } = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+
+        public HubMsgContentDto Content { get; init; } = null!;
+    }
+
+    // 嵌套的内容对象，允许 Ext 和 Quote 为 null 以缩减 JSON 体积
+    public record HubMsgContentDto(
+        string Fallback,
+        object Body,
+        Dictionary<string, string>? Ext,
+        HubQuoteInfoDto? Quote
+    );
+
+    public record HubQuoteInfoDto(
+        Guid MessageId,
+        Guid SenderId,
+        string SenderName,
+        string MessageType,
+        string Preview
+    );
+    public static class MessageEventMapper
+    {
+        /// <summary>
+        /// 将内部集成事件转换为对外推送的 DTO
+        /// </summary>
+        public static MessageHubResponse ToHubResponse(this MsgCreatedEvent @event)
+        {
+            if (@event == null) throw new ArgumentNullException(nameof(@event));
+
+            return new MessageHubResponse
+            {
+                Id = @event.Id,
+                ClientId = @event.ClientId,
+                ChatType = @event.ChatType,
+                MsgType = @event.MsgType,
+                SenderId = @event.SenderId,
+                TargetId = @event.TargetId,
+                State = @event.State,
+                StreamKey = @event.StreamKey,
+                SequenceId = @event.SequenceId,
+                // 嵌套映射
+                Content = @event.Content != null ? new HubMsgContentDto(
+                    @event.Content.Fallback,
+                    @event.Content.Body,
+                    @event.Content.Ext,
+                    @event.Content.Quote != null ? new HubQuoteInfoDto(
+                        @event.Content.Quote.MessageId,
+                        @event.Content.Quote.SenderId,
+                        @event.Content.Quote.SenderName,
+                        @event.Content.Quote.MessageType,
+                        @event.Content.Quote.Preview
+                    ) : null
+                ) : null!
+            };
+        }
+    }
+}

ConnectorService/Hubs/ChatHub.cs

@@ -0,0 +1,53 @@
+using ConnectorService.Services;
+using IM.Commons;
+using Microsoft.AspNetCore.SignalR;
+using StackExchange.Redis;
+using System.Security.Claims;
+
+namespace ConnectorService.Hubs
+{
+    public class ChatHub : Hub
+    {
+        private readonly IConversationIntergrationService conService;
+        private readonly StackExchange.Redis.IDatabase redis;
+
+        public ChatHub(IConversationIntergrationService conService, IConnectionMultiplexer multiplexer)
+        {
+            this.conService = conService;
+            this.redis = multiplexer.GetDatabase();
+        }
+
+        public async override Task OnConnectedAsync()
+        {
+            if (!Context.User.Identity.IsAuthenticated)
+            {
+                Context.Abort();
+                return;
+            }
+
+            var userId = Context.User.FindFirstValue(ClaimTypes.NameIdentifier);
+
+            var res = await conService.GetUserStreamKeysAsync(Guid.Parse(userId));
+            foreach (var streamkey in res)
+            {
+                await Groups.AddToGroupAsync(Context.ConnectionId, streamkey);
+            }
+
+            await redis.SetAddAsync(RedisHelper.GetConnectionIdKey(userId), Context.ConnectionId);
+
+
+            await base.OnConnectedAsync();
+        }
+
+        public async override Task OnDisconnectedAsync(Exception? exception)
+        {
+            if (Context.User.Identity.IsAuthenticated)
+            {
+                var userId = Context.User.FindFirstValue(ClaimTypes.NameIdentifier);
+
+                await redis.SetRemoveAsync(RedisHelper.GetConnectionIdKey(userId), Context.ConnectionId);
+            }
+            await base.OnDisconnectedAsync(exception);
+        }
+    }
+}

ConnectorService/ModuleInit.cs

@@ -0,0 +1,19 @@
+using IM.Commons;
+using IM.Protocols.Grpc.Conversation;
+using Microsoft.Extensions.Options;
+
+namespace ConnectorService
+{
+    public class ModuleInit : IModuleInitializer
+    {
+        public void Initialize(IServiceCollection services)
+        {
+            services.AddRedisCache();
+            services.AddGrpcClient<ConversationInternal.ConversationInternalClient>((sp ,o) =>
+            {
+                var options = sp.GetRequiredService<IOptionsMonitor<GrpcOptions>>();
+                o.Address = new Uri(options.CurrentValue.MessageServiceUrl);
+            });
+        }
+    }
+}

ConnectorService/Program.cs

@@ -0,0 +1,42 @@
+
+using ConnectorService.Hubs;
+using IM.InitCommon;
+
+namespace ConnectorService
+{
+    public class Program
+    {
+        public static void Main(string[] args)
+        {
+            var builder = WebApplication.CreateBuilder(args);
+
+            // Add services to the container.
+
+            builder.ConfigureDbConfiguration();
+
+            builder.Services.AddSignalR();
+
+            // Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
+            builder.Services.AddEndpointsApiExplorer();
+            builder.Services.AddSwaggerGen();
+
+            builder.ConfigExtraServices();
+
+            var app = builder.Build();
+
+            // Configure the HTTP request pipeline.
+            if (app.Environment.IsDevelopment())
+            {
+                app.UseSwagger();
+                app.UseSwaggerUI();
+            }
+
+            app.UseAppDefault();
+
+
+            app.MapHub<ChatHub>("/chat");
+
+            app.Run();
+        }
+    }
+}

ConnectorService/Properties/launchSettings.json

@@ -0,0 +1,41 @@
+{
+  "$schema": "http://json.schemastore.org/launchsettings.json",
+  "iisSettings": {
+    "windowsAuthentication": false,
+    "anonymousAuthentication": true,
+    "iisExpress": {
+      "applicationUrl": "http://localhost:46392",
+      "sslPort": 44313
+    }
+  },
+  "profiles": {
+    "http": {
+      "commandName": "Project",
+      "dotnetRunMessages": true,
+      "launchBrowser": true,
+      "launchUrl": "swagger",
+      "applicationUrl": "http://localhost:5100",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "https": {
+      "commandName": "Project",
+      "dotnetRunMessages": true,
+      "launchBrowser": true,
+      "launchUrl": "swagger",
+      "applicationUrl": "https://localhost:7115;http://localhost:5100",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "IIS Express": {
+      "commandName": "IISExpress",
+      "launchBrowser": true,
+      "launchUrl": "swagger",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    }
+  }
+}

ConnectorService/Services/ConversationIntegrationService.cs

@@ -0,0 +1,30 @@
+using IM.Commons;
+using IM.Protocols.Grpc.Conversation;
+
+namespace ConnectorService.Services
+{
+    public class ConversationIntegrationService : IConversationIntergrationService
+    {
+        private readonly ConversationInternal.ConversationInternalClient client;
+        public async Task<List<string>> GetUserStreamKeysAsync(Guid userId)
+        {
+            var req = new GetUserStreamKeysRequest()
+            {
+                UserId = userId.ToString()
+            };
+            var res = await client.GetUserStreamKeysAsync(req);
+            if(res == null)
+            {
+                return [];
+            }
+
+            var list = new List<string>();
+            foreach(var item in res.StreamKeys)
+            {
+                list.Add(item);
+            }
+
+            return list;
+        }
+    }
+}

ConnectorService/Services/IConversationIntergrationService.cs

@@ -0,0 +1,9 @@
+using IM.Commons;
+
+namespace ConnectorService.Services
+{
+    public interface IConversationIntergrationService
+    {
+        Task<List<string>> GetUserStreamKeysAsync(Guid userId);
+    }
+}

ConnectorService/appsettings.Development.json

@@ -0,0 +1,8 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  }
+}

ConnectorService/appsettings.json

@@ -0,0 +1,9 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  },
+  "AllowedHosts": "*"
+}