外贸展示企业网站,网站建设公司itcask,做一个网站一般要多少钱,给六人游做网站开发的最近做了一个对接华为云视频会议接口#xff0c;订阅华为云会议事件消息的功能。做之前在网上搜索了一番#xff0c;居然发现没有一个这方面的资料。决定整理一下分享出来#xff0c;留给有缘人 具体的需求是#xff0c;我们的app上集成了华为云会议sdk#xff0c;在用户开…       最近做了一个对接华为云视频会议接口订阅华为云会议事件消息的功能。做之前在网上搜索了一番居然发现没有一个这方面的资料。决定整理一下分享出来留给有缘人 具体的需求是我们的app上集成了华为云会议sdk在用户开启聊天的时候没有收到相关消息通知解决思路是当用户在app上调华为云会议sdk接口发起通话成功之后调用服务端接口进行会议事件消息的订阅由我们服务端与华为云会议服务端进行socket通信通过订阅华为云会议事件消息给用户发送相关通知 这里只列出 服务端与华为云会议服务端交互订阅华为云会议事件消息的相关代码不涉及相关业务逻辑 1、获取accessToken 接口(在调用华为云会议的相关接口之前需要通过appid与appKey 进行鉴权获取accessToken)  - http接口 参考执行App ID鉴权_华为云会议 Meeting String nonce UUID.randomUUID().toString(); MapString, 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; charsetUTF-8); MapString, 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 MapString, String headerMap new HashMap(2); headerMap.put(X-Password, conferencePassword); headerMap.put(X-Login-Type, 1); MapString, 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 MapString, String headerMap new HashMap(2); headerMap.put(X-Conference-Authorization, controlToken); MapString, 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%stmpToken%s, context.getConfID(), context.getWsToken());// 连接到WebSocket服务器ListenableFutureWebSocketSession 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,};MapString, 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 WebSocketNotifyHandlerWebSocketMessage {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 WebSocketNotifyHandlerWebSocketMessage {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 WebSocketNotifyHandlerWebSocketMessage {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 WebSocketNotifyHandlerWebSocketMessage {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 WebSocketNotifyHandlerT {private static AtomicBoolean SOCKET_STATUS new AtomicBoolean(false);protected WebSocketNotifyHandlerT 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 BuilderT{private WebSocketNotifyHandlerT head;private WebSocketNotifyHandlerT tail;public BuilderT addHandler(WebSocketNotifyHandlerT handler){if(this.head null){this.head this.tail handler;return this;}this.tail.next(handler);this.tail handler;return this;}public WebSocketNotifyHandlerT get(){return this.head;}} } // websocket 返回的消息体T类型由具体事件类型确定 Data public class MeetingWebSocketMessageT {/*** 消息类型*/String action;/*** 消息随机序列号*/String sequence;/*** 消息id*/String msgID;/*** 消息体*/T data;}