netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSession.java

   1 /*
   2  * Copyright (c) 2013 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.nettyutil;
   9
  10 import io.netty.channel.Channel;
  11 import io.netty.channel.ChannelFuture;
  12 import io.netty.channel.ChannelHandler;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelPromise;
  15 import io.netty.channel.SimpleChannelInboundHandler;
  16 import io.netty.handler.codec.ByteToMessageDecoder;
  17 import io.netty.handler.codec.MessageToByteEncoder;
  18 import java.io.EOFException;
  19 import org.opendaylight.netconf.api.NetconfExiSession;
  20 import org.opendaylight.netconf.api.NetconfMessage;
  21 import org.opendaylight.netconf.api.NetconfSession;
  22 import org.opendaylight.netconf.api.NetconfSessionListener;
  23 import org.opendaylight.netconf.api.NetconfTerminationReason;
  24 import org.opendaylight.netconf.api.xml.XmlElement;
  25 import org.opendaylight.netconf.nettyutil.handler.NetconfEXICodec;
  26 import org.opendaylight.netconf.nettyutil.handler.NetconfEXIToMessageDecoder;
  27 import org.opendaylight.netconf.nettyutil.handler.NetconfMessageToEXIEncoder;
  28 import org.opendaylight.netconf.nettyutil.handler.exi.EXIParameters;
  29 import org.opendaylight.netconf.shaded.exificient.core.exceptions.EXIException;
  30 import org.opendaylight.netconf.shaded.exificient.core.exceptions.UnsupportedOption;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 public abstract class AbstractNetconfSession<S extends NetconfSession, L extends NetconfSessionListener<S>>
  35         extends SimpleChannelInboundHandler<Object> implements NetconfSession, NetconfExiSession {
  36     private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfSession.class);
  37
  38
  39     private final L sessionListener;
  40     private final long sessionId;
  41     private boolean up = false;
  42
  43     private ChannelHandler delayedEncoder;
  44
  45     private final Channel channel;
  46
  47     protected AbstractNetconfSession(final L sessionListener, final Channel channel, final long sessionId) {
  48         this.sessionListener = sessionListener;
  49         this.channel = channel;
  50         this.sessionId = sessionId;
  51         LOG.debug("Session {} created", sessionId);
  52     }
  53
  54     protected abstract S thisInstance();
  55
  56     @Override
  57     public void close() {
  58         channel.close();
  59         up = false;
  60         sessionListener.onSessionTerminated(thisInstance(), new NetconfTerminationReason("Session closed"));
  61     }
  62
  63     protected void handleMessage(final NetconfMessage netconfMessage) {
  64         LOG.debug("handling incoming message");
  65         sessionListener.onMessage(thisInstance(), netconfMessage);
  66     }
  67
  68     @Override
  69     public ChannelFuture sendMessage(final NetconfMessage netconfMessage) {
  70         // From: https://github.com/netty/netty/issues/3887
  71         // Netty can provide "ordering" in the following situations:
  72         // 1. You are doing all writes from the EventLoop thread; OR
  73         // 2. You are doing no writes from the EventLoop thread (i.e. all writes are being done in other thread(s)).
  74         //
  75         // Restconf writes to a netconf mountpoint execute multiple messages
  76         // and one of these was executed from a restconf thread thus breaking ordering so
  77         // we need to execute all messages from an EventLoop thread.
  78
  79         final ChannelPromise promise = channel.newPromise();
  80         channel.eventLoop().execute(() -> {
  81             channel.writeAndFlush(netconfMessage, promise);
  82             if (delayedEncoder != null) {
  83                 replaceMessageEncoder(delayedEncoder);
  84                 delayedEncoder = null;
  85             }
  86         });
  87
  88         return promise;
  89     }
  90
  91     protected void endOfInput() {
  92         LOG.debug("Session {} end of input detected while session was in state {}", this, up ? "up" : "initialized");
  93         if (up) {
  94             this.sessionListener.onSessionDown(thisInstance(), new EOFException("End of input"));
  95         }
  96     }
  97
  98     protected void sessionUp() {
  99         LOG.debug("Session {} up", this);
 100         sessionListener.onSessionUp(thisInstance());
 101         this.up = true;
 102     }
 103
 104     @Override
 105     public String toString() {
 106         final StringBuilder sb = new StringBuilder(getClass().getSimpleName() + "{");
 107         sb.append("sessionId=").append(sessionId);
 108         sb.append(", channel=").append(channel);
 109         sb.append('}');
 110         return sb.toString();
 111     }
 112
 113     protected final void replaceMessageDecoder(final ChannelHandler handler) {
 114         replaceChannelHandler(AbstractChannelInitializer.NETCONF_MESSAGE_DECODER, handler);
 115     }
 116
 117     protected final void replaceMessageEncoder(final ChannelHandler handler) {
 118         replaceChannelHandler(AbstractChannelInitializer.NETCONF_MESSAGE_ENCODER, handler);
 119     }
 120
 121     protected final void replaceMessageEncoderAfterNextMessage(final ChannelHandler handler) {
 122         this.delayedEncoder = handler;
 123     }
 124
 125     protected final void replaceChannelHandler(final String handlerName, final ChannelHandler handler) {
 126         channel.pipeline().replace(handlerName, handlerName, handler);
 127     }
 128
 129     @Override
 130     public final void startExiCommunication(final NetconfMessage startExiMessage) {
 131         final EXIParameters exiParams;
 132         try {
 133             exiParams = EXIParameters.fromXmlElement(XmlElement.fromDomDocument(startExiMessage.getDocument()));
 134         } catch (final UnsupportedOption e) {
 135             LOG.warn("Unable to parse EXI parameters from {} on session {}", startExiMessage, this, e);
 136             throw new IllegalArgumentException("Cannot parse options", e);
 137         }
 138
 139         final NetconfEXICodec exiCodec = NetconfEXICodec.forParameters(exiParams);
 140         final NetconfMessageToEXIEncoder exiEncoder = NetconfMessageToEXIEncoder.create(exiCodec);
 141         final NetconfEXIToMessageDecoder exiDecoder;
 142         try {
 143             exiDecoder = NetconfEXIToMessageDecoder.create(exiCodec);
 144         } catch (EXIException e) {
 145             LOG.warn("Failed to instantiate EXI decodeer for {} on session {}", exiCodec, this, e);
 146             throw new IllegalStateException("Cannot instantiate encoder for options", e);
 147         }
 148
 149         addExiHandlers(exiDecoder, exiEncoder);
 150         LOG.debug("Session {} EXI handlers added to pipeline", this);
 151     }
 152
 153     /**
 154      * Add a set encoder/decoder tuple into the channel pipeline as appropriate.
 155      *
 156      * @param decoder EXI decoder
 157      * @param encoder EXI encoder
 158      */
 159     protected abstract void addExiHandlers(ByteToMessageDecoder decoder, MessageToByteEncoder<NetconfMessage> encoder);
 160
 161     public final boolean isUp() {
 162         return up;
 163     }
 164
 165     public final long getSessionId() {
 166         return sessionId;
 167     }
 168
 169     @Override
 170     @SuppressWarnings("checkstyle:illegalCatch")
 171     public final void channelInactive(final ChannelHandlerContext ctx) {
 172         LOG.debug("Channel {} inactive.", ctx.channel());
 173         endOfInput();
 174         try {
 175             // Forward channel inactive event, all handlers in pipeline might be interested in the event e.g. close
 176             // channel handler of reconnect promise
 177             super.channelInactive(ctx);
 178         } catch (final Exception e) {
 179             throw new RuntimeException("Failed to delegate channel inactive event on channel " + ctx.channel(), e);
 180         }
 181     }
 182
 183     @Override
 184     protected final void channelRead0(final ChannelHandlerContext ctx, final Object msg) {
 185         LOG.debug("Message was received: {}", msg);
 186         handleMessage((NetconfMessage) msg);
 187     }
 188
 189     @Override
 190     public final void handlerAdded(final ChannelHandlerContext ctx) {
 191         sessionUp();
 192     }
 193 }