Fun论设计模式之7：中介者模式（Mediator Pattern）

  1 package com.company.server;
  2 
  3 import io.netty.buffer.ByteBuf;
  4 import io.netty.buffer.Unpooled;
  5 import io.netty.channel.ChannelFuture;
  6 import io.netty.channel.ChannelFutureListener;
  7 import io.netty.channel.ChannelHandlerContext;
  8 import io.netty.channel.ChannelPromise;
  9 import io.netty.channel.SimpleChannelInboundHandler;
 10 import io.netty.handler.codec.http.DefaultFullHttpResponse;
 11 import io.netty.handler.codec.http.FullHttpRequest;
 12 import io.netty.handler.codec.http.FullHttpResponse;
 13 import io.netty.handler.codec.http.HttpHeaders;
 14 import io.netty.handler.codec.http.HttpResponseStatus;
 15 import io.netty.handler.codec.http.HttpVersion;
 16 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
 17 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
 18 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
 19 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 20 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 21 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
 22 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
 23 import io.netty.util.CharsetUtil;
 24 
 25 import org.apache.log4j.Logger;
 26 
 27 import com.company.serviceimpl.BananaService;
 28 import com.company.util.BufToString;
 29 import com.company.util.CODE;
 30 import com.company.util.Request;
 31 import com.company.util.Response;
 32 import com.google.common.base.Strings;
 33 import com.google.gson.JsonSyntaxException;
 34 
 35 
 36 /**
 37  * WebSocket服务端Handler
 38  *
 39  */
 40 public class BananaWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
 41     private static final Logger LOG = Logger.getLogger(BananaWebSocketServerHandler.class.getName());
 42     
 43     private WebSocketServerHandshaker handshaker;
 44     private ChannelHandlerContext ctx;
 45     private String sessionId;
 46     private boolean isLog = true;
 47     
 48     public BananaWebSocketServerHandler() {
 49         super();
 50     }
 51     
 52     public BananaWebSocketServerHandler(boolean isLog) {
 53         this();
 54         this.isLog = isLog;
 55     }
 56 
 57     //netty 5的覆写函数，netty4中用channelRead0代替
 58     public void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
 59         if(this.isLog) {
 60             System.out.print("channel MessageReceived = = " + ctx.name());
 61         }
 62         if (msg instanceof FullHttpRequest) { // 传统的HTTP接入
 63             FullHttpRequest mymsg = (FullHttpRequest) msg;
 64             System.out.println(" with http request : " + BufToString.convertByteBufToString(mymsg.content()));
 65             handleHttpRequest(ctx, mymsg);
 66         } else if (msg instanceof WebSocketFrame) { // WebSocket接入
 67             WebSocketFrame mymsg = (WebSocketFrame) msg;
 68             System.out.println(" with socket request : " + BufToString.convertByteBufToString(mymsg.content()));
 69             handleWebSocketFrame(ctx, mymsg);
 70         }
 71     }
 72     
 73     @Override
 74     public void handlerAdded(ChannelHandlerContext ctx) throws Exception{
 75         System.out.println("channel handlerAdded = = " + ctx.name());
 76     }
 77     
 78     @Override
 79     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception{
 80         System.out.println("channel handlerRemoved = = " + ctx.name());
 81     }
 82     
 83     @Override
 84     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
 85         if(this.isLog) {
 86             System.out.print("channel Read0 = = " + ctx.name());
 87         }
 88         if (msg instanceof FullHttpRequest) { // 传统的HTTP接入
 89             FullHttpRequest mymsg = (FullHttpRequest) msg;
 90             System.out.println(" with http request : " + BufToString.convertByteBufToString(mymsg.content()));
 91             handleHttpRequest(ctx, mymsg);
 92         } else if (msg instanceof WebSocketFrame) { // WebSocket接入
 93             WebSocketFrame mymsg = (WebSocketFrame) msg;
 94             System.out.println(" with socket request : " + BufToString.convertByteBufToString(mymsg.content()));
 95             handleWebSocketFrame(ctx, mymsg);
 96         }
 97     }
 98     
 99     @Override
100     public void channelInactive(ChannelHandlerContext ctx) {
101         if(this.isLog) {
102             System.out.println("channel Inactive = = " + ctx.name());
103         }
104         try {
105             this.close(ctx, null);
106         } catch (Exception e) {
107             e.printStackTrace();
108         }
109     }
110     
111     @Override
112     public void channelUnregistered(ChannelHandlerContext ctx) {
113         if(this.isLog) {
114             System.out.println("channel Unregistered = = " + ctx.name());
115         }
116     }
117 
118     @Override
119     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
120         ctx.flush();
121         System.out.println("channel Flush = = " + ctx.name());
122     }
123     
124     @Override
125     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
126 
127         ctx.close();
128         if(this.isLog) {
129             System.err.println("channel exceptionCaught = = " + ctx.name());
130             cause.printStackTrace();
131         }
132         BananaService.logout(sessionId); // 注销
133         BananaService.notifyDownline(sessionId); // 通知有人下线
134     }
135 
136     //netty 5的覆写函数，netty4中用channelInactive代替
137     public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
138         if(this.isLog) {
139             System.out.println("channel close = = " + ctx.name());
140         }
141         BananaService.logout(sessionId); // 注销
142         BananaService.notifyDownline(sessionId); // 通知有人下线
143         ctx.close();
144     }
145 
146     /**
147      * 处理Http请求，完成WebSocket握手<br/>
148      * 注意：WebSocket连接第一次请求使用的是Http
149      * @param ctx
150      * @param request
151      * @throws Exception
152      */
153     private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
154         // 如果HTTP解码失败，返回HHTP异常
155         if (!request.getDecoderResult().isSuccess() || (!"websocket".equals(request.headers().get("Upgrade")))) {
156             sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
157             return;
158         }
159 
160         // 正常WebSocket的Http连接请求，构造握手响应返回
161         WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://" + request.headers().get(HttpHeaders.Names.HOST), null, false);
162         handshaker = wsFactory.newHandshaker(request);
163         if (handshaker == null) { // 无法处理的websocket版本
164             WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
165         } else { // 向客户端发送websocket握手,完成握手
166             handshaker.handshake(ctx.channel(), request);
167             // 记录管道处理上下文，便于服务器推送数据到客户端
168             this.ctx = ctx;
169         }
170     }
171 
172     /**
173      * 处理Socket请求
174      * @param ctx
175      * @param frame
176      * @throws Exception 
177      */
178     private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
179         // 判断是否是关闭链路的指令
180         if (frame instanceof CloseWebSocketFrame) {
181             handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
182             return;
183         }
184         // 判断是否是Ping消息
185         if (frame instanceof PingWebSocketFrame) {
186             ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
187             return;
188         }
189         // 当前只支持文本消息，不支持二进制消息
190         if (!(frame instanceof TextWebSocketFrame)) {
191             throw new UnsupportedOperationException("当前只支持文本消息，不支持二进制消息");
192         }
193         
194         // 处理来自客户端的WebSocket请求
195         try {
196             /*
197             if(this.isLog) {
198                 System.out.println("handleWebSocketFrame-=-=-" + ((TextWebSocketFrame)frame).text());
199             }
200             */
201             Request request = Request.create(((TextWebSocketFrame)frame).text());
202             Response response = new Response();
203             response.setServiceId(request.getServiceId());
204             if (CODE.online.code.intValue() == request.getServiceId()) { // 客户端注册
205                 String requestId = request.getRequestId();
206                 if (Strings.isNullOrEmpty(requestId)) {
207                     response.setIsSucc(false).setMessage("requestId不能为空");
208                     return;
209                 } else if (Strings.isNullOrEmpty(request.getName())) {
210                     response.setIsSucc(false).setMessage("name不能为空");
211                     return;
212                 } else if (BananaService.bananaWatchMap.containsKey(requestId)) {
213                     response.setIsSucc(false).setMessage("您已经注册了，不能重复注册");
214                     return;
215                 }
216                 if (!BananaService.register(requestId, new BananaService(ctx, request.getName()))) {
217                     response.setIsSucc(false).setMessage("注册失败");
218                 } else {
219                     response.setIsSucc(true).setMessage("注册成功");
220                     
221                     BananaService.bananaWatchMap.forEach((reqId, callBack) -> {
222                         response.getHadOnline().put(reqId, ((BananaService)callBack).getName()); // 将已经上线的人员返回
223                         
224                         if (!reqId.equals(requestId)) {
225                             Request serviceRequest = new Request();
226                             serviceRequest.setServiceId(CODE.online.code);
227                             serviceRequest.setRequestId(requestId);
228                             serviceRequest.setName(request.getName());
229                             try {
230                                 callBack.send(serviceRequest); // 通知有人上线
231                             } catch (Exception e) {
232                                 LOG.warn("回调发送消息给客户端异常", e);
233                             }
234                         }
235                     });
236                 }
237                 sendWebSocket(response.toJson());
238                 this.sessionId = requestId; // 记录会话id，当页面刷新或浏览器关闭时，注销掉此链路
239             } else if (CODE.send_message.code.intValue() == request.getServiceId()) { // 客户端发送消息到聊天群
240                 String requestId = request.getRequestId();
241                 if (Strings.isNullOrEmpty(requestId)) {
242                     response.setIsSucc(false).setMessage("requestId不能为空");
243                 } else if (Strings.isNullOrEmpty(request.getName())) {
244                     response.setIsSucc(false).setMessage("name不能为空");
245                 } else if (Strings.isNullOrEmpty(request.getMessage())) {
246                     response.setIsSucc(false).setMessage("message不能为空");
247                 } else {
248                     response.setIsSucc(true).setMessage("发送消息成功");
249                     
250                     BananaService.bananaWatchMap.forEach((reqId, callBack) -> { // 将消息发送到所有机器
251                         Request serviceRequest = new Request();
252                         serviceRequest.setServiceId(CODE.receive_message.code);
253                         serviceRequest.setRequestId(requestId);
254                         serviceRequest.setName(request.getName());
255                         serviceRequest.setMessage(request.getMessage());
256                         try {
257                             callBack.send(serviceRequest);
258                         } catch (Exception e) {
259                             LOG.warn("回调发送消息给客户端异常", e);
260                         }
261                     });
262                 }
263                 sendWebSocket(response.toJson());
264             } else if (CODE.downline.code.intValue() == request.getServiceId()) { // 客户端下线
265                 String requestId = request.getRequestId();
266                 if (Strings.isNullOrEmpty(requestId)) {
267                     sendWebSocket(response.setIsSucc(false).setMessage("requestId不能为空").toJson());
268                 } else {
269                     BananaService.logout(requestId);
270                     response.setIsSucc(true).setMessage("下线成功");
271                     
272                     BananaService.notifyDownline(requestId); // 通知有人下线
273                     
274                     sendWebSocket(response.toJson());
275                 }
276                 
277             } else {
278                 sendWebSocket(response.setIsSucc(false).setMessage("未知请求").toJson());
279             }
280         } catch (JsonSyntaxException e1) {
281             LOG.warn("Json解析异常", e1);
282             System.err.println("Json解析异常");
283             e1.printStackTrace();
284         } catch (Exception e2) {
285             LOG.error("处理Socket请求异常", e2);
286             System.err.println("处理Socket请求异常");
287             e2.printStackTrace();
288         }
289     }
290 
291     /**
292      * Http返回
293      * @param ctx
294      * @param request
295      * @param response
296      */
297     private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, FullHttpResponse response) {
298         // 返回应答给客户端
299         if (response.getStatus().code() != 200) {
300             ByteBuf buf = Unpooled.copiedBuffer(response.getStatus().toString(), CharsetUtil.UTF_8);
301             response.content().writeBytes(buf);
302             buf.release();
303             HttpHeaders.setContentLength(response, response.content().readableBytes());
304         }
305 
306         // 如果是非Keep-Alive，关闭连接
307         ChannelFuture f = ctx.channel().writeAndFlush(response);
308         if (!HttpHeaders.isKeepAlive(request) || response.getStatus().code() != 200) {
309             f.addListener(ChannelFutureListener.CLOSE);
310         }
311     }
312     
313     /**
314      * WebSocket返回
315      * @param ctx
316      * @param req
317      * @param res
318      */
319     public void sendWebSocket(String msg) throws Exception {
320         if (this.handshaker == null || this.ctx == null || this.ctx.isRemoved()) {
321             throw new Exception("尚未握手成功，无法向客户端发送WebSocket消息");
322         }
323         this.ctx.channel().write(new TextWebSocketFrame(msg));
324         this.ctx.flush();
325     }
326 
327 }