华为云会议对接，华为云会议事件消息订阅

最近做了一个对接华为云视频会议接口，订阅华为云会议事件消息的功能。做之前在网上搜索了一番，居然发现没有一个这方面的资料。决定整理一下分享出来，留给有缘人

具体的需求是，我们的app上集成了华为云会议sdk，在用户开启聊天的时候没有收到相关消息通知，解决思路是，当用户在app上调华为云会议sdk接口发起通话成功之后，调用服务端接口进行会议事件消息的订阅，由我们服务端与华为云会议服务端进行socket通信，通过订阅华为云会议事件消息给用户发送相关通知

这里只列出服务端与华为云会议服务端交互，订阅华为云会议事件消息的相关代码不涉及相关业务逻辑

1、获取accessToken 接口(在调用华为云会议的相关接口之前需要通过appid与appKey 进行鉴权获取accessToken) - http接口

参考：执行App ID鉴权_华为云会议 Meeting

String nonce = UUID.randomUUID().toString();
Map<String, String> headerMap = new HashMap<>(2);
//30分钟过期
long expireTime = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli() + 30 * 60 * 1000;
String data = apiAuthConfig.getAppId() + ":" + ":" + expireTime + ":" + nonce;
String encode = HmacSha256.encode(data, apiAuthConfig.getAppKey());
headerMap.put("Authorization", "HMAC-SHA256 signature=" + encode);
headerMap.put("Content-Type", "application/json; charset=UTF-8");
Map<String, Object> bodyMap = new HashMap<>(8);
bodyMap.put("appId", apiAuthConfig.getAppId());
bodyMap.put("clientType", 72);
bodyMap.put("expireTime", expireTime);
bodyMap.put("nonce", nonce);
String result = CreateSslClientDefault.doPost(apiAuthConfig.getDomain() + GET_API_ACCESS_TOKEN_URL, Jsons.toJson(bodyMap), headerMap);

2、获取会控token接口 - http接口

参考：获取会控Token_华为云会议 Meeting

Map<String, String> headerMap = new HashMap<>(2);
headerMap.put("X-Password", conferencePassword);
headerMap.put("X-Login-Type", "1");
Map<String, Object> queryMap = new HashMap<>(1);
queryMap.put("conferenceID", conferenceId);
String url = apiAuthConfig.getDomain() + GET_CONTROL_CONFERENCES_TOKEN_URL;
String result = CreateSslClientDefault.doGet(url, queryMap, headerMap);
log.info("getControlConferencesToken result : {}", result);

3、获取 websocket建联token

参考：获取WebSocket建链Token_华为云会议 Meeting

Map<String, String> headerMap = new HashMap<>(2);
headerMap.put("X-Conference-Authorization", controlToken);
Map<String, Object> queryMap = new HashMap<>(1);
queryMap.put("conferenceID", confID);
String url = apiAuthConfig.getDomain() + GET_WS_CONFERENCES_TOKEN_URL;
String result = CreateSslClientDefault.doGet(url, queryMap, headerMap);

4、websocket消息事件订阅

参考：信息订阅_华为云会议 Meeting

public class MeetingWebSocketClient implements Runnable {private MeetingHandlerContext context;public MeetingWebSocketClient(MeetingHandlerContext meetingHandlerContext){this.context = meetingHandlerContext;}@Overridepublic void run() {try{log.info("HwMeetingWebSocketClient start, confID : {}", context.getConfID());WebSocketClient client = new StandardWebSocketClient();// 添加自定义头部信息，如有需要WebSocketHttpHeaders headers = new WebSocketHttpHeaders();String url = context.getWsURL() + String.format(HwMeetingConstant.GET_CONTROL_INCREMENT_CONN_URL + "?confID=%s&tmpToken=%s", context.getConfID(), context.getWsToken());// 连接到WebSocket服务器ListenableFuture<WebSocketSession> future = client.doHandshake(new MeetingWebSocketHandler(context), url, headers);// 发送要订阅的消息sendSubscribeMessage(future.get());}catch (Exception e){e.printStackTrace();}}private void sendSubscribeMessage(WebSocketSession session) throws IOException {// 发送订阅消息String[] subscribeTypes = new String[]{/* ConfBasicInfoNotify,*/ConfDynamicInfoNotify,ParticipantsNotify,AttendeesNotify,InviteResultNotify,};Map<String, Object> data = new HashMap();data.put("subscribeType", subscribeTypes);data.put("confToken", context.getClToken());MeetingWebSocketMessage webSocketMessage = new MeetingWebSocketMessage();webSocketMessage.setAction("Subscribe");webSocketMessage.setSequence(UUID.randomUUID().toString());webSocketMessage.setData(Jsons.toJson(data));TextMessage subscribeMessage = new TextMessage(Jsons.toJson(webSocketMessage));session.sendMessage(subscribeMessage);}
}

5、websocket消息事件处理

public class MeetingWebSocketHandler implements WebSocketHandler {static int cpuNum = Runtime.getRuntime().availableProcessors();// 所有的 socket 链接的心跳都用同一个线程池去处理public static final ScheduledExecutorService HEART = Executors.newScheduledThreadPool(cpuNum + 1, (r) ->{Thread thread = new Thread(r);thread.setName("thread-heart");return thread;});private WebSocketNotifyHandler handler;private MeetingHandlerContext context;private ScheduledFuture<?> heartbeat;public MeetingWebSocketHandler(MeetingHandlerContext context){this.context = context;}@Overridepublic void afterConnectionEstablished(WebSocketSession session) throws Exception {log.info("web socket connection open");// 加载所有的事件处理器handler = new WebSocketNotifyHandler.Builder().addHandler(new AttendeesNotifyHandler(context)).addHandler(new ParticipantsNotifyHandler(context)).addHandler(new InviteResultNotifyHandler(context)).addHandler(new ConfDynamicInfoNotifyHandler(context)).get();// 启动发送心跳的任务(发送心跳消息 一分钟发送一次)startHeartBeatTask(session);// 通知所有的handler当前链接已建立handler.setConnOpen();}@Overridepublic void handleMessage(WebSocketSession session, final WebSocketMessage<?> message) throws Exception {MeetingWebSocketMessage event = JSON.parseObject(message.getPayload().toString(), MeetingWebSocketMessage.class);// 接受到 心跳结果消息、订阅结果消息 直接忽略if(event.getAction().equals(HeartBeat) || event.getAction().equals(Subscribe)){return;}String messageIdKey = RedisKeyHelper.getConferencesMessageIdKey(event.getMsgID());// 如果是重复消息则不处理if(duplicateMessageCheck(event, messageIdKey)){return;}// 执行消息处理器doHandler(handler, message);}private boolean duplicateMessageCheck(MeetingWebSocketMessage event, String redisCacheKey) {if(event == null || StringUtils.isBlank(event.getMsgID())){return true;}if(StringUtils.isNotBlank(context.getRedisContent().opsForValue().get(redisCacheKey))){return true;}return false;}@Overridepublic void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {// 链接异常log.error("web socket transport error : {}", exception.getMessage());}@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {log.info("web socket connection closed");// 当连服务端关闭连接时，中断心跳请求if(heartbeat != null){heartbeat.cancel(true);}// 通知所有的handler当前链接已断开handler.setConnClose();}@Overridepublic boolean supportsPartialMessages() {return false;}private void doHandler(WebSocketNotifyHandler handler, WebSocketMessage message){handler.doHandler(message);if(handler.next == null){return;}doHandler(handler.next, message);}private void startHeartBeatTask(WebSocketSession session){// 发送心跳消息 一分钟发送一次heartbeat = HEART.scheduleWithFixedDelay(() ->{log.info("confID : {}, HeartBeat...", context.getConfID());if(session.isOpen()){MeetingWebSocketMessage heartBeatMessageData = new MeetingWebSocketMessage();heartBeatMessageData.setAction("HeartBeat");heartBeatMessageData.setSequence(UUID.randomUUID().toString());TextMessage  heartMessage = new TextMessage(Jsons.toJson(heartBeatMessageData));try {session.sendMessage(heartMessage);} catch (Exception e) {log.error("HeartBeat error : {}", e.getMessage());}}}, 1, 1, TimeUnit.MINUTES);}
}

// 会议邀请消息推送
@Slf4j
public class AttendeesNotifyHandler extends WebSocketNotifyHandler<WebSocketMessage> {@Overridepublic void doHandler(final WebSocketMessage socketMessage) {MeetingWebSocketMessage event = JSON.parseObject(socketMessage.getPayload().toString(), MeetingWebSocketMessage.class);if (!event.getAction().equals(AttendeesNotify)) {return;}AttendeesNotifyMessage message = JSON.parseObject(socketMessage.getPayload().toString(), AttendeesNotifyMessage.class);if(message == null || message.getData() == null || message.getData().size() <= 0){return;}log.info("AttendeesNotifyHandler msg: {}", Jsons.toJson(message));}
}

// 会议状态信息推送
@Slf4j
public class ConfDynamicInfoNotifyHandler extends WebSocketNotifyHandler<WebSocketMessage> {@Overridepublic void doHandler(final WebSocketMessage socketMessage) {MeetingWebSocketMessage event = JSON.parseObject(socketMessage.getPayload().toString(), MeetingWebSocketMessage.class);if (!event.getAction().equals(ConfDynamicInfoNotify)) {return;}MeetingStatusNotifyMessage message = JSON.parseObject(socketMessage.getPayload().toString(), MeetingStatusNotifyMessage.class);log.info("ConfDynamicInfoNotifyHandler msg: {}", Jsons.toJson(message));}
}

// 会议邀请结果消息推送
@Slf4j
public class InviteResultNotifyHandler extends WebSocketNotifyHandler<WebSocketMessage> {@Overridepublic void doHandler(final WebSocketMessage socketMessage) {MeetingWebSocketMessage event = JSON.parseObject(socketMessage.getPayload().toString(), MeetingWebSocketMessage.class);if (!event.getAction().equals(InviteResultNotify)) {return;}InviteResultNotifyMessage message = JSON.parseObject(socketMessage.getPayload().toString(), InviteResultNotifyMessage.class);if(message == null || message.getData() == null || message.getData().size() <= 0){return;}log.info("InviteResultNotifyHandler msg: {}", Jsons.toJson(message));}
}

// 在线人数消息推送
@Slf4j
public class ParticipantsNotifyHandler extends WebSocketNotifyHandler<WebSocketMessage> {@Overridepublic void doHandler(final WebSocketMessage socketMessage) {MeetingWebSocketMessage event = JSON.parseObject(socketMessage.getPayload().toString(), MeetingWebSocketMessage.class);if (!event.getAction().equals(ParticipantsNotify)) {return;}ParticipantsNotifyMessage message = JSON.parseObject(socketMessage.getPayload().toString(), ParticipantsNotifyMessage.class);if(message == null || message.getData() == null || message.getData().size() <= 0){return;}log.info("ParticipantsNotifyHandler msg: {}", Jsons.toJson(message));}
}

// 消息事件处理器构造类public abstract class WebSocketNotifyHandler<T> {private static AtomicBoolean SOCKET_STATUS = new AtomicBoolean(false);protected WebSocketNotifyHandler<T> next;private void next(WebSocketNotifyHandler handler){this.next = handler;}public abstract void doHandler(final T data);public Boolean getConnStatus(){return SOCKET_STATUS.get();}public void setConnOpen(){SOCKET_STATUS.compareAndSet(false, true);}public void setConnClose(){SOCKET_STATUS.compareAndSet(true, false);}public static class Builder<T>{private WebSocketNotifyHandler<T> head;private WebSocketNotifyHandler<T> tail;public Builder<T> addHandler(WebSocketNotifyHandler<T> handler){if(this.head == null){this.head = this.tail = handler;return this;}this.tail.next(handler);this.tail = handler;return this;}public WebSocketNotifyHandler<T> get(){return this.head;}}
}

// websocket 返回的消息体，T类型由具体事件类型确定
@Data
public class MeetingWebSocketMessage<T> {/*** 消息类型*/String action;/*** 消息随机序列号*/String sequence;/*** 消息id*/String msgID;/*** 消息体*/T data;}