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