PHXH
/
csharp-sdk-upm
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore: 抽象出各模块的 Controller

oneRain 2020-03-25 16:42:30 +08:00
parent cd453ce134
commit 890fdc76f4
15 changed files with 956 additions and 571 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Realtime/Conversation/LCIMChatRoom.cs
View File

@ -17,9 +17,9 @@ namespace LeanCloud.Realtime {
Cid = Id,
Limit = limit
};
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.Members);
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Members);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
GenericCommand response = await Client.Connection.SendRequest(request);
List<string> memberList = response.ConvMessage.M.ToList();
return memberList;
}

377
Realtime/Conversation/LCIMConversation.cs
View File

@ -5,7 +5,6 @@ using System.Linq;
using Newtonsoft.Json;
using Google.Protobuf;
using LeanCloud.Realtime.Protocol;
using LeanCloud.Storage.Internal.Codec;
using LeanCloud.Storage;
namespace LeanCloud.Realtime {
@ -71,12 +70,14 @@ namespace LeanCloud.Realtime {
get; private set;
}
protected readonly LCIMClient client;
protected LCIMClient Client {
get; private set;
}
private Dictionary<string, object> customProperties;
internal LCIMConversation(LCIMClient client) {
this.client = client;
Client = client;
customProperties = new Dictionary<string, object>();
}
@ -85,13 +86,7 @@ namespace LeanCloud.Realtime {
/// </summary>
/// <returns></returns>
public async Task<int> GetMembersCount() {
ConvCommand conv = new ConvCommand {
Cid = Id,
};
GenericCommand command = client.NewCommand(CommandType.Conv, OpType.Count);
command.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(command);
return response.ConvMessage.Count;
return await Client.ConversationController.GetMembersCount(Id);
}
/// <summary>
@ -99,45 +94,26 @@ namespace LeanCloud.Realtime {
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public async Task<LCIMConversation> Read() {
public async Task Read() {
if (LastMessage == null) {
return this;
return;
}
ReadCommand read = new ReadCommand();
ReadTuple tuple = new ReadTuple {
Cid = Id,
Mid = LastMessage.Id,
Timestamp = LastMessage.SentTimestamp
};
read.Convs.Add(tuple);
GenericCommand request = client.NewCommand(CommandType.Read, OpType.Open);
request.ReadMessage = read;
await client.connection.SendRequest(request);
return this;
await Client.ConversationController.Read(Id, LastMessage);
}
public async Task<LCIMConversation> Save() {
ConvCommand conv = new ConvCommand {
Cid = Id,
};
// 注意序列化是否与存储一致
string json = JsonConvert.SerializeObject(LCEncoder.Encode(customProperties));
conv.Attr = new JsonObjectMessage {
Data = json
};
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.Update);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
JsonObjectMessage attr = response.ConvMessage.AttrModified;
// 更新自定义属性
if (attr != null) {
Dictionary<string, object> data = JsonConvert.DeserializeObject<Dictionary<string, object>>(attr.Data);
Dictionary<string, object> objectData = LCDecoder.Decode(data) as Dictionary<string, object>;
foreach (KeyValuePair<string, object> kv in objectData) {
customProperties[kv.Key] = kv.Value;
/// <summary>
/// 修改对话属性
/// </summary>
/// <param name="attributes"></param>
/// <returns></returns>
public async Task UpdateInfo(Dictionary<string, object> attributes) {
if (attributes == null || attributes.Count == 0) {
throw new ArgumentNullException(nameof(attributes));
}
Dictionary<string, object> updatedAttr = await Client.ConversationController.UpdateInfo(Id, attributes);
if (updatedAttr != null) {
MergeInfo(updatedAttr);
}
return this;
}
/// <summary>
@ -145,30 +121,11 @@ namespace LeanCloud.Realtime {
/// </summary>
/// <param name="clientIds">用户 Id</param>
/// <returns></returns>
public async Task<LCIMPartiallySuccessResult> Add(IEnumerable<string> clientIds) {
public async Task<LCIMPartiallySuccessResult> AddMembers(IEnumerable<string> clientIds) {
if (clientIds == null || clientIds.Count() == 0) {
throw new ArgumentNullException(nameof(clientIds));
}
ConvCommand conv = new ConvCommand {
Cid = Id,
};
conv.M.AddRange(clientIds);
// 签名参数
if (client.SignatureFactory != null) {
LCIMSignature signature = client.SignatureFactory.CreateConversationSignature(Id,
client.ClientId,
clientIds,
LCIMSignatureAction.Invite);
conv.S = signature.Signature;
conv.T = signature.Timestamp;
conv.N = signature.Nonce;
}
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.Add);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
List<string> allowedIds = response.ConvMessage.AllowedPids.ToList();
List<ErrorCommand> errors = response.ConvMessage.FailedPids.ToList();
return NewPartiallySuccessResult(allowedIds, errors);
return await Client.ConversationController.AddMembers(Id, clientIds);
}
/// <summary>
@ -176,56 +133,35 @@ namespace LeanCloud.Realtime {
/// </summary>
/// <param name="removeIds">用户 Id</param>
/// <returns></returns>
public async Task<LCIMPartiallySuccessResult> Remove(IEnumerable<string> removeIds) {
public async Task<LCIMPartiallySuccessResult> RemoveMembers(IEnumerable<string> removeIds) {
if (removeIds == null || removeIds.Count() == 0) {
throw new ArgumentNullException(nameof(removeIds));
}
ConvCommand conv = new ConvCommand {
Cid = Id,
};
conv.M.AddRange(removeIds);
// 签名参数
if (client.SignatureFactory != null) {
LCIMSignature signature = client.SignatureFactory.CreateConversationSignature(Id,
client.ClientId,
removeIds,
LCIMSignatureAction.Kick);
conv.S = signature.Signature;
conv.T = signature.Timestamp;
conv.N = signature.Nonce;
}
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.Remove);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
List<string> allowedIds = response.ConvMessage.AllowedPids.ToList();
List<ErrorCommand> errors = response.ConvMessage.FailedPids.ToList();
return NewPartiallySuccessResult(allowedIds, errors);
return await Client.ConversationController.RemoveMembers(Id, removeIds);
}
/// <summary>
/// 加入对话
/// </summary>
/// <returns></returns>
public async Task<LCIMConversation> Join() {
LCIMPartiallySuccessResult result = await Add(new string[] { client.ClientId });
public async Task Join() {
LCIMPartiallySuccessResult result = await AddMembers(new string[] { Client.Id });
if (!result.IsSuccess) {
LCIMOperationFailure error = result.FailureList[0];
throw new LCException(error.Code, error.Reason);
}
return this;
}
/// <summary>
/// 离开对话
/// </summary>
/// <returns></returns>
public async Task<LCIMConversation> Quit() {
LCIMPartiallySuccessResult result = await Remove(new string[] { client.ClientId });
public async Task Quit() {
LCIMPartiallySuccessResult result = await RemoveMembers(new string[] { Client.Id });
if (!result.IsSuccess) {
LCIMOperationFailure error = result.FailureList[0];
throw new LCException(error.Code, error.Reason);
}
return this;
}
/// <summary>
@ -234,24 +170,10 @@ namespace LeanCloud.Realtime {
/// <param name="message"></param>
/// <returns></returns>
public async Task<LCIMMessage> Send(LCIMMessage message) {
DirectCommand direct = new DirectCommand {
FromPeerId = client.ClientId,
Cid = Id,
};
if (message is LCIMTypedMessage typedMessage) {
direct.Msg = JsonConvert.SerializeObject(typedMessage.Encode());
} else if (message is LCIMBinaryMessage binaryMessage) {
direct.BinaryMsg = ByteString.CopyFrom(binaryMessage.Data);
} else {
throw new ArgumentException("Message MUST BE LCIMTypedMessage or LCIMBinaryMessage.");
if (message == null) {
throw new ArgumentNullException(nameof(message));
}
GenericCommand command = client.NewDirectCommand();
command.DirectMessage = direct;
GenericCommand response = await client.connection.SendRequest(command);
// 消息发送应答
AckCommand ack = response.AckMessage;
message.Id = ack.Uid;
message.DeliveredTimestamp = ack.T;
await Client.MessageController.Send(Id, message);
return message;
}
@ -259,30 +181,18 @@ namespace LeanCloud.Realtime {
/// 静音
/// </summary>
/// <returns></returns>
public async Task<LCIMConversation> Mute() {
ConvCommand conv = new ConvCommand {
Cid = Id
};
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.Mute);
request.ConvMessage = conv;
await client.connection.SendRequest(request);
public async Task Mute() {
await Client.ConversationController.Mute(Id);
IsMute = true;
return this;
}
/// <summary>
/// 取消静音
/// </summary>
/// <returns></returns>
public async Task<LCIMConversation> Unmute() {
ConvCommand conv = new ConvCommand {
Cid = Id
};
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.Unmute);
request.ConvMessage = conv;
await client.connection.SendRequest(request);
public async Task Unmute() {
await Client.ConversationController.Unmute(Id);
IsMute = false;
return this;
}
/// <summary>
@ -294,14 +204,7 @@ namespace LeanCloud.Realtime {
if (clientIds == null || clientIds.Count() == 0) {
throw new ArgumentNullException(nameof(clientIds));
}
ConvCommand conv = new ConvCommand {
Cid = Id
};
conv.M.AddRange(clientIds);
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.AddShutup);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
return NewPartiallySuccessResult(response.ConvMessage.AllowedPids, response.ConvMessage.FailedPids);
return await Client.ConversationController.MuteMembers(Id, clientIds);
}
/// <summary>
@ -313,14 +216,7 @@ namespace LeanCloud.Realtime {
if (clientIds == null || clientIds.Count() == 0) {
throw new ArgumentNullException(nameof(clientIds));
}
ConvCommand conv = new ConvCommand {
Cid = Id
};
conv.M.AddRange(clientIds);
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.Remove);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
return NewPartiallySuccessResult(response.ConvMessage.AllowedPids, response.ConvMessage.FailedPids);
return await Client.ConversationController.UnmuteMembers(Id, clientIds);
}
/// <summary>
@ -332,46 +228,14 @@ namespace LeanCloud.Realtime {
if (clientIds == null || clientIds.Count() == 0) {
throw new ArgumentNullException(nameof(clientIds));
}
BlacklistCommand blacklist = new BlacklistCommand {
SrcCid = Id,
};
blacklist.ToPids.AddRange(clientIds);
if (client.SignatureFactory != null) {
LCIMSignature signature = client.SignatureFactory.CreateBlacklistSignature(Id,
client.ClientId,
clientIds,
LCIMSignatureAction.ConversationBlockClients);
blacklist.S = signature.Signature;
blacklist.T = signature.Timestamp;
blacklist.N = signature.Nonce;
}
GenericCommand request = client.NewCommand(CommandType.Blacklist, OpType.Block);
request.BlacklistMessage = blacklist;
GenericCommand response = await client.connection.SendRequest(request);
return NewPartiallySuccessResult(response.BlacklistMessage.AllowedPids, response.BlacklistMessage.FailedPids);
return await Client.ConversationController.BlockMembers(Id, clientIds);
}
public async Task<LCIMPartiallySuccessResult> UnblockMembers(IEnumerable<string> clientIds) {
if (clientIds == null || clientIds.Count() == 0) {
throw new ArgumentNullException(nameof(clientIds));
}
BlacklistCommand blacklist = new BlacklistCommand {
SrcCid = Id,
};
blacklist.ToPids.AddRange(clientIds);
if (client.SignatureFactory != null) {
LCIMSignature signature = client.SignatureFactory.CreateBlacklistSignature(Id,
client.ClientId,
clientIds,
LCIMSignatureAction.ConversationUnblockClients);
blacklist.S = signature.Signature;
blacklist.T = signature.Timestamp;
blacklist.N = signature.Nonce;
}
GenericCommand request = client.NewCommand(CommandType.Blacklist, OpType.Unblock);
request.BlacklistMessage = blacklist;
GenericCommand response = await client.connection.SendRequest(request);
return NewPartiallySuccessResult(response.BlacklistMessage.AllowedPids, response.BlacklistMessage.FailedPids);
return await Client.ConversationController.UnblockMembers(Id, clientIds);
}
/// <summary>
@ -379,21 +243,11 @@ namespace LeanCloud.Realtime {
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public async Task<LCIMRecalledMessage> Recall(LCIMMessage message) {
public async Task RecallMessage(LCIMMessage message) {
if (message == null) {
throw new ArgumentNullException(nameof(message));
}
PatchCommand patch = new PatchCommand();
PatchItem item = new PatchItem {
Cid = Id,
Mid = message.Id,
Recall = true
};
patch.Patches.Add(item);
GenericCommand request = client.NewCommand(CommandType.Patch, OpType.Modify);
request.PatchMessage = patch;
GenericCommand response = await client.connection.SendRequest(request);
return null;
await Client.MessageController.RecallMessage(Id, message);
}
/// <summary>
@ -402,61 +256,45 @@ namespace LeanCloud.Realtime {
/// <param name="oldMessage"></param>
/// <param name="newMessage"></param>
/// <returns></returns>
public async Task<LCIMMessage> Update(LCIMMessage oldMessage, LCIMMessage newMessage) {
public async Task UpdateMessage(LCIMMessage oldMessage, LCIMMessage newMessage) {
if (oldMessage == null) {
throw new ArgumentNullException(nameof(oldMessage));
}
if (newMessage == null) {
throw new ArgumentNullException(nameof(newMessage));
}
PatchCommand patch = new PatchCommand();
PatchItem item = new PatchItem {
Cid = Id,
Mid = oldMessage.Id,
Timestamp = oldMessage.DeliveredTimestamp,
Recall = false,
};
if (newMessage is LCIMTypedMessage typedMessage) {
item.Data = JsonConvert.SerializeObject(typedMessage.Encode());
} else if (newMessage is LCIMBinaryMessage binaryMessage) {
item.BinaryMsg = ByteString.CopyFrom(binaryMessage.Data);
}
if (newMessage.MentionList != null) {
item.MentionPids.AddRange(newMessage.MentionList);
}
if (newMessage.MentionAll) {
item.MentionAll = newMessage.MentionAll;
}
patch.Patches.Add(item);
GenericCommand request = client.NewCommand(CommandType.Patch, OpType.Modify);
request.PatchMessage = patch;
GenericCommand response = await client.connection.SendRequest(request);
return null;
await Client.MessageController.UpdateMessage(Id, oldMessage, newMessage);
}
public async Task<LCIMConversation> UpdateMemberRole(string memberId, string role) {
/// <summary>
/// 更新对话中成员的角色
/// </summary>
/// <param name="memberId"></param>
/// <param name="role"></param>
/// <returns></returns>
public async Task UpdateMemberRole(string memberId, string role) {
if (string.IsNullOrEmpty(memberId)) {
throw new ArgumentNullException(nameof(memberId));
}
if (role != LCIMConversationMemberInfo.Manager && role != LCIMConversationMemberInfo.Member) {
throw new ArgumentException("role MUST be Manager Or Memebr");
}
ConvCommand conv = new ConvCommand {
Cid = Id,
TargetClientId = memberId,
Info = new ConvMemberInfo {
Pid = memberId,
Role = role
}
};
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.MemberInfoUpdate);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
// TODO 同步 members
return this;
await Client.ConversationController.UpdateMemberRole(Id, memberId, role);
}
/// <summary>
/// 获取对话中成员的角色（只返回管理员）
/// </summary>
/// <returns></returns>
public async Task<List<LCIMConversationMemberInfo>> GetAllMemberInfo() {
return await Client.ConversationController.GetAllMemberInfo(Id);
}
/// <summary>
/// 获取对话中指定成员的角色
/// </summary>
/// <param name="memberId"></param>
/// <returns></returns>
public async Task<LCIMConversationMemberInfo> GetMemberInfo(string memberId) {
if (string.IsNullOrEmpty(memberId)) {
throw new ArgumentNullException(nameof(memberId));
@ -470,88 +308,20 @@ namespace LeanCloud.Realtime {
return null;
}
public async Task<List<LCIMConversationMemberInfo>> GetAllMemberInfo() {
string path = "classes/_ConversationMemberInfo";
Dictionary<string, object> headers = new Dictionary<string, object> {
{ "X-LC-IM-Session-Token", client.SessionToken }
};
Dictionary<string, object> queryParams = new Dictionary<string, object> {
{ "client_id", client.ClientId },
{ "cid", Id }
};
Dictionary<string, object> response = await LCApplication.HttpClient.Get<Dictionary<string, object>>(path,
headers: headers, queryParams: queryParams);
List<object> results = response["results"] as List<object>;
List<LCIMConversationMemberInfo> memberList = new List<LCIMConversationMemberInfo>();
foreach (Dictionary<string, object> item in results) {
LCIMConversationMemberInfo member = new LCIMConversationMemberInfo {
ConversationId = item["cid"] as string,
MemberId = item["clientId"] as string,
Role = item["role"] as string
};
memberList.Add(member);
}
return memberList;
public async Task<LCIMPageResult> QueryMutedMembers(int limit = 10, string next = null) {
return await Client.ConversationController.QueryMutedMembers(Id, limit, next);
}
public async Task<LCIMPageResult> QueryMutedMembers(int limit = 50, string next = null) {
ConvCommand conv = new ConvCommand {
Cid = Id,
Limit = limit,
Next = next
};
GenericCommand request = client.NewCommand(CommandType.Conv, OpType.QueryShutup);
request.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(request);
return new LCIMPageResult {
Results = response.ConvMessage.M.ToList(),
Next = response.ConvMessage.Next
};
public async Task<LCIMPageResult> QueryBlockedMembers(int limit = 10, string next = null) {
return await Client.ConversationController.QueryBlockedMembers(Id, limit, next);
}
public async Task<List<LCIMMessage>> QueryMessage(LCIMMessageQueryEndpoint start = null,
public async Task<List<LCIMMessage>> QueryMessages(LCIMMessageQueryEndpoint start = null,
LCIMMessageQueryEndpoint end = null,
LCIMMessageQueryDirection direction = LCIMMessageQueryDirection.NewToOld,
int limit = 20,
int messageType = 0) {
LogsCommand logs = new LogsCommand {
Cid = Id
};
if (start != null) {
logs.T = start.SentTimestamp;
logs.Mid = start.MessageId;
logs.TIncluded = start.IsClosed;
}
if (end != null) {
logs.Tt = end.SentTimestamp;
logs.Tmid = end.MessageId;
logs.TtIncluded = end.IsClosed;
}
logs.Direction = direction == LCIMMessageQueryDirection.NewToOld ?
LogsCommand.Types.QueryDirection.Old : LogsCommand.Types.QueryDirection.New;
logs.Limit = limit;
if (messageType != 0) {
logs.Lctype = messageType;
}
GenericCommand request = client.NewCommand(CommandType.Logs, OpType.Open);
request.LogsMessage = logs;
GenericCommand response = await client.connection.SendRequest(request);
// TODO 反序列化聊天记录
return null;
}
private LCIMPartiallySuccessResult NewPartiallySuccessResult(IEnumerable<string> succesfulIds, IEnumerable<ErrorCommand> errors) {
LCIMPartiallySuccessResult result = new LCIMPartiallySuccessResult {
SuccessfulClientIdList = succesfulIds.ToList()
};
if (errors != null) {
result.FailureList = new List<LCIMOperationFailure>();
foreach (ErrorCommand error in errors) {
result.FailureList.Add(new LCIMOperationFailure(error));
}
}
return result;
return await Client.MessageController.QueryMessages(Id, start, end, direction, limit, messageType);
}
internal void MergeFrom(ConvCommand conv) {
@ -595,5 +365,14 @@ namespace LeanCloud.Realtime {
MutedMemberIdList = muo as List<string>;
}
}
internal void MergeInfo(Dictionary<string, object> attr) {
if (attr == null || attr.Count == 0) {
return;
}
foreach (KeyValuePair<string, object> kv in attr) {
customProperties[kv.Key] = kv.Value;
}
}
}
}

12
Realtime/Conversation/LCIMConversationQuery.cs
View File

@ -234,12 +234,16 @@ namespace LeanCloud.Realtime {
get; set;
}
/// <summary>
/// 查找
/// </summary>
/// <returns></returns>
public async Task<List<LCIMConversation>> Find() {
GenericCommand command = new GenericCommand {
Cmd = CommandType.Conv,
Op = OpType.Query,
AppId = LCApplication.AppId,
PeerId = client.ClientId,
PeerId = client.Id,
};
ConvCommand conv = new ConvCommand();
string where = condition.BuildWhere();
@ -249,16 +253,16 @@ namespace LeanCloud.Realtime {
};
}
command.ConvMessage = conv;
GenericCommand response = await client.connection.SendRequest(command);
GenericCommand response = await client.Connection.SendRequest(command);
JsonObjectMessage results = response.ConvMessage.Results;
List<object> convs = JsonConvert.DeserializeObject<List<object>>(results.Data, new LCJsonConverter());
List<LCIMConversation> convList = new List<LCIMConversation>(convs.Count);
foreach (object c in convs) {
Dictionary<string, object> cd = c as Dictionary<string, object>;
string convId = cd["objectId"] as string;
if (!client.conversationDict.TryGetValue(convId, out LCIMConversation conversation)) {
if (!client.ConversationDict.TryGetValue(convId, out LCIMConversation conversation)) {
conversation = new LCIMConversation(client);
client.conversationDict[convId] = conversation;
client.ConversationDict[convId] = conversation;
}
conversation.MergeFrom(cd);
convList.Add(conversation);

23
Realtime/Internal/Controller/LCIMController.cs Normal file
View File

@ -0,0 +1,23 @@
using System.Threading.Tasks;
using LeanCloud.Realtime.Protocol;
using LeanCloud.Realtime.Internal.WebSocket;
namespace LeanCloud.Realtime.Internal.Controller {
internal abstract class LCIMController {
protected LCIMClient Client {
get; set;
}
internal LCIMController(LCIMClient client) {
Client = client;
}
internal abstract Task OnNotification(GenericCommand notification);
protected LCWebSocketConnection Connection {
get {
return Client.Connection;
}
}
}
}

402
Realtime/Internal/Controller/LCIMConversationController.cs Normal file
View File

@ -0,0 +1,402 @@
using System;
using System.Linq;
using System.Collections.Generic;
using System.Threading.Tasks;
using Newtonsoft.Json;
using LeanCloud.Realtime.Protocol;
using LeanCloud.Storage.Internal;
using LeanCloud.Storage.Internal.Codec;
using Google.Protobuf;
namespace LeanCloud.Realtime.Internal.Controller {
internal class LCIMConversationController : LCIMController {
internal LCIMConversationController(LCIMClient client) : base(client) {
}
/// <summary>
/// 创建对话
/// </summary>
/// <param name="members"></param>
/// <param name="name"></param>
/// <param name="transient"></param>
/// <param name="unique"></param>
/// <param name="temporary"></param>
/// <param name="temporaryTtl"></param>
/// <param name="properties"></param>
/// <returns></returns>
internal async Task<LCIMConversation> CreateConv(
IEnumerable<string> members = null,
string name = null,
bool transient = false,
bool unique = true,
bool temporary = false,
int temporaryTtl = 86400,
Dictionary<string, object> properties = null) {
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Start);
ConvCommand conv = new ConvCommand {
Transient = transient,
Unique = unique,
};
if (members != null) {
conv.M.AddRange(members);
}
if (!string.IsNullOrEmpty(name)) {
conv.N = name;
}
if (temporary) {
conv.TempConv = temporary;
conv.TempConvTTL = temporaryTtl;
}
if (properties != null) {
conv.Attr = new JsonObjectMessage {
Data = JsonConvert.SerializeObject(LCEncoder.Encode(properties))
};
}
if (Client.SignatureFactory != null) {
LCIMSignature signature = Client.SignatureFactory.CreateStartConversationSignature(Client.Id, members);
conv.S = signature.Signature;
conv.T = signature.Timestamp;
conv.N = signature.Nonce;
}
request.ConvMessage = conv;
GenericCommand response = await Connection.SendRequest(request);
string convId = response.ConvMessage.Cid;
if (!Client.ConversationDict.TryGetValue(convId, out LCIMConversation conversation)) {
if (transient) {
conversation = new LCIMChatRoom(Client);
} else if (temporary) {
conversation = new LCIMTemporaryConversation(Client);
} else if (properties != null && properties.ContainsKey("system")) {
conversation = new LCIMServiceConversation(Client);
} else {
conversation = new LCIMConversation(Client);
}
Client.ConversationDict[convId] = conversation;
}
// 合并请求数据
conversation.Name = name;
conversation.MemberIdList = members?.ToList();
// 合并服务端推送的数据
conversation.MergeFrom(response.ConvMessage);
return conversation;
}
internal async Task<int> GetMembersCount(string convId) {
ConvCommand conv = new ConvCommand {
Cid = convId,
};
GenericCommand command = Client.NewCommand(CommandType.Conv, OpType.Count);
command.ConvMessage = conv;
GenericCommand response = await Connection.SendRequest(command);
return response.ConvMessage.Count;
}
internal async Task Read(string convId, LCIMMessage message) {
ReadCommand read = new ReadCommand();
ReadTuple tuple = new ReadTuple {
Cid = convId,
Mid = message.Id,
Timestamp = message.SentTimestamp
};
read.Convs.Add(tuple);
GenericCommand request = Client.NewCommand(CommandType.Read, OpType.Open);
request.ReadMessage = read;
await Client.Connection.SendRequest(request);
}
internal async Task<Dictionary<string, object>> UpdateInfo(string convId, Dictionary<string, object> attributes) {
ConvCommand conv = new ConvCommand {
Cid = convId,
};
conv.Attr = new JsonObjectMessage {
Data = JsonConvert.SerializeObject(attributes)
};
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Update);
request.ConvMessage = conv;
GenericCommand response = await Client.Connection.SendRequest(request);
JsonObjectMessage attr = response.ConvMessage.AttrModified;
// 更新自定义属性
if (attr != null) {
Dictionary<string, object> updatedAttr = JsonConvert.DeserializeObject<Dictionary<string, object>>(attr.Data);
return updatedAttr;
}
return null;
}
internal async Task<LCIMPartiallySuccessResult> AddMembers(string convId, IEnumerable<string> clientIds) {
ConvCommand conv = new ConvCommand {
Cid = convId,
};
conv.M.AddRange(clientIds);
// 签名参数
if (Client.SignatureFactory != null) {
LCIMSignature signature = Client.SignatureFactory.CreateConversationSignature(convId,
Client.Id,
clientIds,
LCIMSignatureAction.Invite);
conv.S = signature.Signature;
conv.T = signature.Timestamp;
conv.N = signature.Nonce;
}
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Add);
request.ConvMessage = conv;
GenericCommand response = await Client.Connection.SendRequest(request);
List<string> allowedIds = response.ConvMessage.AllowedPids.ToList();
List<ErrorCommand> errors = response.ConvMessage.FailedPids.ToList();
return NewPartiallySuccessResult(allowedIds, errors);
}
internal async Task<LCIMPartiallySuccessResult> RemoveMembers(string convId, IEnumerable<string> removeIds) {
ConvCommand conv = new ConvCommand {
Cid = convId,
};
conv.M.AddRange(removeIds);
// 签名参数
if (Client.SignatureFactory != null) {
LCIMSignature signature = Client.SignatureFactory.CreateConversationSignature(convId,
Client.Id,
removeIds,
LCIMSignatureAction.Kick);
conv.S = signature.Signature;
conv.T = signature.Timestamp;
conv.N = signature.Nonce;
}
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Remove);
request.ConvMessage = conv;
GenericCommand response = await Client.Connection.SendRequest(request);
List<string> allowedIds = response.ConvMessage.AllowedPids.ToList();
List<ErrorCommand> errors = response.ConvMessage.FailedPids.ToList();
return NewPartiallySuccessResult(allowedIds, errors);
}
internal async Task Mute(string convId) {
ConvCommand conv = new ConvCommand {
Cid = convId
};
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Mute);
request.ConvMessage = conv;
await Client.Connection.SendRequest(request);
}
internal async Task Unmute(string convId) {
ConvCommand conv = new ConvCommand {
Cid = convId
};
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Unmute);
request.ConvMessage = conv;
await Client.Connection.SendRequest(request);
}
internal async Task<LCIMPartiallySuccessResult> MuteMembers(string convId, IEnumerable<string> clientIds) {
if (clientIds == null || clientIds.Count() == 0) {
throw new ArgumentNullException(nameof(clientIds));
}
ConvCommand conv = new ConvCommand {
Cid = convId
};
conv.M.AddRange(clientIds);
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.AddShutup);
request.ConvMessage = conv;
GenericCommand response = await Client.Connection.SendRequest(request);
return NewPartiallySuccessResult(response.ConvMessage.AllowedPids, response.ConvMessage.FailedPids);
}
internal async Task<LCIMPartiallySuccessResult> UnmuteMembers(string convId, IEnumerable<string> clientIds) {
ConvCommand conv = new ConvCommand {
Cid = convId
};
conv.M.AddRange(clientIds);
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.Remove);
request.ConvMessage = conv;
GenericCommand response = await Client.Connection.SendRequest(request);
return NewPartiallySuccessResult(response.ConvMessage.AllowedPids, response.ConvMessage.FailedPids);
}
internal async Task<LCIMPartiallySuccessResult> BlockMembers(string convId, IEnumerable<string> clientIds) {
BlacklistCommand blacklist = new BlacklistCommand {
SrcCid = convId,
};
blacklist.ToPids.AddRange(clientIds);
if (Client.SignatureFactory != null) {
LCIMSignature signature = Client.SignatureFactory.CreateBlacklistSignature(convId,
Client.Id,
clientIds,
LCIMSignatureAction.ConversationBlockClients);
blacklist.S = signature.Signature;
blacklist.T = signature.Timestamp;
blacklist.N = signature.Nonce;
}
GenericCommand request = Client.NewCommand(CommandType.Blacklist, OpType.Block);
request.BlacklistMessage = blacklist;
GenericCommand response = await Client.Connection.SendRequest(request);
return NewPartiallySuccessResult(response.BlacklistMessage.AllowedPids, response.BlacklistMessage.FailedPids);
}
internal async Task<LCIMPartiallySuccessResult> UnblockMembers(string convId, IEnumerable<string> clientIds) {
BlacklistCommand blacklist = new BlacklistCommand {
SrcCid = convId,
};
blacklist.ToPids.AddRange(clientIds);
if (Client.SignatureFactory != null) {
LCIMSignature signature = Client.SignatureFactory.CreateBlacklistSignature(convId,
Client.Id,
clientIds,
LCIMSignatureAction.ConversationUnblockClients);
blacklist.S = signature.Signature;
blacklist.T = signature.Timestamp;
blacklist.N = signature.Nonce;
}
GenericCommand request = Client.NewCommand(CommandType.Blacklist, OpType.Unblock);
request.BlacklistMessage = blacklist;
GenericCommand response = await Client.Connection.SendRequest(request);
return NewPartiallySuccessResult(response.BlacklistMessage.AllowedPids, response.BlacklistMessage.FailedPids);
}
internal async Task UpdateMemberRole(string convId, string memberId, string role) {
ConvCommand conv = new ConvCommand {
Cid = convId,
TargetClientId = memberId,
Info = new ConvMemberInfo {
Pid = memberId,
Role = role
}
};
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.MemberInfoUpdate);
request.ConvMessage = conv;
GenericCommand response = await Client.Connection.SendRequest(request);
}
internal async Task<List<LCIMConversationMemberInfo>> GetAllMemberInfo(string convId) {
string path = "classes/_ConversationMemberInfo";
string token = await Client.SessionController.GetToken();
Dictionary<string, object> headers = new Dictionary<string, object> {
{ "X-LC-IM-Session-Token", token }
};
Dictionary<string, object> queryParams = new Dictionary<string, object> {
{ "client_id", Client.Id },
{ "cid", convId }
};
Dictionary<string, object> response = await LCApplication.HttpClient.Get<Dictionary<string, object>>(path,
headers: headers, queryParams: queryParams);
List<object> results = response["results"] as List<object>;
List<LCIMConversationMemberInfo> memberList = new List<LCIMConversationMemberInfo>();
foreach (Dictionary<string, object> item in results) {
LCIMConversationMemberInfo member = new LCIMConversationMemberInfo {
ConversationId = item["cid"] as string,
MemberId = item["clientId"] as string,
Role = item["role"] as string
};
memberList.Add(member);
}
return memberList;
}
internal async Task<LCIMPageResult> QueryMutedMembers(string convId, int limit = 10, string next = null) {
ConvCommand conv = new ConvCommand {
Cid = convId,
Limit = limit,
Next = next
};
GenericCommand request = Client.NewCommand(CommandType.Conv, OpType.QueryShutup);
request.ConvMessage = conv;
GenericCommand response = await Client.Connection.SendRequest(request);
return new LCIMPageResult {
Results = response.ConvMessage.M.ToList(),
Next = response.ConvMessage.Next
};
}
internal async Task<LCIMPageResult> QueryBlockedMembers(string convId, int limit = 10, string next = null) {
BlacklistCommand black = new BlacklistCommand {
SrcCid = convId,
Limit = limit,
Next = next
};
GenericCommand request = Client.NewCommand(CommandType.Blacklist, OpType.Query);
request.BlacklistMessage = black;
GenericCommand response = await Client.Connection.SendRequest(request);
return new LCIMPageResult {
Results = response.BlacklistMessage.BlockedPids.ToList(),
Next = response.BlacklistMessage.Next
};
}
private LCIMPartiallySuccessResult NewPartiallySuccessResult(IEnumerable<string> succesfulIds, IEnumerable<ErrorCommand> errors) {
LCIMPartiallySuccessResult result = new LCIMPartiallySuccessResult {
SuccessfulClientIdList = succesfulIds.ToList()
};
if (errors != null) {
result.FailureList = new List<LCIMOperationFailure>();
foreach (ErrorCommand error in errors) {
result.FailureList.Add(new LCIMOperationFailure(error));
}
}
return result;
}
internal override async Task OnNotification(GenericCommand notification) {
ConvCommand conv = notification.ConvMessage;
switch (notification.Op) {
case OpType.Joined:
await OnConversationJoined(conv);
break;
case OpType.MembersJoined:
await OnConversationMembersJoined(conv);
break;
case OpType.Left:
await OnConversationLeft(conv);
break;
case OpType.MembersLeft:
await OnConversationMemberLeft(conv);
break;
case OpType.Updated:
await OnConversationPropertiesUpdated(conv);
break;
case OpType.MemberInfoChanged:
await OnConversationMemberInfoChanged(conv);
break;
default:
break;
}
}
private async Task OnConversationJoined(ConvCommand conv) {
LCIMConversation conversation = await Client.GetOrQueryConversation(conv.Cid);
conversation.MergeFrom(conv);
Client.OnInvited?.Invoke(conversation, conv.InitBy);
}
private async Task OnConversationMembersJoined(ConvCommand conv) {
LCIMConversation conversation = await Client.GetOrQueryConversation(conv.Cid);
conversation.MergeFrom(conv);
Client.OnMembersJoined?.Invoke(conversation, conv.M.ToList(), conv.InitBy);
}
private async Task OnConversationLeft(ConvCommand conv) {
LCIMConversation conversation = await Client.GetOrQueryConversation(conv.Cid);
Client.OnKicked?.Invoke(conversation, conv.InitBy);
}
private async Task OnConversationMemberLeft(ConvCommand conv) {
LCIMConversation conversation = await Client.GetOrQueryConversation(conv.Cid);
List<string> leftIdList = conv.M.ToList();
Client.OnMembersLeft?.Invoke(conversation, leftIdList, conv.InitBy);
}
private async Task OnConversationPropertiesUpdated(ConvCommand conv) {
LCIMConversation conversation = await Client.GetOrQueryConversation(conv.Cid);
Dictionary<string, object> updatedAttr = JsonConvert.DeserializeObject<Dictionary<string, object>>(conv.AttrModified.Data,
new LCJsonConverter());
// 更新内存数据
conversation.MergeInfo(updatedAttr);
Client.OnConversationInfoUpdated?.Invoke(conversation, updatedAttr, conv.InitBy);
}
private async Task OnConversationMemberInfoChanged(ConvCommand conv) {
LCIMConversation conversation = await Client.GetOrQueryConversation(conv.Cid);
ConvMemberInfo memberInfo = conv.Info;
Client.OnMemberInfoUpdated?.Invoke(conversation, memberInfo.Pid, memberInfo.Role, conv.InitBy);
}
}
}

16
Realtime/Internal/Controller/LCIMGoAwayController.cs Normal file
View File

@ -0,0 +1,16 @@
using System.Threading.Tasks;
using LeanCloud.Realtime.Protocol;
namespace LeanCloud.Realtime.Internal.Controller {
internal class LCIMGoAwayController : LCIMController {
internal LCIMGoAwayController(LCIMClient client) : base(client) {
}
internal override async Task OnNotification(GenericCommand notification) {
// TODO 清空缓存，断开连接，等待重新连接
Connection.Router.Reset();
await Connection.Close();
}
}
}

150
Realtime/Internal/Controller/LCIMMessageController.cs Normal file
View File

@ -0,0 +1,150 @@
using System;
using System.Threading.Tasks;
using System.Collections.Generic;
using Newtonsoft.Json;
using Google.Protobuf;
using LeanCloud.Storage.Internal;
using LeanCloud.Realtime.Protocol;
namespace LeanCloud.Realtime.Internal.Controller {
internal class LCIMMessageController : LCIMController {
internal LCIMMessageController(LCIMClient client) : base(client) {
}
internal async Task<LCIMMessage> Send(string convId, LCIMMessage message) {
DirectCommand direct = new DirectCommand {
FromPeerId = Client.Id,
Cid = convId,
};
if (message is LCIMTypedMessage typedMessage) {
direct.Msg = JsonConvert.SerializeObject(typedMessage.Encode());
} else if (message is LCIMBinaryMessage binaryMessage) {
direct.BinaryMsg = ByteString.CopyFrom(binaryMessage.Data);
} else {
throw new ArgumentException("Message MUST BE LCIMTypedMessage or LCIMBinaryMessage.");
}
GenericCommand command = Client.NewDirectCommand();
command.DirectMessage = direct;
GenericCommand response = await Client.Connection.SendRequest(command);
// 消息发送应答
AckCommand ack = response.AckMessage;
message.Id = ack.Uid;
message.DeliveredTimestamp = ack.T;
return message;
}
internal async Task RecallMessage(string convId, LCIMMessage message) {
PatchCommand patch = new PatchCommand();
PatchItem item = new PatchItem {
Cid = convId,
Mid = message.Id,
Recall = true
};
patch.Patches.Add(item);
GenericCommand request = Client.NewCommand(CommandType.Patch, OpType.Modify);
request.PatchMessage = patch;
await Client.Connection.SendRequest(request);
}
internal async Task UpdateMessage(string convId, LCIMMessage oldMessage, LCIMMessage newMessage) {
PatchCommand patch = new PatchCommand();
PatchItem item = new PatchItem {
Cid = convId,
Mid = oldMessage.Id,
Timestamp = oldMessage.DeliveredTimestamp,
Recall = false,
};
if (newMessage is LCIMTypedMessage typedMessage) {
item.Data = JsonConvert.SerializeObject(typedMessage.Encode());
} else if (newMessage is LCIMBinaryMessage binaryMessage) {
item.BinaryMsg = ByteString.CopyFrom(binaryMessage.Data);
}
if (newMessage.MentionList != null) {
item.MentionPids.AddRange(newMessage.MentionList);
}
if (newMessage.MentionAll) {
item.MentionAll = newMessage.MentionAll;
}
patch.Patches.Add(item);
GenericCommand request = Client.NewCommand(CommandType.Patch, OpType.Modify);
request.PatchMessage = patch;
GenericCommand response = await Client.Connection.SendRequest(request);
}
internal async Task<List<LCIMMessage>> QueryMessages(string convId,
LCIMMessageQueryEndpoint start = null,
LCIMMessageQueryEndpoint end = null,
LCIMMessageQueryDirection direction = LCIMMessageQueryDirection.NewToOld,
int limit = 20,
int messageType = 0) {
LogsCommand logs = new LogsCommand {
Cid = convId
};
if (start != null) {
logs.T = start.SentTimestamp;
logs.Mid = start.MessageId;
logs.TIncluded = start.IsClosed;
}
if (end != null) {
logs.Tt = end.SentTimestamp;
logs.Tmid = end.MessageId;
logs.TtIncluded = end.IsClosed;
}
logs.Direction = direction == LCIMMessageQueryDirection.NewToOld ?
LogsCommand.Types.QueryDirection.Old : LogsCommand.Types.QueryDirection.New;
logs.Limit = limit;
if (messageType != 0) {
logs.Lctype = messageType;
}
GenericCommand request = Client.NewCommand(CommandType.Logs, OpType.Open);
request.LogsMessage = logs;
GenericCommand response = await Client.Connection.SendRequest(request);
// TODO 反序列化聊天记录
return null;
}
internal override async Task OnNotification(GenericCommand notification) {
DirectCommand direct = notification.DirectMessage;
LCIMMessage message = null;
if (direct.HasBinaryMsg) {
// 二进制消息
byte[] bytes = direct.BinaryMsg.ToByteArray();
message = new LCIMBinaryMessage(bytes);
} else {
// 文本消息
string messageData = direct.Msg;
Dictionary<string, object> msg = JsonConvert.DeserializeObject<Dictionary<string, object>>(messageData,
new LCJsonConverter());
int msgType = (int)(long)msg["_lctype"];
switch (msgType) {
case -1:
message = new LCIMTextMessage();
break;
case -2:
message = new LCIMImageMessage();
break;
case -3:
message = new LCIMAudioMessage();
break;
case -4:
message = new LCIMVideoMessage();
break;
case -5:
message = new LCIMLocationMessage();
break;
case -6:
message = new LCIMFileMessage();
break;
default:
break;
}
message.Decode(direct);
}
// 获取对话
LCIMConversation conversation = await Client.GetOrQueryConversation(direct.Cid);
Client.OnMessage?.Invoke(conversation, message);
}
}
}

83
Realtime/Internal/Controller/LCIMSessionController.cs Normal file
View File

@ -0,0 +1,83 @@
using System;
using System.Threading.Tasks;
using LeanCloud.Realtime.Protocol;
namespace LeanCloud.Realtime.Internal.Controller {
internal class LCIMSessionController : LCIMController {
private string token;
private DateTimeOffset expiredAt;
internal LCIMSessionController(LCIMClient client) : base(client) {
}
internal async Task Open() {
SessionCommand session = NewSessionCommand();
GenericCommand request = Client.NewCommand(CommandType.Session, OpType.Open);
request.SessionMessage = session;
GenericCommand response = await Client.Connection.SendRequest(request);
UpdateSession(response.SessionMessage);
}
internal async Task Close() {
GenericCommand request = Client.NewCommand(CommandType.Session, OpType.Close);
await Client.Connection.SendRequest(request);
}
internal async Task<string> GetToken() {
if (IsExpired) {
await Refresh();
}
return token;
}
private async Task Refresh() {
SessionCommand session = NewSessionCommand();
GenericCommand request = Client.NewCommand(CommandType.Session, OpType.Refresh);
request.SessionMessage = session;
GenericCommand response = await Client.Connection.SendRequest(request);
UpdateSession(response.SessionMessage);
}
private SessionCommand NewSessionCommand() {
SessionCommand session = new SessionCommand();
if (Client.SignatureFactory != null) {
LCIMSignature signature = Client.SignatureFactory.CreateConnectSignature(Client.Id);
session.S = signature.Signature;
session.T = signature.Timestamp;
session.N = signature.Nonce;
}
return session;
}
private void UpdateSession(SessionCommand session) {
token = session.St;
int ttl = session.StTtl;
expiredAt = DateTimeOffset.Now + TimeSpan.FromSeconds(ttl);
}
internal override async Task OnNotification(GenericCommand notification) {
switch (notification.Op) {
case OpType.Closed:
await OnClosed(notification.SessionMessage);
break;
default:
break;
}
}
private bool IsExpired {
get {
return DateTimeOffset.Now > expiredAt;
}
}
private async Task OnClosed(SessionCommand session) {
int code = session.Code;
string reason = session.Reason;
string detail = session.Detail;
await Connection.Close();
Client.OnClose?.Invoke(code, reason, detail);
}
}
}

26
Realtime/Internal/Controller/LCIMUnreadController.cs Normal file
View File

@ -0,0 +1,26 @@
using System.Threading.Tasks;
using System.Collections.Generic;
using Newtonsoft.Json;
using LeanCloud.Realtime.Protocol;
namespace LeanCloud.Realtime.Internal.Controller {
internal class LCIMUnreadController : LCIMController {
internal LCIMUnreadController(LCIMClient client) : base(client) {
}
internal override async Task OnNotification(GenericCommand notification) {
UnreadCommand unread = notification.UnreadMessage;
List<LCIMConversation> conversationList = new List<LCIMConversation>();
foreach (UnreadTuple conv in unread.Convs) {
// 查询对话
LCIMConversation conversation = await Client.GetOrQueryConversation(conv.Cid);
conversation.Unread = conv.Unread;
// TODO 反序列化对话
// 最后一条消息
JsonConvert.DeserializeObject<Dictionary<string, object>>(conv.Data);
conversationList.Add(conversation);
}
Client.OnUnreadMessagesCountUpdated?.Invoke(conversationList);
}
}
}

4
Realtime/Internal/Router/LCRTMRouter.cs
View File

@ -19,6 +19,10 @@ namespace LeanCloud.Realtime.Internal.Router {
return rtmServer.Server;
}
internal void Reset() {
rtmServer = null;
}
async Task<LCRTMServer> Fetch() {
string server = await LCApplication.AppRouter.GetRealtimeServer();
string url = $"{server}/v1/route?appId={LCApplication.AppId}&secure=1";

26
Realtime/Internal/WebSocket/LCWebSocketConnection.cs
View File

@ -11,7 +11,8 @@ using Google.Protobuf;
namespace LeanCloud.Realtime.Internal.WebSocket {
internal class LCWebSocketConnection {
private const int KEEP_ALIVE_INTERVAL = 10;
private const int RECV_BUFFER_SIZE = 1024;
// .net standard 2.0 好像在拼合 Frame 时有 bug，所以将这个值调整大一些
private const int RECV_BUFFER_SIZE = 1024 * 5;
private ClientWebSocket ws;
@ -19,22 +20,31 @@ namespace LeanCloud.Realtime.Internal.WebSocket {
private readonly object requestILock = new object();
private Dictionary<int, TaskCompletionSource<GenericCommand>> responses;
private readonly Dictionary<int, TaskCompletionSource<GenericCommand>> responses;
private string id;
private readonly string id;
internal LCRTMRouter Router {
get; private set;
}
internal Func<GenericCommand, Task> OnNotification {
get; set;
}
internal Action<int, string> OnClose {
get; set;
}
internal LCWebSocketConnection(string id) {
Router = new LCRTMRouter();
this.id = id;
responses = new Dictionary<int, TaskCompletionSource<GenericCommand>>();
}
internal async Task Connect() {
LCRTMRouter rtmRouter = new LCRTMRouter();
string rtmServer = await rtmRouter.GetServer();
string rtmServer = await Router.GetServer();
ws = new ClientWebSocket();
ws.Options.AddSubProtocol("lc.protobuf2.3");
@ -71,8 +81,7 @@ namespace LeanCloud.Realtime.Internal.WebSocket {
do {
result = await ws.ReceiveAsync(new ArraySegment<byte>(buffer), default);
if (result.MessageType == WebSocketMessageType.Close) {
// TODO 区分主动断开和被动断开
OnClose?.Invoke(-1, null);
return;
}
// 拼合 WebSocket Frame
@ -91,8 +100,9 @@ namespace LeanCloud.Realtime.Internal.WebSocket {
}
}
} catch (Exception e) {
// TODO 连接断开
// 连接断开
LCLogger.Error(e.Message);
await ws.CloseAsync(WebSocketCloseStatus.EndpointUnavailable, "read error", default);
}
}

387
Realtime/LCIMClient.cs
View File

@ -4,37 +4,42 @@ using System.Threading.Tasks;
using System.Linq;
using LeanCloud.Realtime.Internal.WebSocket;
using LeanCloud.Realtime.Protocol;
using LeanCloud.Storage.Internal.Codec;
using LeanCloud.Storage.Internal;
using Newtonsoft.Json;
using LeanCloud.Realtime.Internal.Controller;
namespace LeanCloud.Realtime {
public class LCIMClient {
internal LCWebSocketConnection connection;
internal Dictionary<string, LCIMConversation> ConversationDict;
internal Dictionary<string, LCIMConversation> conversationDict;
public string ClientId {
public string Id {
get; private set;
}
// TODO 判断过期
internal string SessionToken {
get; private set;
}
#region 事件
/// <summary>
/// 当前用户被加入某个对话的黑名单
/// </summary>
public Action OnBlocked {
public Action<LCIMConversation, string> OnBlocked {
get; set;
}
/// <summary>
/// 当前客户端在某个对话中被禁言
/// 当用户被解除黑名单
/// </summary>
public Action<LCIMConversation, string> OnUnblocked {
get; set;
}
/// <summary>
/// 当前用户在某个对话中被禁言
/// </summary>
public Action<LCIMConversation, string> OnMuted;
/// <summary>
/// 当前用户在某个对话中被解除禁言
/// </summary>
public Action<LCIMConversation, string> OnUnmuted;
/// <summary>
/// 客户端连接断开
/// </summary>
@ -52,7 +57,7 @@ namespace LeanCloud.Realtime {
/// <summary>
/// 当前客户端被服务端强行下线
/// </summary>
public Action OnOffline {
public Action<int, string, string> OnClose {
get; set;
}
@ -134,26 +139,26 @@ namespace LeanCloud.Realtime {
/// <summary>
/// 有成员的对话信息被更新
/// </summary>
public Action<LCIMConversation, string, Dictionary<string, object>, string> OnMemberInfoUpdated;
public Action<LCIMConversation, string, string, string> OnMemberInfoUpdated;
/// <summary>
/// 当前用户收到消息
/// </summary>
public Action<LCIMConversation, LCIMMessage> OnMessageReceived {
public Action<LCIMConversation, LCIMMessage> OnMessage {
get; set;
}
/// <summary>
/// 消息被撤回
/// </summary>
public Action<LCIMConversation, LCIMMessage> OnMessageRecall {
public Action<LCIMConversation, LCIMMessage> OnMessageRecalled {
get; set;
}
/// <summary>
/// 消息被修改
/// </summary>
public Action<LCIMConversation, LCIMMessage> OnMessageUpdate {
public Action<LCIMConversation, LCIMMessage> OnMessageUpdated {
get; set;
}
@ -164,14 +169,62 @@ namespace LeanCloud.Realtime {
get; set;
}
/// <summary>
///
/// </summary>
public Action OnLastDeliveredAtUpdated {
get; set;
}
public Action OnLastReadAtUpdated {
get; set;
}
#endregion
internal ILCIMSignatureFactory SignatureFactory {
get; private set;
}
public LCIMClient(string clientId, ILCIMSignatureFactory signatureFactory = null) {
ClientId = clientId;
internal LCWebSocketConnection Connection {
get; set;
}
internal LCIMSessionController SessionController {
get; private set;
}
internal LCIMMessageController MessageController {
get; private set;
}
internal LCIMUnreadController UnreadController {
get; private set;
}
internal LCIMGoAwayController GoAwayController {
get; private set;
}
internal LCIMConversationController ConversationController {
get; private set;
}
public LCIMClient(string clientId,
ILCIMSignatureFactory signatureFactory = null) {
Id = clientId;
SignatureFactory = signatureFactory;
conversationDict = new Dictionary<string, LCIMConversation>();
ConversationDict = new Dictionary<string, LCIMConversation>();
SessionController = new LCIMSessionController(this);
ConversationController = new LCIMConversationController(this);
MessageController = new LCIMMessageController(this);
UnreadController = new LCIMUnreadController(this);
GoAwayController = new LCIMGoAwayController(this);
Connection = new LCWebSocketConnection(Id) {
OnNotification = OnNotification
};
}
/// <summary>
@ -179,22 +232,9 @@ namespace LeanCloud.Realtime {
/// </summary>
/// <returns></returns>
public async Task Open() {
connection = new LCWebSocketConnection(ClientId) {
OnNotification = OnNotification
};
await connection.Connect();
// Open Session
GenericCommand request = NewCommand(CommandType.Session, OpType.Open);
SessionCommand session = new SessionCommand();
if (SignatureFactory != null) {
LCIMSignature signature = SignatureFactory.CreateConnectSignature(ClientId);
session.S = signature.Signature;
session.T = signature.Timestamp;
session.N = signature.Nonce;
}
request.SessionMessage = session;
GenericCommand response = await connection.SendRequest(request);
SessionToken = response.SessionMessage.St;
await Connection.Connect();
// 打开 Session
await SessionController.Open();
}
/// <summary>
@ -202,91 +242,63 @@ namespace LeanCloud.Realtime {
/// </summary>
/// <returns></returns>
public async Task Close() {
GenericCommand request = NewCommand(CommandType.Session, OpType.Close);
await connection.SendRequest(request);
await connection.Close();
}
public async Task<LCIMChatRoom> CreateChatRoom(
string name,
Dictionary<string, object> properties = null) {
LCIMChatRoom chatRoom = await CreateConv(name: name, transient: true, properties: properties) as LCIMChatRoom;
return chatRoom;
// 关闭 session
await SessionController.Close();
await Connection.Close();
}
/// <summary>
/// 创建普通对话
/// </summary>
/// <param name="members"></param>
/// <param name="name"></param>
/// <param name="unique"></param>
/// <param name="properties"></param>
/// <returns></returns>
public async Task<LCIMConversation> CreateConversation(
IEnumerable<string> members,
string name = null,
bool unique = true,
Dictionary<string, object> properties = null) {
return await CreateConv(members: members, name: name, unique: unique, properties: properties);
return await ConversationController.CreateConv(members: members,
name: name,
unique: unique,
properties: properties);
}
/// <summary>
/// 创建聊天室
/// </summary>
/// <param name="name"></param>
/// <param name="properties"></param>
/// <returns></returns>
public async Task<LCIMChatRoom> CreateChatRoom(
string name,
Dictionary<string, object> properties = null) {
LCIMChatRoom chatRoom = await ConversationController.CreateConv(name: name,
transient: true,
properties: properties) as LCIMChatRoom;
return chatRoom;
}
/// <summary>
/// 创建临时对话
/// </summary>
/// <param name="members"></param>
/// <param name="ttl"></param>
/// <param name="properties"></param>
/// <returns></returns>
public async Task<LCIMTemporaryConversation> CreateTemporaryConversation(
IEnumerable<string> members,
int ttl = 86400,
Dictionary<string, object> properties = null) {
LCIMTemporaryConversation tempConversation = await CreateConv(members: members, temporary: true, temporaryTtl: ttl, properties: properties) as LCIMTemporaryConversation;
LCIMTemporaryConversation tempConversation = await ConversationController.CreateConv(members: members,
temporary: true,
temporaryTtl: ttl,
properties: properties) as LCIMTemporaryConversation;
return tempConversation;
}
private async Task<LCIMConversation> CreateConv(
IEnumerable<string> members = null,
string name = null,
bool transient = false,
bool unique = true,
bool temporary = false,
int temporaryTtl = 86400,
Dictionary<string, object> properties = null) {
GenericCommand request = NewCommand(CommandType.Conv, OpType.Start);
ConvCommand conv = new ConvCommand {
Transient = transient,
Unique = unique,
};
if (members != null) {
conv.M.AddRange(members);
}
if (!string.IsNullOrEmpty(name)) {
conv.N = name;
}
if (temporary) {
conv.TempConv = temporary;
conv.TempConvTTL = temporaryTtl;
}
if (properties != null) {
conv.Attr = new JsonObjectMessage {
Data = JsonConvert.SerializeObject(LCEncoder.Encode(properties))
};
}
if (SignatureFactory != null) {
LCIMSignature signature = SignatureFactory.CreateStartConversationSignature(ClientId, members);
conv.S = signature.Signature;
conv.T = signature.Timestamp;
conv.N = signature.Nonce;
}
request.ConvMessage = conv;
GenericCommand response = await connection.SendRequest(request);
string convId = response.ConvMessage.Cid;
if (!conversationDict.TryGetValue(convId, out LCIMConversation conversation)) {
if (transient) {
conversation = new LCIMChatRoom(this);
} else if (temporary) {
conversation = new LCIMTemporaryConversation(this);
} else if (properties != null && properties.ContainsKey("system")) {
conversation = new LCIMServiceConversation(this);
} else {
conversation = new LCIMConversation(this);
}
conversationDict[convId] = conversation;
}
// 合并请求数据
conversation.Name = name;
conversation.MemberIdList = members?.ToList();
// 合并服务端推送的数据
conversation.MergeFrom(response.ConvMessage);
return conversation;
}
/// <summary>
/// 获取某个特定的对话
/// </summary>
@ -331,164 +343,34 @@ namespace LeanCloud.Realtime {
return new LCIMConversationQuery(this);
}
#region 通知处理
private async Task OnNotification(GenericCommand notification) {
switch (notification.Cmd) {
case CommandType.Session:
await OnSessionNotification(notification);
await SessionController.OnNotification(notification);
break;
case CommandType.Conv:
OnConversationNotification(notification);
await ConversationController.OnNotification(notification);
break;
case CommandType.Direct:
await OnDirectNotification(notification.DirectMessage);
await MessageController.OnNotification(notification);
break;
case CommandType.Unread:
await OnUnreadNotification(notification.UnreadMessage);
await UnreadController.OnNotification(notification);
break;
case CommandType.Goaway:
await GoAwayController.OnNotification(notification);
break;
default:
break;
}
}
private async Task OnSessionNotification(GenericCommand notification) {
switch (notification.Op) {
case OpType.Closed:
await OnSessionClosed(notification.SessionMessage);
break;
default:
break;
}
}
private async Task OnSessionClosed(SessionCommand session) {
int code = session.Code;
string reason = session.Reason;
string detail = session.Detail;
await connection.Close();
// TODO 关闭连接后回调给开发者
}
private void OnConversationNotification(GenericCommand notification) {
ConvCommand conv = notification.ConvMessage;
switch (notification.Op) {
case OpType.Joined:
OnConversationJoined(conv);
break;
case OpType.MembersJoined:
OnConversationMembersJoined(conv);
break;
case OpType.Left:
OnConversationLeft(conv);
break;
case OpType.MembersLeft:
OnConversationMemberLeft(conv);
break;
case OpType.Updated:
OnConversationPropertiesUpdated(conv);
break;
case OpType.MemberInfoChanged:
OnConversationMemberInfoChanged(conv);
break;
default:
break;
}
}
private async void OnConversationJoined(ConvCommand conv) {
LCIMConversation conversation = await GetOrQueryConversation(conv.Cid);
conversation.MergeFrom(conv);
OnInvited?.Invoke(conversation, conv.InitBy);
}
private async void OnConversationMembersJoined(ConvCommand conv) {
LCIMConversation conversation = await GetOrQueryConversation(conv.Cid);
conversation.MergeFrom(conv);
OnMembersJoined?.Invoke(conversation, conv.M.ToList(), conv.InitBy);
}
private void OnConversationLeft(ConvCommand conv) {
if (conversationDict.TryGetValue(conv.Cid, out LCIMConversation conversation)) {
OnKicked?.Invoke(conversation, conv.InitBy);
}
}
private void OnConversationMemberLeft(ConvCommand conv) {
if (conversationDict.TryGetValue(conv.Cid, out LCIMConversation conversation)) {
List<string> leftIdList = conv.M.ToList();
OnMembersLeft?.Invoke(conversation, leftIdList, conv.InitBy);
}
}
private void OnConversationPropertiesUpdated(ConvCommand conv) {
if (conversationDict.TryGetValue(conv.Cid, out LCIMConversation conversation)) {
// TODO 修改对话属性，并回调给开发者
OnConversationInfoUpdated?.Invoke(conversation, null, conv.InitBy);
}
}
private void OnConversationMemberInfoChanged(ConvCommand conv) {
}
private async Task OnDirectNotification(DirectCommand direct) {
LCIMMessage message = null;
if (direct.HasBinaryMsg) {
// 二进制消息
byte[] bytes = direct.BinaryMsg.ToByteArray();
message = new LCIMBinaryMessage(bytes);
} else {
// 文本消息
string messageData = direct.Msg;
Dictionary<string, object> msg = JsonConvert.DeserializeObject<Dictionary<string, object>>(messageData,
new LCJsonConverter());
int msgType = (int)(long)msg["_lctype"];
switch (msgType) {
case -1:
message = new LCIMTextMessage();
break;
case -2:
message = new LCIMImageMessage();
break;
case -3:
message = new LCIMAudioMessage();
break;
case -4:
message = new LCIMVideoMessage();
break;
case -5:
message = new LCIMLocationMessage();
break;
case -6:
message = new LCIMFileMessage();
break;
default:
break;
}
message.Decode(direct);
}
// 获取对话
LCIMConversation conversation = await GetOrQueryConversation(direct.Cid);
OnMessageReceived?.Invoke(conversation, message);
}
private async Task OnUnreadNotification(UnreadCommand unread) {
List<LCIMConversation> conversationList = new List<LCIMConversation>();
foreach (UnreadTuple conv in unread.Convs) {
// 查询对话
LCIMConversation conversation = await GetOrQueryConversation(conv.Cid);
conversation.Unread = conv.Unread;
// TODO 反序列化对话
// 最后一条消息
JsonConvert.DeserializeObject<Dictionary<string, object>>(conv.Data);
conversationList.Add(conversation);
}
OnUnreadMessagesCountUpdated?.Invoke(conversationList);
}
#endregion
internal async Task<LCIMConversation> GetOrQueryConversation(string convId) {
if (conversationDict.TryGetValue(convId, out LCIMConversation conversation)) {
if (ConversationDict.TryGetValue(convId, out LCIMConversation conversation)) {
return conversation;
}
conversation = await GetConversation(convId);
@ -496,11 +378,16 @@ namespace LeanCloud.Realtime {
}
internal GenericCommand NewCommand(CommandType cmd, OpType op) {
GenericCommand command = NewCommand(cmd);
command.Op = op;
return command;
}
internal GenericCommand NewCommand(CommandType cmd) {
return new GenericCommand {
Cmd = cmd,
Op = op,
AppId = LCApplication.AppId,
PeerId = ClientId,
PeerId = Id,
};
}
@ -508,7 +395,7 @@ namespace LeanCloud.Realtime {
return new GenericCommand {
Cmd = CommandType.Direct,
AppId = LCApplication.AppId,
PeerId = ClientId,
PeerId = Id,
};
}
}

1
Realtime/Realtime.csproj
View File

@ -20,5 +20,6 @@
<Folder Include="Message\" />
<Folder Include="Internal\WebSocket\" />
<Folder Include="Signature\" />
<Folder Include="Internal\Controller\" />
</ItemGroup>
</Project>

2
Test/Realtime.Test/Conversation.cs
View File

@ -114,7 +114,7 @@ namespace Realtime.Test {
conversation.Name = "leancloud";
conversation["k1"] = "v1";
conversation["k2"] = "v2";
await conversation.Save();
await conversation.UpdateInfo();
Assert.AreEqual(conversation.Name, "leancloud");
Assert.AreEqual(conversation["k1"], "v1");

12
Test/RealtimeConsole/Program.cs
View File

@ -46,9 +46,9 @@ namespace RealtimeConsole {
//_ = OpenAndClose();
//SendMessage().Wait();
SendMessage().Wait();
_ = Unread();
//Unread().Wait();
Console.ReadKey(true);
}
@ -134,7 +134,7 @@ namespace RealtimeConsole {
LCIMClient c5 = new LCIMClient("c5");
await c5.Open();
await conversation.Add(new string[] { "c5" });
await conversation.AddMembers(new string[] { "c5" });
}
static async Task Signature() {
@ -152,9 +152,9 @@ namespace RealtimeConsole {
LCIMChatRoom chatRoom = await hello.CreateChatRoom(name);
Console.WriteLine(chatRoom.Name);
await chatRoom.Add(new string[] { "world" });
await chatRoom.AddMembers(new string[] { "world" });
await chatRoom.Remove(new string[] { "world" });
await chatRoom.RemoveMembers(new string[] { "world" });
}
static async Task TemporaryConversation() {
@ -190,7 +190,7 @@ namespace RealtimeConsole {
LCIMClient world = new LCIMClient("world");
await world.Open();
world.OnMessageReceived = (conv, message) => {
world.OnMessage = (conv, message) => {
Console.WriteLine(message);
if (message is LCIMTypedMessage typedMessage) {
Console.WriteLine(typedMessage["k1"]);