restconf/restconf-nb-bierman02/src/main/java/org/opendaylight/netconf/sal/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.netconf.sal.streams.websockets;
  10
  11 import static io.netty.handler.codec.http.HttpHeaders.Names.HOST;
  12 import static io.netty.handler.codec.http.HttpHeaders.isKeepAlive;
  13 import static io.netty.handler.codec.http.HttpHeaders.setContentLength;
  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.HttpResponseStatus.INTERNAL_SERVER_ERROR;
  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.netconf.sal.restconf.impl.RestconfImpl;
  39 import org.opendaylight.netconf.sal.streams.listeners.ListenerAdapter;
  40 import org.opendaylight.netconf.sal.streams.listeners.NotificationListenerAdapter;
  41 import org.opendaylight.netconf.sal.streams.listeners.Notificator;
  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
  68      *            ChannelHandlerContext
  69      * @param req
  70      *            FullHttpRequest
  71      */
  72     private void handleHttpRequest(final ChannelHandlerContext ctx, final FullHttpRequest req) {
  73         // Handle a bad request.
  74         if (!req.getDecoderResult().isSuccess()) {
  75             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
  76             return;
  77         }
  78
  79         // Allow only GET methods.
  80         if (req.getMethod() != GET) {
  81             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
  82             return;
  83         }
  84
  85         final String streamName = Notificator.createStreamNameFromUri(req.getUri());
  86         if (streamName.contains(RestconfImpl.DATA_SUBSCR)) {
  87             final ListenerAdapter listener = Notificator.getListenerFor(streamName);
  88             if (listener != null) {
  89                 listener.addSubscriber(ctx.channel());
  90                 LOG.debug("Subscriber successfully registered.");
  91             } else {
  92                 LOG.error("Listener for stream with name '{}' was not found.", streamName);
  93                 sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR));
  94             }
  95         } else if (streamName.contains(RestconfImpl.NOTIFICATION_STREAM)) {
  96             final List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
  97             if (listeners != null && !listeners.isEmpty()) {
  98                 for (final NotificationListenerAdapter listener : listeners) {
  99                     listener.addSubscriber(ctx.channel());
 100                     LOG.debug("Subscriber successfully registered.");
 101                 }
 102             } else {
 103                 LOG.error("Listener for stream with name '{}' was not found.", streamName);
 104                 sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR));
 105             }
 106         }
 107
 108         // Handshake
 109         final WebSocketServerHandshakerFactory wsFactory =
 110                 new WebSocketServerHandshakerFactory(getWebSocketLocation(req),
 111                 null, false);
 112         this.handshaker = wsFactory.newHandshaker(req);
 113         if (this.handshaker == null) {
 114             WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
 115         } else {
 116             this.handshaker.handshake(ctx.channel(), req);
 117         }
 118
 119     }
 120
 121     /**
 122      * Checks response status, send response and close connection if necessary.
 123      *
 124      * @param ctx
 125      *            ChannelHandlerContext
 126      * @param req
 127      *            HttpRequest
 128      * @param res
 129      *            FullHttpResponse
 130      */
 131     private static void sendHttpResponse(final ChannelHandlerContext ctx, final HttpRequest req,
 132             final FullHttpResponse res) {
 133         // Generate an error page if response getStatus code is not OK (200).
 134         if (res.getStatus().code() != 200) {
 135             final ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
 136             res.content().writeBytes(buf);
 137             buf.release();
 138             setContentLength(res, res.content().readableBytes());
 139         }
 140
 141         // Send the response and close the connection if necessary.
 142         final ChannelFuture f = ctx.channel().writeAndFlush(res);
 143         if (!isKeepAlive(req) || res.getStatus().code() != 200) {
 144             f.addListener(ChannelFutureListener.CLOSE);
 145         }
 146     }
 147
 148     /**
 149      * Handles web socket frame.
 150      *
 151      * @param ctx
 152      *            {@link ChannelHandlerContext}
 153      * @param frame
 154      *            {@link WebSocketFrame}
 155      */
 156     private void handleWebSocketFrame(final ChannelHandlerContext ctx, final WebSocketFrame frame) {
 157         if (frame instanceof CloseWebSocketFrame) {
 158             this.handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
 159             final String streamName = Notificator.createStreamNameFromUri(((CloseWebSocketFrame) frame).reasonText());
 160             if (streamName.contains(RestconfImpl.DATA_SUBSCR)) {
 161                 final ListenerAdapter listener = Notificator.getListenerFor(streamName);
 162                 if (listener != null) {
 163                     listener.removeSubscriber(ctx.channel());
 164                     LOG.debug("Subscriber successfully registered.");
 165
 166                     Notificator.removeListenerIfNoSubscriberExists(listener);
 167                 }
 168             } else if (streamName.contains(RestconfImpl.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) {
 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(HOST) + req.getUri();
 197     }
 198
 199 }