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