restconf/restconf-nb-rfc8040/src/main/java/org/opendaylight/restconf/nb/rfc8040/streams/websockets/WebSocketServerHandler.java

   1 /*
   2  * Copyright (c) 2014, 2015 Cisco Systems, Inc. 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.restconf.nb.rfc8040.streams.websockets;
  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.HttpMethod.GET;
  14 import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
  15 import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
  16 import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
  17 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
  18
  19 import io.netty.buffer.ByteBuf;
  20 import io.netty.buffer.Unpooled;
  21 import io.netty.channel.ChannelFuture;
  22 import io.netty.channel.ChannelFutureListener;
  23 import io.netty.channel.ChannelHandlerContext;
  24 import io.netty.channel.SimpleChannelInboundHandler;
  25 import io.netty.handler.codec.http.DefaultFullHttpResponse;
  26 import io.netty.handler.codec.http.FullHttpRequest;
  27 import io.netty.handler.codec.http.FullHttpResponse;
  28 import io.netty.handler.codec.http.HttpHeaders.Names;
  29 import io.netty.handler.codec.http.HttpRequest;
  30 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
  31 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
  32 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
  33 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
  34 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
  35 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
  36 import io.netty.util.CharsetUtil;
  37 import java.io.IOException;
  38 import java.util.List;
  39 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenerAdapter;
  40 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.NotificationListenerAdapter;
  41 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.Notificator;
  42 import org.opendaylight.restconf.nb.rfc8040.utils.RestconfConstants;
  43 import org.slf4j.Logger;
  44 import org.slf4j.LoggerFactory;
  45
  46 /**
  47  * {@link WebSocketServerHandler} is implementation of {@link SimpleChannelInboundHandler} which allow handle
  48  * {@link FullHttpRequest} and {@link WebSocketFrame} messages.
  49  */
  50 public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
  51
  52     private static final Logger LOG = LoggerFactory.getLogger(WebSocketServerHandler.class);
  53
  54     private WebSocketServerHandshaker handshaker;
  55
  56     @Override
  57     protected void channelRead0(final ChannelHandlerContext ctx, final Object msg) throws Exception {
  58         if (msg instanceof FullHttpRequest) {
  59             handleHttpRequest(ctx, (FullHttpRequest) msg);
  60         } else if (msg instanceof WebSocketFrame) {
  61             handleWebSocketFrame(ctx, (WebSocketFrame) msg);
  62         }
  63     }
  64
  65     /**
  66      * Checks if HTTP request method is GET and if is possible to decode HTTP result of request.
  67      *
  68      * @param ctx
  69      *            ChannelHandlerContext
  70      * @param req
  71      *            FullHttpRequest
  72      */
  73     private void handleHttpRequest(final ChannelHandlerContext ctx, final FullHttpRequest req) throws Exception {
  74         // Handle a bad request.
  75         if (!req.getDecoderResult().isSuccess()) {
  76             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
  77             return;
  78         }
  79
  80         // Allow only GET methods.
  81         if (req.getMethod() != GET) {
  82             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
  83             return;
  84         }
  85
  86         final String streamName = Notificator.createStreamNameFromUri(req.getUri());
  87         if (streamName.contains(RestconfConstants.DATA_SUBSCR)) {
  88             final ListenerAdapter listener = Notificator.getListenerFor(streamName);
  89             if (listener != null) {
  90                 listener.addSubscriber(ctx.channel());
  91                 LOG.debug("Subscriber successfully registered.");
  92             } else {
  93                 LOG.error("Listener for stream with name '{}' was not found.", streamName);
  94                 sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR));
  95             }
  96         } else if (streamName.contains(RestconfConstants.NOTIFICATION_STREAM)) {
  97             final List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
  98             if (listeners != null && !listeners.isEmpty()) {
  99                 for (final NotificationListenerAdapter listener : listeners) {
 100                     listener.addSubscriber(ctx.channel());
 101                     LOG.debug("Subscriber successfully registered.");
 102                 }
 103             } else {
 104                 LOG.error("Listener for stream with name '{}' was not found.", streamName);
 105                 sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR));
 106             }
 107         }
 108
 109         // Handshake
 110         final WebSocketServerHandshakerFactory wsFactory =
 111                 new WebSocketServerHandshakerFactory(getWebSocketLocation(req),
 112                 null, false);
 113         this.handshaker = wsFactory.newHandshaker(req);
 114         if (this.handshaker == null) {
 115             WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
 116         } else {
 117             this.handshaker.handshake(ctx.channel(), req);
 118         }
 119
 120     }
 121
 122     /**
 123      * Checks response status, send response and close connection if necessary.
 124      *
 125      * @param ctx
 126      *            ChannelHandlerContext
 127      * @param req
 128      *            HttpRequest
 129      * @param res
 130      *            FullHttpResponse
 131      */
 132     private static void sendHttpResponse(final ChannelHandlerContext ctx, final HttpRequest req,
 133             final FullHttpResponse res) {
 134         // Generate an error page if response getStatus code is not OK (200).
 135         if (res.getStatus().code() != 200) {
 136             final ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
 137             res.content().writeBytes(buf);
 138             buf.release();
 139             setContentLength(res, res.content().readableBytes());
 140         }
 141
 142         // Send the response and close the connection if necessary.
 143         final ChannelFuture f = ctx.channel().writeAndFlush(res);
 144         if (!isKeepAlive(req) || res.getStatus().code() != 200) {
 145             f.addListener(ChannelFutureListener.CLOSE);
 146         }
 147     }
 148
 149     /**
 150      * Handles web socket frame.
 151      *
 152      * @param ctx
 153      *            {@link ChannelHandlerContext}
 154      * @param frame
 155      *            {@link WebSocketFrame}
 156      */
 157     private void handleWebSocketFrame(final ChannelHandlerContext ctx, final WebSocketFrame frame) throws IOException {
 158         if (frame instanceof CloseWebSocketFrame) {
 159             this.handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
 160             final String streamName = Notificator.createStreamNameFromUri(((CloseWebSocketFrame) frame).reasonText());
 161             if (streamName.contains(RestconfConstants.DATA_SUBSCR)) {
 162                 final ListenerAdapter listener = Notificator.getListenerFor(streamName);
 163                 if (listener != null) {
 164                     listener.removeSubscriber(ctx.channel());
 165                     LOG.debug("Subscriber successfully registered.");
 166                     Notificator.removeListenerIfNoSubscriberExists(listener);
 167                 }
 168             } else if (streamName.contains(RestconfConstants.NOTIFICATION_STREAM)) {
 169                 final List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
 170                 if (listeners != null && !listeners.isEmpty()) {
 171                     for (final NotificationListenerAdapter listener : listeners) {
 172                         listener.removeSubscriber(ctx.channel());
 173                     }
 174                 }
 175             }
 176             return;
 177         } else if (frame instanceof PingWebSocketFrame) {
 178             ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
 179             return;
 180         }
 181     }
 182
 183     @Override
 184     public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {
 185         ctx.close();
 186     }
 187
 188     /**
 189      * Get web socket location from HTTP request.
 190      *
 191      * @param req
 192      *            HTTP request from which the location will be returned
 193      * @return String representation of web socket location.
 194      */
 195     private static String getWebSocketLocation(final HttpRequest req) {
 196         return "ws://" + req.headers().get(Names.HOST) + req.getUri();
 197     }
 198
 199 }