sdnr/wireless-transport/code-Carbon-SR1/apps/websocketmanager/impl/src/main/java/org/opendaylight/mwtn/impl/websocket/WebSocketServerHandler.java

   1 /*
   2 * Copyright (c) 2016 Wipro Ltd. and others. All rights reserved.
   3 *
   4 * This program and the accompanying materials are made available under the
   5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6 * and is available at http://www.eclipse.org/legal/epl-v10.html
   7 */
   8
   9 package org.opendaylight.mwtn.impl.websocket;
  10
  11 import static io.netty.handler.codec.http.HttpHeaders.isKeepAlive;
  12 import static io.netty.handler.codec.http.HttpHeaders.setContentLength;
  13 import static io.netty.handler.codec.http.HttpHeaders.Names.HOST;
  14 import static io.netty.handler.codec.http.HttpMethod.GET;
  15 import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
  16 import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
  17 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
  18 import static org.opendaylight.mwtn.impl.utils.Utils.hmChannelContexts;
  19 import static org.opendaylight.mwtn.impl.utils.Utils.hmClientScopes;
  20
  21 import java.util.Map;
  22
  23 import org.json.JSONObject;
  24 import org.opendaylight.mwtn.impl.dto.UserDto;
  25 import org.opendaylight.mwtn.impl.utils.Utils;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 import io.netty.buffer.ByteBuf;
  30 import io.netty.buffer.Unpooled;
  31 import io.netty.channel.ChannelFuture;
  32 import io.netty.channel.ChannelFutureListener;
  33 import io.netty.channel.ChannelHandlerContext;
  34 import io.netty.channel.SimpleChannelInboundHandler;
  35 import io.netty.handler.codec.http.DefaultFullHttpResponse;
  36 import io.netty.handler.codec.http.FullHttpRequest;
  37 import io.netty.handler.codec.http.FullHttpResponse;
  38 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
  39 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
  40 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
  41 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  42 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
  43 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
  44 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
  45 import io.netty.util.CharsetUtil;
  46
  47 public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
  48         private static final Logger LOG = LoggerFactory.getLogger(WebSocketServerHandler.class.getName());
  49
  50         private static final String WEBSOCKET_PATH = "/websocket";
  51
  52         public static final String KEY_NODENAME = "nodename";
  53         public static final String KEY_EVENTTYPE = "eventtype";
  54         public static final String KEY_XMLEVENT = "xmlevent";
  55
  56         private WebSocketServerHandshaker handshaker;
  57
  58         @Override
  59         public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  60                 super.channelRegistered(ctx);
  61                 hmChannelContexts.put(String.valueOf(ctx.hashCode()), ctx);
  62                 LOG.info("Channel registered " + ctx.channel().toString());
  63         }
  64
  65         @Override
  66         public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
  67                 super.channelUnregistered(ctx);
  68                 if (ctx != null) {
  69                         String channelHashCode = String.valueOf(ctx.hashCode());
  70                         hmChannelContexts.remove(channelHashCode);
  71                         LOG.info("Channel unregistered " + ctx.channel().toString());
  72                 }
  73         }
  74
  75         @Override
  76         public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
  77                 if (msg instanceof FullHttpRequest) {
  78                         handleHttpRequest(ctx, (FullHttpRequest) msg);
  79                 } else if (msg instanceof WebSocketFrame) {
  80                         handleWebSocketFrame(ctx, (WebSocketFrame) msg);
  81                 }
  82         }
  83
  84         public static void sendMessage(String nodeName, String eventType, String xmlEvent) {
  85                 if (hmChannelContexts != null && hmChannelContexts.size() > 0) {
  86                         for (Map.Entry<String, ChannelHandlerContext> entry : hmChannelContexts.entrySet()) {
  87                                 ChannelHandlerContext ctx = entry.getValue();
  88
  89                                 try {
  90                                         UserDto clientDto = hmClientScopes.get(String.valueOf(ctx.hashCode()));
  91                                         if (clientDto.getScopes().get(eventType) != null) {
  92                                                 sendBroadcast(ctx, xmlEvent);
  93                                         }
  94                                 } catch (Exception ioe) {
  95                                         LOG.warn(ioe.getMessage());
  96                                 }
  97                         }
  98                 }
  99         }
 100         public static void sendMessage(String xmlEvent) {
 101                 if (hmChannelContexts != null && hmChannelContexts.size() > 0) {
 102                         for (Map.Entry<String, ChannelHandlerContext> entry : hmChannelContexts.entrySet()) {
 103                                 ChannelHandlerContext ctx = entry.getValue();
 104
 105                                 try {
 106                                         sendBroadcast(ctx, xmlEvent);
 107
 108                                 } catch (Exception ioe) {
 109                                         LOG.warn(ioe.getMessage());
 110                                 }
 111                         }
 112                 }
 113         }
 114         private static void sendBroadcast(ChannelHandlerContext ctx, String message) {
 115                 try {
 116                         ctx.channel().write(new TextWebSocketFrame(message));
 117                 } catch (Exception e) {
 118                         LOG.warn(e.getMessage());
 119                 }
 120                 ctx.channel().flush();
 121         }
 122
 123         @Override
 124         public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
 125                 ctx.flush();
 126         }
 127
 128         private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
 129                 // Handle a bad request.
 130                 if (!req.getDecoderResult().isSuccess()) {
 131                         sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
 132                         return;
 133                 }
 134
 135                 // Allow only GET methods.
 136                 if (req.getMethod() != GET) {
 137                         sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
 138                         return;
 139                 }
 140
 141                 // Handshake
 142                 WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(getWebSocketLocation(req),
 143                                 null, false);
 144                 handshaker = wsFactory.newHandshaker(req);
 145                 if (handshaker == null) {
 146                         WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
 147                 } else {
 148                         handshaker.handshake(ctx.channel(), req);
 149                 }
 150         }
 151
 152         private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
 153
 154                 // Check for closing frame
 155                 if (frame instanceof CloseWebSocketFrame) {
 156                         handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
 157                         return;
 158                 }
 159                 if (frame instanceof PingWebSocketFrame) {
 160                         ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
 161                         return;
 162                 }
 163                 if (!(frame instanceof TextWebSocketFrame)) {
 164                         throw new UnsupportedOperationException(
 165                                         String.format("%s frame types not supported", frame.getClass().getName()));
 166                 }
 167
 168                 String request = ((TextWebSocketFrame) frame).text();
 169                 LOG.info(String.format("%s received %s", ctx.channel(), request));
 170                 if(!manageClientRequest(ctx, request));
 171                         manageClientRequest2(ctx, request);
 172                 }
 173
 174         /*
 175          * broadcast message to all your clients
 176          */
 177         private void manageClientRequest2(ChannelHandlerContext ctx, String request) {
 178                 try {
 179                         JSONObject o=new JSONObject(request);
 180                         if(o.has(KEY_NODENAME) && o.has(KEY_EVENTTYPE) && o.has(KEY_EVENTTYPE))
 181                                 sendMessage(o.getString(KEY_NODENAME),o.getString(KEY_EVENTTYPE),o.getString(KEY_XMLEVENT));
 182                 } catch (Exception e) {
 183                         LOG.warn("handle ws request failed:"+e.getMessage());
 184                 }
 185         }
 186         private boolean manageClientRequest(ChannelHandlerContext ctx, String request) {
 187                 boolean ret=false;
 188                 try {
 189                         JSONObject jsonMessage = new JSONObject(request);
 190                         if(jsonMessage.has(Utils.MSG_KEY_DATA))
 191                         {
 192                                 String data = jsonMessage.getString(Utils.MSG_KEY_DATA);
 193                                 if (data.equals(Utils.MSG_KEY_SCOPES)) {
 194                                         ret=true;
 195                                         String sessionId = String.valueOf(ctx.hashCode());
 196                                         UserDto clientDto = new UserDto();
 197                                         clientDto.setScopes(jsonMessage.getJSONArray(Utils.MSG_KEY_SCOPES));
 198                                         clientDto.setUserId(sessionId);
 199                                         hmClientScopes.put(sessionId, clientDto);
 200                                         ctx.channel().write(new TextWebSocketFrame(
 201                                                         "You are connected to the Opendaylight Websocket server and scopes are : " + request + ""));
 202                                 }
 203                         }
 204                 } catch (Exception e) {
 205                         LOG.warn("problem set scope: "+e.getMessage());
 206                         ctx.channel().write(new TextWebSocketFrame("Your request to the Opendaylight Websocket server is >> "
 207                                         + request + " << which failed because of following exception >> " + e.toString()));
 208                 }
 209                 return ret;
 210         }
 211
 212         private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
 213                 // Generate an error page if response getStatus code is not OK (200).
 214                 if (res.getStatus().code() != 200) {
 215                         ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
 216                         res.content().writeBytes(buf);
 217                         buf.release();
 218                         setContentLength(res, res.content().readableBytes());
 219                 }
 220
 221                 // Send the response and close the connection if necessary.
 222                 ChannelFuture f = ctx.channel().writeAndFlush(res);
 223                 if (!isKeepAlive(req) || res.getStatus().code() != 200) {
 224                         f.addListener(ChannelFutureListener.CLOSE);
 225                 }
 226         }
 227
 228         @Override
 229         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 230                 LOG.info("Exception caught {}");
 231                 ctx.close();
 232         }
 233
 234         private static String getWebSocketLocation(FullHttpRequest req) {
 235                 return "ws://" + req.headers().get(HOST) + WEBSOCKET_PATH;
 236         }
 237
 238 }