2 * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.netconf.nettyutil;
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;
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);
39 private final L sessionListener;
40 private final long sessionId;
41 private boolean up = false;
43 private ChannelHandler delayedEncoder;
45 private final Channel channel;
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);
54 protected abstract S thisInstance();
60 sessionListener.onSessionTerminated(thisInstance(), new NetconfTerminationReason("Session closed"));
63 protected void handleMessage(final NetconfMessage netconfMessage) {
64 LOG.debug("handling incoming message");
65 sessionListener.onMessage(thisInstance(), netconfMessage);
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)).
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.
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;
91 protected void endOfInput() {
92 LOG.debug("Session {} end of input detected while session was in state {}", this, up ? "up" : "initialized");
94 this.sessionListener.onSessionDown(thisInstance(), new EOFException("End of input"));
98 protected void sessionUp() {
99 LOG.debug("Session {} up", this);
100 sessionListener.onSessionUp(thisInstance());
105 public String toString() {
106 final StringBuilder sb = new StringBuilder(getClass().getSimpleName() + "{");
107 sb.append("sessionId=").append(sessionId);
108 sb.append(", channel=").append(channel);
110 return sb.toString();
113 protected final void replaceMessageDecoder(final ChannelHandler handler) {
114 replaceChannelHandler(AbstractChannelInitializer.NETCONF_MESSAGE_DECODER, handler);
117 protected final void replaceMessageEncoder(final ChannelHandler handler) {
118 replaceChannelHandler(AbstractChannelInitializer.NETCONF_MESSAGE_ENCODER, handler);
121 protected final void replaceMessageEncoderAfterNextMessage(final ChannelHandler handler) {
122 this.delayedEncoder = handler;
125 protected final void replaceChannelHandler(final String handlerName, final ChannelHandler handler) {
126 channel.pipeline().replace(handlerName, handlerName, handler);
130 public final void startExiCommunication(final NetconfMessage startExiMessage) {
131 final EXIParameters exiParams;
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);
139 final NetconfEXICodec exiCodec = NetconfEXICodec.forParameters(exiParams);
140 final NetconfMessageToEXIEncoder exiEncoder = NetconfMessageToEXIEncoder.create(exiCodec);
141 final NetconfEXIToMessageDecoder exiDecoder;
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);
149 addExiHandlers(exiDecoder, exiEncoder);
150 LOG.debug("Session {} EXI handlers added to pipeline", this);
154 * Add a set encoder/decoder tuple into the channel pipeline as appropriate.
156 * @param decoder EXI decoder
157 * @param encoder EXI encoder
159 protected abstract void addExiHandlers(ByteToMessageDecoder decoder, MessageToByteEncoder<NetconfMessage> encoder);
161 public final boolean isUp() {
165 public final long getSessionId() {
170 @SuppressWarnings("checkstyle:illegalCatch")
171 public final void channelInactive(final ChannelHandlerContext ctx) {
172 LOG.debug("Channel {} inactive.", ctx.channel());
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);
184 protected final void channelRead0(final ChannelHandlerContext ctx, final Object msg) {
185 LOG.debug("Message was received: {}", msg);
186 handleMessage((NetconfMessage) msg);
190 public final void handlerAdded(final ChannelHandlerContext ctx) {