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.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     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 ChannelHandlerContext
  67      * @param req FullHttpRequest
  68      */
  69     private void handleHttpRequest(final ChannelHandlerContext ctx, final FullHttpRequest req) {
  70         // Handle a bad request.
  71         if (!req.decoderResult().isSuccess()) {
  72             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
  73             return;
  74         }
  75
  76         // Allow only GET methods.
  77         if (req.method() != GET) {
  78             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
  79             return;
  80         }
  81
  82         final String streamName = Notificator.createStreamNameFromUri(req.uri());
  83         if (streamName.contains(RestconfImpl.DATA_SUBSCR)) {
  84             final ListenerAdapter listener = Notificator.getListenerFor(streamName);
  85             if (listener != null) {
  86                 listener.addSubscriber(ctx.channel());
  87                 LOG.debug("Subscriber successfully registered.");
  88             } else {
  89                 LOG.error("Listener for stream with name '{}' was not found.", streamName);
  90                 sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR));
  91             }
  92         } else if (streamName.contains(RestconfImpl.NOTIFICATION_STREAM)) {
  93             final List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
  94             if (listeners != null && !listeners.isEmpty()) {
  95                 for (final NotificationListenerAdapter listener : listeners) {
  96                     listener.addSubscriber(ctx.channel());
  97                     LOG.debug("Subscriber successfully registered.");
  98                 }
  99             } else {
 100                 LOG.error("Listener for stream with name '{}' was not found.", streamName);
 101                 sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR));
 102             }
 103         }
 104
 105         // Handshake
 106         final WebSocketServerHandshakerFactory wsFactory =
 107                 new WebSocketServerHandshakerFactory(getWebSocketLocation(req),
 108                 null, false);
 109         this.handshaker = wsFactory.newHandshaker(req);
 110         if (this.handshaker == null) {
 111             WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
 112         } else {
 113             this.handshaker.handshake(ctx.channel(), req);
 114         }
 115     }
 116
 117     /**
 118      * Checks response status, send response and close connection if necessary.
 119      *
 120      * @param ctx ChannelHandlerContext
 121      * @param req HttpRequest
 122      * @param res FullHttpResponse
 123      */
 124     private static void sendHttpResponse(final ChannelHandlerContext ctx, final HttpRequest req,
 125             final FullHttpResponse res) {
 126         // Generate an error page if response getStatus code is not OK (200).
 127         final boolean notOkay = !OK.equals(res.status());
 128         if (notOkay) {
 129             final ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
 130             res.content().writeBytes(buf);
 131             buf.release();
 132             setContentLength(res, res.content().readableBytes());
 133         }
 134
 135         // Send the response and close the connection if necessary.
 136         final ChannelFuture f = ctx.channel().writeAndFlush(res);
 137         if (notOkay || !isKeepAlive(req)) {
 138             f.addListener(ChannelFutureListener.CLOSE);
 139         }
 140     }
 141
 142     /**
 143      * Handles web socket frame.
 144      *
 145      * @param ctx {@link ChannelHandlerContext}
 146      * @param frame {@link WebSocketFrame}
 147      */
 148     private void handleWebSocketFrame(final ChannelHandlerContext ctx, final WebSocketFrame frame) {
 149         if (frame instanceof CloseWebSocketFrame) {
 150             this.handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
 151             final String streamName = Notificator.createStreamNameFromUri(((CloseWebSocketFrame) frame).reasonText());
 152             if (streamName.contains(RestconfImpl.DATA_SUBSCR)) {
 153                 final ListenerAdapter listener = Notificator.getListenerFor(streamName);
 154                 if (listener != null) {
 155                     listener.removeSubscriber(ctx.channel());
 156                     LOG.debug("Subscriber successfully registered.");
 157
 158                     Notificator.removeListenerIfNoSubscriberExists(listener);
 159                 }
 160             } else if (streamName.contains(RestconfImpl.NOTIFICATION_STREAM)) {
 161                 final List<NotificationListenerAdapter> listeners = Notificator.getNotificationListenerFor(streamName);
 162                 if (listeners != null && !listeners.isEmpty()) {
 163                     for (final NotificationListenerAdapter listener : listeners) {
 164                         listener.removeSubscriber(ctx.channel());
 165                     }
 166                 }
 167             }
 168             return;
 169         } else if (frame instanceof PingWebSocketFrame) {
 170             ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
 171             return;
 172         }
 173     }
 174
 175     @Override
 176     public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
 177         ctx.close();
 178     }
 179
 180     /**
 181      * Get web socket location from HTTP request.
 182      *
 183      * @param req HTTP request from which the location will be returned
 184      * @return String representation of web socket location.
 185      */
 186     private static String getWebSocketLocation(final HttpRequest req) {
 187         return "ws://" + req.headers().get(HOST) + req.uri();
 188     }
 189 }