四,详解消息 hook 与系统对话
本章导读
在前一篇安全与签名、玩转聊天室和临时对话中,我们解释了一些第三方鉴权方面的问题,在这里我们会更进一步,给大家说明:
- 即时通讯的消息 Hook 机制
- 系统对话的使用方法
万能的 Hook 机制
完全开放的架构,支持强大的业务扩展能力,是即时通讯服务的特色之一,这种优势的体现就是这里将要给大家介绍的「Hook 机制」。
Hook 与即时通讯服务的关系
Hook 也可以称为「钩子」,是一种特殊的消息处理机制,与 Windows 平台下的中断机制类似,允许应用方拦 截并处理即时通讯过程中的多种事件和消息,从而达到实现自定义业务逻辑的目的。
以 _messageRecieved Hook 为例,它在消息送达服务器后会被调用,在 Hook 内可以捕获消息内容、消息发送者、消息接收者等信息,这些信息均能在 Hook 内做修改并将修改后的值转交回服务器,服务器会使用修改后的消息继续完成消息投递工作。最终收消息用户收到的会是被 Hook 修改过后的消息,而不再是最初送达服务器的原始消息。Hook 也可以选择拒绝消息发送,服务器会在给客户端回复消息被 Hook 拒绝后丢弃消息不再完成后续消息处理及转发流程。
需要注意的是,默认情况下如果 Hook 调用失败,例如超时、返回状态码非 200 的结果等,服务器会忽略 Hook 的错误继续处理原始请求。如果你需要改变这个行为,可以在 开发者中心 > 你的游戏 > 游戏服务 > 云服务 > 即时通讯 > 设置 > 即时通讯设置 内开启「Hook 调用失败时返回错误给客户端并放弃继续处理请求」。开启后如果 Hook 调用失败,服务器会返回错误信息给客户端告知 Hook 调用错误,并拒绝继续处理请求。
消息类 Hook
一条消息,在即时通讯的流程中,从终端用户 A 发送开始,到其他用户接收到为止,考虑到存在接收方在线/不在线的可能,会经历多个不同阶段,这里每一个阶段都会触发 Hook 函数:
- _messageReceived
消息达到服务器,群组成员已解析完成之后,发送给收件人之前调用。开发者在这里还可以修改消息内容,实时改变消息接收者的列表,以及其他类似操作。 - _messageSent
消息发送完成后调用。开发者在这里可以完成业务统计,或将消息中转备份到己方服务器,以及其他类似操作。 - _receiversOffline
消息发送完成,存在离线的收件人,在发推送给收件人之前调用。开发者在这里可以动态修改离线推送的通知内容,或通知目的设备的列表,以及其他类似操作。 - _messageUpdate
收到消息修改请求,发送修改后的消息给收件人之前调用。与新发消息一样,开发者在这里可以再次修改消息内容,实时改变消息接收者的列表,以及其他类似操作。
对话类 Hook
在对话创建和成员变动等更改性操作前后,都可以触发 Hook 函数,进行额外的处理:
- _conversationStart
创建对话,在签名校验(如果开启)之后,实际创建之前调用。开发者在这里可以为新的「对话」添加其他内部属性,或完成操作鉴权,以及其他类似操作。 - _conversationStarted
创建对话完成后调用。开发者在这里可以完成业务统计,或将对话数据中转备份到己方服务器,以及其他类似操作。 - _conversationAdd
向对话添加成员,在签名校验(如果开启)之后,实际加入之前调用,包括主动加入和被其他用户加入两种情况。开发者可以在这里根据内部权限设置批准或驳回这一请求,以及其他类似操作。 - _conversationRemove
从对话中踢出成员,在签名校验(如果开启)之后,实际踢出之前调用,用户自己退出对话不会调用。开发者可以在这里根据内部权限设置批准或驳回这一请求,以及其他类似操作。 - _conversationAdded
用户加入对话,在加入成功后调用。 - _conversationRemoved
用户离开对话,在离开成功后调用。 - _conversationUpdate
修改对话名称、自定义属性,设置或取消对话消息提醒,在实际修改之前调用。开发者在这里可以为新的「对话」添加其他内部属性,或完成操作鉴权,以及其他类似操作。
客户端上下线 Hook
在客户端上线和下线的时候,可以触发 Hook 函数:
- _clientOnline
客户端上线,客户端登录成功后调用。 - _clientOffline
客户端下线,客户端登出成功或意外下线后调用。
开发者可以利用这两个 Hook 函数,结合 LeanCache 来完成一组客户端实时状态查询的 endpoint,具体可以参考文档《即时通讯中的在线状态查询》。
Hook 与云引擎的关系
因为 Hook 发生在即时通讯的在线处理环节,而即时通讯服务端每秒钟需要处理的消息和对话事件数量远超大家的想象,出于性能考虑,我们要求开发者使用云引擎来实现 Hook 函数。
即时通讯的云引擎 Hook 要求云引擎部署在云引擎的 生产环境,测试环境仅用于开发者手动调用测 试。由于缓存的原因,首次部署的云引擎 Hook 需要至多三分钟来正式生效,后续修改会实时生效。
Hook API 细节与使用场景详解
与 conversation
相关的 hook 可以在应用签名之外增加额外的权限判断,控制对话是否允许被建立、某些用户是否允许被加入对话等。你可以用这一 hook 实现黑名单功能。
_messageReceived
这个 hook 发生在消息到达云端之后。如果是群组消息,我们会解析出所有消息收件人。
你可以通过返回参数控制消息是否需要被丢弃,删除个别收件人,还可以修改消息内容,例如过滤应用中的敏感词。返回空对象(response.success({})
)则会执行系统默认的流程。
请注意,在这个 hook 的代码实现的任何分支上 请确保最终会调用 response.success
返回结果,使得消息可以尽快投递给收件人。这个 hook 将 阻塞发送流程,因此请尽量减少无谓的代码调用,提升效率。
如果你使用了默认提供的富媒体消息格式,云引擎参数中的 content
接收的是 JSON 结构的字符串形式。关于这个结构的详细说明,请参考即时通讯 REST API 使用指南的《富媒体消息格式说明》一节。
参数:
参数 | 说明 |
---|---|
fromPeer | 消息发送者的 ID。 |
convId | 消息所属对话的 ID。 |
toPeers | 解析出的对话相关的 clientId 。 |
transient | 是否是 transient 消息。 |
bin | 原始消息内容是否为二进制消息。 |
content | 消息体字符串。如果 bin 为 true ,则该字段为原始消息内容做 Base64 转码后的结果。 |
receipt | 是否要求回执。 |
timestamp | 服务器收到消息的时间戳(毫秒)。 |
system | 是否属于系统对话消息。 |
sourceIP | 消息发送者的 IP。 |
参数示例:
{
"fromPeer": "Tom",
"receipt": false,
"groupId": null,
"system": null,
"content": "{\"_lctext\":\"来我们去 XX 传奇玩吧\",\"_lctype\":-1}",
"convId": "5789a33a1b8694ad267d8040",
"toPeers": ["Jerry"],
"bin": false,
"transient": false,
"sourceIP": "121.239.62.103",
"timestamp": 1472200796764
}
返回值:
参数 | 约束 | 说明 |
---|---|---|
drop | 可选 | 如果返回真值,消息将被丢弃。 |
code | 可选 | 当 drop 为 true 时可以下发一个应用自定义的整型错误码。 |
detail | 可选 | 当 drop 为 true 时可以下发一个应用自定义的错误说明字符串。 |
bin | 可选 | 返回的 content 内是否为二进制消息,如果不提供则为请求中的 bin 值。 |
content | 可选 | 修改后的 content ,如果不提供则保留原消息。如果 bin 为 true ,则 content 需要是二进制消息内容做 Base64 转码后的结果。 |
toPeers | 可选 | 数组,修改后的收件人,如果不提供则保留原收件人。 |
示例代码:
- JavaScript
- Python
- PHP
- Java
- C#
- Go
AV.Cloud.onIMMessageReceived((request) => {
let content = request.params.content;
let processedContent = content.replace("XX 传奇", "**");
// 必须含有以下语句给服务端一个正确的返回,否则会引起异常
return {
content: processedContent,
};
});
import json
@engine.define
def _messageReceived(**params):
content = json.loads(params['content'])
text = content['_lctext']
content['_lctext'] = text.replace('XX 传奇', '**')
# 必须含有以下语句给服务端一个正确的返回,否则会引起异常
return {
'content': json.dumps(content)
}
Cloud::define("_messageReceived", function($params, $user) {
$content = json_decode($params["content"], true);
$text = $content["_lctext"];
$content["_lctext"] = preg_replace("XX 传奇", "**", $text);
// 必须含有以下语句给服务端一个正确的返回,否则会引起异常
return array("content" => json_encode($content));
});
@IMHook(type = IMHookType.messageReceived)
public static Map<String, Object> onMessageReceived(Map<String, Object> params) {
Map<String, Object> result = new HashMap<String, Object>();
String content = (String)params.get("content");
String processedContent = content.replace("XX 传奇", "**");
result.put("content", processedContent);
// 必须含有以下语句给服务端一个正确的返回,否则会引起异常
return result;
}
[LCEngineRealtimeHook(LCEngineRealtimeHookType.MessageReceived)]
public static object OnMessageReceived(Dictionary<string, object> parameters) {
string content = parameters["content"] as string;
string processedContent = content.Replace("XX 中介", "**");
return new Dictionary<string, object> {
{ "content", processedContent }
};
}
// 暂不支持
而实际上启用上述代码之后,一条消息的时序图如下:
- 上图假设的是对话所有成员都在线,而如果有成员不在线,流程有些不一样,下一节会做介绍。
- RTM 表示即时通讯服务集群,Engine 表示云引擎服务集群,它们基于内网通讯。
_receiversOffline
这个 hook 发生在有收件人离线的情况下,你可以通过它来自定义离线推送行为,包括推送内容、被推送用户或略过推送。你也可以直接在 hook 中触发自定义的推送。发往聊天室的消息不会触发此 hook。
参数:
参数 | 说明 |
---|---|
fromPeer | 消息发送者 ID。 |
convId | 消息所属对话的 ID。 |
offlinePeers | 数组,离线的收件人列表。 |
content | 消息内容。 |
timestamp | 服务器收到消息的时间戳( 毫秒)。 |
mentionAll | 布尔类型,表示本消息是否 @ 了所有成员。 |
mentionOfflinePeers | 被本消息 @ 且离线的成员 ID。如果 mentionAll 为 true ,则该参数为空,表示所有 offlinePeers 参数内的成员全部被 @。 |
返回值:
参数 | 约束 | 说明 |
---|---|---|
skip | 可选 | 如果为真将跳过推送(比如已经在云引擎里触发了推送或者其他通知)。 |
offlinePeers | 可选 | 数组,筛选过的推送收件人。 |
pushMessage | 可选 | 推送内容,支持自定义 JSON 结构。 |
force | 可选 | 如果为真将强制推送给 offlinePeers 里 mute 的用户,默认 false 。 |
示例代码:
- JavaScript
- Python
- PHP
- Java
- C#
- Go
AV.Cloud.onIMReceiversOffline((request) => {
let params = request.params;
let content = params.content;
// params.content 为消息的内容
let shortContent = content;
if (shortContent.length > 6) {
shortContent = content.slice(0, 6);
}
console.log("shortContent", shortContent);
return {
pushMessage: JSON.stringify({
// 自增未读消息的数目,不想自增就设为数字
badge: "Increment",
sound: "default",
// 使用开发证书
_profile: "dev",
alert: shortContent,
}),
};
});
@engine.define
def _receiversOffline(**params):
print('_receiversOffline start')
# params['content'] 为消息内容
content = params['content']
short_content = content[:6]
print('short_content:', short_content)
payloads = {
# 自增未读消息的数目,不想自增就设为数字
'badge': 'Increment',
'sound': 'default',
# 使用开发证书
'_profile': 'dev',
'alert': short_content,
}
print('_receiversOffline end')
return {
'pushMessage': json.dumps(payloads),
}
Cloud::define('_receiversOffline', function($params, $user) {
error_log('_receiversOffline start');
// content 为消息的内容
$shortContent = $params["content"];
if (strlen($shortContent) > 6) {
$shortContent = substr($shortContent, 0, 6);
}
$json = array(
// 自增未读消息的数目,不想自增就设为数字
"badge" => "Increment",
"sound" => "default",
// 使用开发证书
"_profile" => "dev",
"alert" => shortContent
);
$pushMessage = json_encode($json);
return array(
"pushMessage" => $pushMessage,
);
});
@IMHook(type = IMHookType.receiversOffline)
public static Map<String, Object> onReceiversOffline(Map<String, Object> params) {
// content 为消息内容
String alert = (String)params.get("content");
if(alert.length() > 6){
alert = alert.substring(0, 6);
}
System.out.println(alert);
Map<String, Object> result = new HashMap<String, Object>();
JSONObject object = new JSONObject();
// 自增未读消息的数目
// 不想自增就设为数字
object.put("badge", "Increment");
object.put("sound", "default");
// 使用开发证书
object.put("_profile", "dev");
object.put("alert", alert);
result.put("pushMessage", object.toString());
return result;
}
[LCEngineRealtimeHook(LCEngineRealtimeHookType.ReceiversOffline)]
public static Dictionary<string, object> OnReceiversOffline(Dictionary<string, object> parameters) {
string alert = parameters["content"] as string;
if (alert.Length > 6) {
alert = alert.Substring(0, 6);
}
Dictionary<string, object> pushMessage = new Dictionary<string, object> {
{ "badge", "Increment" },
{ "sound", "default" },
{ "_profile", "dev" },
{ "alert", alert },
};
return new Dictionary<string, object> {
{ "pushMessage", JsonSerializer.Serialize(pushMessage) }
};
}
// 暂不支持
_messageSent
在消息发送完成后执行,对消息发送性能没有影响,可以用来执行相对耗时的逻辑。
参数:
参数 | 说明 |
---|---|
fromPeer | 消息发送者的 ID。 |
convId | 消息所属对话的 ID。 |
msgId | 消息 ID。 |
onlinePeers | 当前在线发送的用户 ID。 |
offlinePeers | 当前离线的用户 ID。 |
transient | 是否是 transient 消息。 |
system | 是否是 system conversation。 |
bin | 是否是二进制消息。 |
content | 消息体字符串。 |
receipt | 是否要求回执。 |
timestamp | 服务器收到消息的时间戳(毫秒)。 |
sourceIP | 消息发送者的 IP。 |
参数示例:
{
"fromPeer": "Tom",
"receipt": false,
"onlinePeers": [],
"content": "12345678",
"convId": "5789a33a1b8694ad267d8040",
"msgId": "fptKnuYYQMGdiSt_Zs7zDA",
"bin": false,
"transient": false,
"sourceIP": "114.219.127.186",
"offlinePeers": ["Jerry"],
"timestamp": 1472703266522
}
返回值:
这个 hook 不会对返回值进行检查。只需返回 {}
即可。
示例代码:
下面代码演示了日志记录相关的操作(在消息发送完后,在云引擎中打印一下日志):
- JavaScript
- Python
- PHP
- Java
- C#
- Go
AV.Cloud.onIMMessageSent((request) => {
console.log("params", request.params);
});
@engine.define
def _messageSent(**params):
print('_messageSent start')
print('params:', params)
print('_messageSent end')
return {}
Cloud::define('_messageSent', function($params, $user) {
error_log('_messageSent start');
error_log('params' . json_encode($params));
return array();
});
@IMHook(type = IMHookType.messageSent)
public static Map<String, Object> onMessageSent(Map<String, Object> params) {
System.out.println(params);
Map<String, Object> result = new HashMap<String, Object>();
return result;
}
[LCEngineRealtimeHook(LCEngineRealtimeHookType.MessageSent)]
public static Dictionary<string, object> OnMessageSent(Dictionary<string, object> parameters) {
Console.WriteLine(JsonSerializer.Serialize(parameters));
return default;
}
// 暂不支持
_messageUpdate
这个 hook 发生在修改消息请求到达云端,云端正式修改消息之前。
你可以通过返回参数控制修改消息请求是否需要被丢弃,删除个别收件人,或再次修改这个修改消息请求中的消息内容。
请注意,在这个 hook 的代码实现的任何分支上 请确保最终会调用 response.success
返回结果,使得修改消息可以尽快投递给收件人。这个 hook 将 阻塞发送流程,因此请尽量减少无谓的代码调用,提升效率。
如果你使用了默认提供的富媒体消息格式,云引擎参数中的 content
接收的是 JSON 结构的字符串形式。关于这个结构的详细说明,请参考即时通讯 REST API 使用指南的《富媒体消息》一节。
参数:
参数 | 说明 |
---|---|
fromPeer | 消息发送者的 ID。 |
convId | 消息所属对话的 ID。 |
toPeers | 解析出的对话相关的 clientId 。 |
bin | 原始消息内容是否为二进制消息。 |
content | 消息体字符串。如果 bin 为 true ,则该字段为原始消息内容做 Base64 转码后的结果。 |
timestamp | 服务器收到消息的时间戳(毫秒)。 |
msgId | 被修改的消息 ID。 |
sourceIP | 消息发送者的 IP。 |
recall | 是否撤回消息。 |
system | 是否属于系统对话消息。 |
返回值:
参数 | 约束 | 说明 |
---|---|---|
drop | 可选 | 如果返回真值,修改消息请求将被丢弃。 |
code | 可选 | 当 drop 为 true 时可以下发一个应用自定义的整型错误码。 |
detail | 可选 | 当 drop 为 true 时可以下发一个应用自定义的错误说明字符串。 |
bin | 可选 | 返回的 content 内是否为二进制消息,如果不提供则为请求中的 bin 值。 |
content | 可选 | 修改后的 content ,如果不提供则保留原消息。如果 bin 为 true ,则 content 需要是二进制消息内容做 Base64 转码后的结果。 |
toPeers | 可选 | 数组,修改后的收件人,如果不提供则保留原收件人。 |
_conversationStart
在创建对话时调用,发生在签名验证(如果开启)之后、创建对话之前。
参数:
参数 | 说明 |
---|---|
initBy | 由谁发起的 clientId 。 |
members | 初始成员数组,包含初始成员。 |
attr | 创建对话时的额外属性。 |
参数示例:
{
"initBy": "Tom",
"members": ["Tom", "Jerry"],
"attr": {
"name": "Tom & Jerry"
}
}
返回值:
参数 | 约束 | 说明 |
---|---|---|
reject | 可选 | 是否拒绝,默认为 false 。 |
code | 可选 | 当 reject 为 true 时可以下发一个应用自定义的整型错误码。 |
detail | 可选 | 当 reject 为 true 时可以下发一个应用自定义的错误说明字符串。 |
例如,初始成员不足四人,不允许创建对话:
- JavaScript
- Python
- PHP
- Java
- C#
- Go
AV.Cloud.onIMConversationStart((request) => {
if (request.params.members.length < 4) {
return {
reject: true,
code: 1234,
detail: "至少邀请 3 人开启对话",
};
} else {
return {};
}
});
@engine.define
def _conversationStart(**params):
if len(params["members"]) < 4:
return {
"reject": True,
"code": 1234,
"detail": "至少邀请 3 人开启对话",
}
else:
return {}
Cloud::define('_conversationStart', function($params, $user) {
if (count($params["members"]) < 4) {
return [
"reject" => true,
"code" => 1234,
"detail" => "至少邀请 3 人开启对话",
];
} else {
return array();
}
});
@IMHook(type = IMHookType.conversationStart)
public static Map<String, Object> onConversationStart(Map<String, Object> params) {
String[] members = (String[])params.get("members");
Map<String, Object> result = new HashMap<String, Object>();
if (members.length < 4) {
result.put("reject", true);
result.put("code", 1234);
result.put("detail", "至少邀请 3 人开启对话");
}
return result;
}
[LCEngineRealtimeHook(LCEngineRealtimeHookType.ConversationStart)]
public static object OnConversationStart(Dictionary<string, object> parameters) {
List<object> members = parameters["members"] as List<object>;
if (members.Count < 4) {
return new Dictionary<string, object> {
{ "reject", true },
{ "code", 1234 },
{ "detail", "至少邀请 3 人开启对话" }
};
}
return default;
}
// 暂不支持
_conversationStarted
对话创建后调用。
参数:
参数 | 说明 |
---|---|
convId | 新生成的对话 ID。 |
返回值:
这个 hook 不对返回值进行处理,只需返回 {}
即可。
例如,创建对话后,将对话的 ID 存储到 LeanCache 的最近创建对话列表:
- JavaScript
- Python
- PHP
- Java
- C#
- Go
AV.Cloud.onIMConversationStarted((request) => {
redisClient.lpush("recent_conversations", request.params.convId);
return {};
});
@engine.define
def _conversationStarted(**params):
redis_client.lpush("recent_conversations", params["convId"])
return {}
Cloud::define('_conversationStarted', function($params, $user) {
$redis->lpush("recent_conversations", $params["convId"]);
return array();
});
@IMHook(type = IMHookType.conversationStarted)
public static Map<String, Object> onConversationStarted(Map<String, Object> params) throws Exception {
String convId = (String)params.get("convId");
jedis.lpush("recent_conversations", params.get("convId"));
Map<String, Object> result = new HashMap<String, Object>();
return result;
}