opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/streams/websockets/WebSocketServerHandler.java

   1 package org.opendaylight.controller.sal.streams.websockets;
   2
   3 import static io.netty.handler.codec.http.HttpHeaders.isKeepAlive;
   4 import static io.netty.handler.codec.http.HttpHeaders.setContentLength;
   5 import static io.netty.handler.codec.http.HttpHeaders.Names.HOST;
   6 import static io.netty.handler.codec.http.HttpMethod.GET;
   7 import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
   8 import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
   9 import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
  10 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
  11 import io.netty.buffer.ByteBuf;
  12 import io.netty.buffer.Unpooled;
  13 import io.netty.channel.ChannelFuture;
  14 import io.netty.channel.ChannelFutureListener;
  15 import io.netty.channel.ChannelHandlerContext;
  16 import io.netty.channel.SimpleChannelInboundHandler;
  17 import io.netty.handler.codec.http.DefaultFullHttpResponse;
  18 import io.netty.handler.codec.http.FullHttpRequest;
  19 import io.netty.handler.codec.http.FullHttpResponse;
  20 import io.netty.handler.codec.http.HttpRequest;
  21 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
  22 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
  23 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
  24 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
  25 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
  26 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
  27 import io.netty.util.CharsetUtil;
  28
  29 import java.io.IOException;
  30
  31 import org.opendaylight.controller.sal.streams.listeners.ListenerAdapter;
  32 import org.opendaylight.controller.sal.streams.listeners.Notificator;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
  37
  38     private static final Logger logger = LoggerFactory.getLogger(WebSocketServerHandler.class);
  39
  40     private WebSocketServerHandshaker handshaker;
  41
  42     @Override
  43     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
  44         if (msg instanceof FullHttpRequest) {
  45             handleHttpRequest(ctx, (FullHttpRequest) msg);
  46         } else if (msg instanceof WebSocketFrame) {
  47             handleWebSocketFrame(ctx, (WebSocketFrame) msg);
  48         }
  49     }
  50
  51     private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req)
  52             throws Exception {
  53         // Handle a bad request.
  54         if (!req.getDecoderResult().isSuccess()) {
  55             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
  56             return;
  57         }
  58
  59         // Allow only GET methods.
  60         if (req.getMethod() != GET) {
  61             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
  62             return;
  63         }
  64
  65         String streamName = Notificator.createStreamNameFromUri(req.getUri());
  66         ListenerAdapter listener = Notificator.getListenerFor(streamName);
  67         if (listener != null) {
  68             listener.addSubscriber(ctx.channel());
  69             logger.debug("Subscriber successfully registered.");
  70         } else {
  71             logger.error("Listener for stream with name '{}' was not found.", streamName);
  72             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR));
  73         }
  74
  75         // Handshake
  76         WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
  77                 getWebSocketLocation(req), null, false);
  78         handshaker = wsFactory.newHandshaker(req);
  79         if (handshaker == null) {
  80             WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
  81         } else {
  82             handshaker.handshake(ctx.channel(), req);
  83         }
  84
  85     }
  86
  87     private static void sendHttpResponse(ChannelHandlerContext ctx,
  88             HttpRequest req, FullHttpResponse res) {
  89         // Generate an error page if response getStatus code is not OK (200).
  90         if (res.getStatus().code() != 200) {
  91             ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
  92             res.content().writeBytes(buf);
  93             buf.release();
  94             setContentLength(res, res.content().readableBytes());
  95         }
  96
  97         // Send the response and close the connection if necessary.
  98         ChannelFuture f = ctx.channel().writeAndFlush(res);
  99         if (!isKeepAlive(req) || res.getStatus().code() != 200) {
 100             f.addListener(ChannelFutureListener.CLOSE);
 101         }
 102     }
 103
 104     private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws IOException {
 105         if (frame instanceof CloseWebSocketFrame) {
 106             handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
 107             String streamName = Notificator.createStreamNameFromUri(((CloseWebSocketFrame) frame).reasonText());
 108             ListenerAdapter listener = Notificator.getListenerFor(streamName);
 109             if (listener != null) {
 110                 listener.removeSubscriber(ctx.channel());
 111                 logger.debug("Subscriber successfully registered.");
 112             }
 113             Notificator.removeListenerIfNoSubscriberExists(listener);
 114             return;
 115         } else if (frame instanceof PingWebSocketFrame) {
 116             ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
 117             return;
 118         }
 119     }
 120
 121     @Override
 122     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
 123             throws Exception {
 124         if (cause instanceof java.nio.channels.ClosedChannelException == false) {
 125             //cause.printStackTrace();
 126         }
 127         ctx.close();
 128     }
 129
 130     private static String getWebSocketLocation(HttpRequest req) {
 131         return "http://" + req.headers().get(HOST) + req.getUri();
 132     }
 133
 134 }