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