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