Spring3Websocket

package cn.com.mengdou.im; import static io.netty.handler.codec.http.HttpHeaders.Names.ACCESS_CONTROL_ALLOW_ORIGIN; import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH; import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE; import static io.netty.handler.codec.http.HttpHeaders.Names.HOST; import static io.netty.handler.codec.http.HttpMethod.GET; import static io.netty.handler.codec.http.HttpMethod.POST; import static io.netty.handler.codec.http.HttpResponseStatus.OK; import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.FullHttpResponse; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.QueryStringDecoder; import io.netty.handler.codec.http.multipart.Attribute; import io.netty.handler.codec.http.multipart.DefaultHttpDataFactory; import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder; import io.netty.handler.codec.http.multipart.InterfaceHttpData; import io.netty.handler.codec.http.multipart.InterfaceHttpData.HttpDataType; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker; import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory; import io.netty.util.CharsetUtil; import io.netty.util.concurrent.GlobalEventExecutor; import java.io.IOException; import java.net.InetSocketAddress; import java.util.HashMap; import java.util.List; import java.util.Map; /** * * @description:消息业务处理类 * @fileName:IMServerHandler.java * @createTime:2015年7月23日 下午7:38:22 * @author:肖震 * @version 1.0.0 * */ public class IMServerHandler extends SimpleChannelInboundHandler<Object> { // (1) public static ChannelGroup channels = new DefaultChannelGroup( GlobalEventExecutor.INSTANCE); private static final String WEBSOCKET_PATH = "/websocket"; // 在线用户状态map,由ip：在线状态组成 private static final Map<String, String> onlineMap = new HashMap<String, String>(); // 在线用户channel map,由ip：channel组成 private static final Map<String, Channel> onlineUserChannelMap = new HashMap<String, Channel>(); private WebSocketServerHandshaker handshaker; @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { // (2) Channel incoming = ctx.channel(); for (Channel channel : channels) { channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n"); } channels.add(ctx.channel()); } @Override public void channelReadComplete(ChannelHandlerContext ctx) { ctx.flush(); } @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { Channel incoming = ctx.channel(); for (Channel channel : channels) { channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 离开\n"); } channels.remove(ctx.channel()); } @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { // http请求 if (msg instanceof FullHttpRequest) { handleHttpRequest(ctx, (FullHttpRequest) msg); } else if (msg instanceof WebSocketFrame)// websocket请求 { handleWebSocketFrame(ctx, (WebSocketFrame) msg); } } /** * * @function: 获得本地webscoket地址 * @param req：http请求对象 * @return String ：返回地址字符串 * @exception * @author:肖震 * @since 1.0.0 */ private static String getWebSocketLocation(FullHttpRequest req) { String location = req.headers().get(HOST) + WEBSOCKET_PATH; return "ws://" + location; } /** * * @function: 处理websocket请求业务 * @param ctx * @param frame * @exception * @author:肖震 * @since 1.0.0 */ private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) { // 是否关闭 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)) { throw new UnsupportedOperationException(String.format( "%s frame types not supported", frame.getClass().getName())); } // 接收到的消息 String s = ((TextWebSocketFrame) frame).text(); // 发送者的通道 Channel incoming = ctx.channel(); System.out.println("收到客户[" + incoming.remoteAddress() + "]消息：" + s); // 从发送消息解析出接收人的用户id String acceptUserId = null; // 解析出接收者的用户id if (s != null && s.contains("||-||")) { acceptUserId = s.split("||-||")[1]; } // 点对点发送消息 if (acceptUserId != null) { Channel channel = onlineUserChannelMap.get(acceptUserId); if (channel != null) { channel.writeAndFlush(new TextWebSocketFrame("[" + channel.remoteAddress() + "]:" + s + "\n")); } } else // 群发消息 { for (Channel channel : channels) { if (channel != incoming) { channel.writeAndFlush(new TextWebSocketFrame("[" + channel.remoteAddress() + "]:" + s + "\n")); } else { channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.remoteAddress() + "]:" + s + "\n")); } } } } /** * * @function: http请求处理 * @param ctx * @param req * @throws Exception * @exception * @author:肖震 * @since 1.0.0 */ private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception { System.out.println("handleHttpRequest method==========" + req.getMethod()); System.out.println("handleHttpRequest uri==========" + req.getUri()); // http get请求 if (req.getMethod() == GET) { httpGetDeal(ctx, req); } if (req.getMethod() == POST) { httpPostDeal(ctx, req); return; } } /** * @function: http post方式处理 * @param ctx * @param req * @throws IOException * @exception * @author:肖震 * @since 1.0.0 */ private void httpPostDeal(ChannelHandlerContext ctx, FullHttpRequest req) throws IOException { // 处理POST请求 HttpPostRequestDecoder decoder = new HttpPostRequestDecoder( new DefaultHttpDataFactory(false), req); InterfaceHttpData postData = decoder.getBodyHttpData("q"); // // // 读取从客户端传过来的参数 String question = ""; if (postData.getHttpDataType() == HttpDataType.Attribute) { Attribute attribute = (Attribute) postData; question = attribute.getValue(); System.out.println("q:" + question); } if (question != null && !question.equals("")) { String data = question; FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(data.getBytes(CharsetUtil.UTF_8))); res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8"); res.headers().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*"); res.headers().set(CONTENT_LENGTH, res.content().readableBytes()); sendHttpResponse(ctx, req, res); } return; } /** * @function: http get�