视频通话

zengh

2021-08-12 263671b4a257a20c6f4e3c1eab2a5065ca27b3a9

视频通话

 src/main/java/org/springblade/modules/webscoket/ChannelSupervise.java

New file
@@ -0,0 +1,38 @@
package org.springblade.modules.webscoket;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class ChannelSupervise {
    private   static ChannelGroup GlobalGroup=new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private  static ConcurrentMap<String, ChannelId> ChannelMap=new ConcurrentHashMap();
    private  static Map<String, String> map = new HashMap<String, String>();;
    public  static void addChannel(Channel channel,String name){
        GlobalGroup.add(channel);
        ChannelMap.put(channel.id().asShortText(),channel.id());
        map.put(channel.id().asShortText(),name);
    }
    public static void removeChannel(Channel channel){
        GlobalGroup.remove(channel);
        ChannelMap.remove(channel.id().asShortText());
        map.remove(channel.id().asShortText());
    }
    public static  Channel findChannel(String id){
        return GlobalGroup.find(ChannelMap.get(id));
    }
    public static String findName(String id){
    return map.get(id);
    }
    public static void send2All(TextWebSocketFrame tws){
        GlobalGroup.writeAndFlush(tws);
    }

}

 src/main/java/org/springblade/modules/webscoket/WebSocketHandler.java

New file
@@ -0,0 +1,193 @@
package org.springblade.modules.webscoket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import org.springblade.modules.nettyServer.NettyConfig;
import org.springblade.modules.suser.service.ISuserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
public class WebSocketHandler extends SimpleChannelInboundHandler<Object> {

    private WebSocketServerHandshaker handshaker;

    private  String on=null;

    @Autowired
    private  ISuserService suserService;

    private static WebSocketHandler webSocketHandler;

    @PostConstruct
    public void init() {
        webSocketHandler = this;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            //以http请求形式接入，但是走的是websocket
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            //处理websocket客户端的消息
            handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //添加连接
        System.out.println("客户端加入连接：" + ctx.channel());
        //ChannelSupervise.addChannel(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //断开连接
        System.out.println("客户端断开连接：" + ctx.channel());
        //用户离线状态
        String name = ChannelSupervise.findName(ctx.channel().id().asShortText());
        System.out.println(name);
        if ( name != null &&!name.equals("ping") ){
            NettyConfig.getChannelGroup().remove(ctx.channel());
            removeUserId(ctx);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws NumberFormatException, Exception {
        // 判断是否关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判断是否ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(
                new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 本例程仅支持文本消息，不支持二进制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            System.out.println("本例程仅支持文本消息，不支持二进制消息");
            throw new UnsupportedOperationException(String.format(
                "%s frame types not supported", frame.getClass().getName()));
        }
        // 返回应答消息
        String request = ((TextWebSocketFrame) frame).text();

        if (!request.equals("ping")){

            JSONObject jsonObj = JSON.parseObject(request);
            String type = jsonObj.get("type").toString();

            if (type != null && type.equals("login")){
                //登录链接
                String id = jsonObj.get("id").toString();
                NettyConfig.getUserChannelMap().put(id,ctx.channel());

                System.out.println(jsonObj.get("type"));
                System.out.println(jsonObj.get("id"));

                //将用户id作为自定义属性加入到channel 中，方便随时channel中获取用户id
                AttributeKey<String> key = AttributeKey.valueOf("userId");
                ctx.channel().attr(key).setIfAbsent(id);
                //把用户信息添加到通道里
                ChannelSupervise.addChannel(ctx.channel(),id);
            }






        }

//        TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString()
//            + ctx.channel().id() + "：" + ctx.channel());
//        // 返回【谁发的发给谁】
//
//        ctx.channel().writeAndFlush(tws);
        // 群发
        //  ChannelSupervise.send2All(tws);
    }

    /**
     * 唯一的一次http请求，用于创建websocket
     *
     * @throws InterruptedException
     */
    private void handleHttpRequest(final ChannelHandlerContext ctx,
                                   FullHttpRequest req) throws InterruptedException {
        //要求Upgrade为websocket，过滤掉get/Post
        if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
            //若不是websocket方式，则创建BAD_REQUEST的req，返回给客户端
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        //握手
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
            "ws://localhost:9034/websocket", null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory
                .sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }

    /**
     * 拒绝不合法的请求，并返回错误信息
     */
    private static void sendHttpResponse(ChannelHandlerContext ctx,
                                         FullHttpRequest req, DefaultFullHttpResponse res) {
        // 返回应答给客户端
        if (res.status().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(),
                CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        }
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        // 如果是非Keep-Alive，关闭连接
//        if (!isKeepAlive(req) || res.status().code() != 200) {
//            f.addListener(ChannelFutureListener.CLOSE);
//        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    /**
     * 删除用户与channel 对应关系
     * @param ctx
     */
    private void removeUserId(ChannelHandlerContext ctx){
        AttributeKey<String> key = AttributeKey.valueOf("userId");
        String userId = ctx.channel().attr(key).get();
        NettyConfig.getUserChannelMap().remove(userId);
    }

}

 src/main/java/org/springblade/modules/webscoket/WebSocketServer.java

New file
@@ -0,0 +1,59 @@
package org.springblade.modules.webscoket;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebSocketServer {
    private int port = 9034;

    public WebSocketServer(int port) {
        bind(port);
    }

    public void bind(int port) {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                EventLoopGroup bossGroup = new NioEventLoopGroup();
                EventLoopGroup workerGroup = new NioEventLoopGroup();
                try {
                    ServerBootstrap serverBootstrap = new ServerBootstrap();
                    serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        //保持连接
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) {
//                                ch.pipeline().addLast("logging",new LoggingHandler("DEBUG"));//设置log监听器，并且日志级别为debug，方便观察运行流程
                                ch.pipeline().addLast("http-codec", new HttpServerCodec());//设置解码器
                                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));//聚合器，使用websocket会用到
                                ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());//用于大数据的分区传输
                                ch.pipeline().addLast("handler", new WebSocketHandler());//自定义的业务handler
                            }
                        });

                    ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                    System.out.println("WebSocketServer启动成功");
                    channelFuture.channel().closeFuture().sync();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    bossGroup.shutdownGracefully();
                    workerGroup.shutdownGracefully();
                }
            }
        });
        thread.start();
    }
}

 src/main/java/org/springblade/modules/webscoket/controller/PushMsgController.java

New file
@@ -0,0 +1,59 @@
package org.springblade.modules.webscoket.controller;

import org.springblade.core.tool.api.R;
import org.springblade.modules.webscoket.service.IPushMsgService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * @author lq
 * @date 2020/4/1 11:22
 */
@RestController
@RequestMapping("pushMsg")
public class PushMsgController {

    @Autowired
    private IPushMsgService pushMsgService;

    @PostMapping("/pushUser")
    public String pushUser(String userId, String msg) {
        pushMsgService.pushMsg(userId, msg);
        return "消息发送成功：" + msg;
    }

    @PostMapping("/pushAll")
    public String pushAll(String msg) {
        pushMsgService.pushMsg(msg);
        return "消息发送成功：" + msg;
    }

    @GetMapping("/inviteVideoCall")
    public R<Map> inviteVideoCall(String userId, String type, String name,String faqiid) {
        //获取当前时间戳作为房间号
        String roomId = "";
        Map<String, Object> map = new HashMap<String, Object>();
        String time = String.valueOf(new Date().getTime());
        int msg = pushMsgService.inviteVideoCall(userId, time, type, name,faqiid);
        map.put("type", type);
        map.put("roomId", time);
        map.put("res", msg);

        return R.data(map);
    }

    @GetMapping("/closeVideoCall")
    public void closeVideoCall(String sentId, String acceptId) {

        pushMsgService.closeVideoCall(sentId, acceptId);

    }

}

 src/main/java/org/springblade/modules/webscoket/index.html

New file
@@ -0,0 +1,57 @@
<html>
<head>
    <meta http-equiv="Content-Type" content="text/html; charset = utf-8"/>
    <title>WebSocket客户端</title>
    <script type="text/javascript">
        var socket;
        if(!window.WebSocket){
            window.WebSocket = window.MozWebSocket;
        }

        if(window.WebSocket){
            socket = new WebSocket("ws://localhost:9034/websocket");
            socket.onmessage = function(event){
                var ta = document.getElementById('responseContent');
                ta.value += event.data + "\r\n";
            };

            socket.onopen = function(event){
                var ta = document.getElementById('responseContent');
                ta.value = "你当前的浏览器支持WebSocket,请进行后续操作\r\n";
            };

            socket.onclose = function(event){
                var ta = document.getElementById('responseContent');
                ta.value = "";

                ta.value = "WebSocket连接已经关闭\r\n";
            };
        }else{
            alert("您的浏览器不支持WebSocket");
        }



        function send(message){
            if(!window.WebSocket){
                return;
            }
            if(socket.readyState == WebSocket.OPEN){
                socket.send(message);
            }else{
                alert("WebSocket连接没有建立成功！！");
            }
        }
    </script>
</head>
<body>
<form onSubmit="return false;">
    <input type = "text" name = "message" value = ""/>
    <br/><br/>
    <input type = "button" value = "发送WebSocket请求消息" onClick = "send(this.form.message.value)"/>
    <hr color="red"/>
    <h2>客户端接收到服务端返回的应答消息</h2>
    <textarea id = "responseContent" style = "width:1024px; height:300px"></textarea>
</form>
</body>
</html>

 src/main/java/org/springblade/modules/webscoket/service/IPushMsgService.java

New file
@@ -0,0 +1,35 @@
package org.springblade.modules.webscoket.service;

/**
 * @author 123456
 */
public interface IPushMsgService {

    /**
     * 给指定用户发送消息
     *
     * @param userId
     * @param msg
     */
    void pushMsg(String userId, String msg);

    /**
     * 给所有用户发送消息
     *
     * @param msg
     */
    void pushMsg(String msg);

    /**
     * 给指定用户发起视频邀请
     */
    int inviteVideoCall(String userid, String time, String type, String name,String faqiid);

    /**
     * 关闭视频请求
     *
     * @return
     */
    void closeVideoCall(String sentId, String acceptId);

}

 src/main/java/org/springblade/modules/webscoket/service/impl/PushMsgServiceImpl.java

New file
@@ -0,0 +1,65 @@
package org.springblade.modules.webscoket.service.impl;

import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springblade.modules.nettyServer.NettyConfig;
import org.springblade.modules.webscoket.service.IPushMsgService;
import org.springframework.stereotype.Service;

/**
 * @author lq
 * @date 2020/4/1 11:20
 */
@Service
public class PushMsgServiceImpl implements IPushMsgService {
    @Override
    public void pushMsg(String userId, String msg) {
        Channel channel = NettyConfig.getUserChannelMap().get(userId);
        if (channel != null) {
            channel.writeAndFlush(new TextWebSocketFrame(msg));
        }

    }

    @Override
    public void pushMsg(String msg) {
        ChannelGroup group = NettyConfig.getChannelGroup();
        String name = group.name();
        System.out.println("空间大小：" + group.size() + ",名字：" + name);
        group.writeAndFlush(new TextWebSocketFrame(msg));
    }

    @Override
    public int inviteVideoCall(String userId, String time, String type, String name,String faqiid) {
        //返回值
        int res = 0;
        Channel channel = NettyConfig.getUserChannelMap().get(userId);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("type", type);
        jsonObject.put("roomId", time);
        jsonObject.put("name", name);
        jsonObject.put("faqiid", faqiid);
        if (channel != null) {
            channel.writeAndFlush(new TextWebSocketFrame(String.valueOf(jsonObject)));
            res = 1;
        }
        return res;
    }

    @Override
    public void closeVideoCall(String sentId, String acceptId) {
        Channel channel1 = NettyConfig.getUserChannelMap().get(sentId);
        Channel channel2 = NettyConfig.getUserChannelMap().get(acceptId);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("type", "close");
        if (channel1 != null) {
            channel1.writeAndFlush(new TextWebSocketFrame(String.valueOf(jsonObject)));
        }
        if (channel2 != null) {
            channel2.writeAndFlush(new TextWebSocketFrame(String.valueOf(jsonObject)));
        }

    }
}