netconf/callhome-protocol/src/main/java/org/opendaylight/netconf/callhome/protocol/MinaSshNettyChannel.java

   1 /*
   2  * Copyright (c) 2016 Brocade Communication Systems 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
   9 package org.opendaylight.netconf.callhome.protocol;
  10
  11 import com.google.common.base.Preconditions;
  12 import io.netty.buffer.ByteBuf;
  13 import io.netty.channel.AbstractServerChannel;
  14 import io.netty.channel.ChannelConfig;
  15 import io.netty.channel.ChannelHandlerContext;
  16 import io.netty.channel.ChannelMetadata;
  17 import io.netty.channel.ChannelOutboundBuffer;
  18 import io.netty.channel.ChannelOutboundHandlerAdapter;
  19 import io.netty.channel.ChannelPromise;
  20 import io.netty.channel.DefaultChannelConfig;
  21 import io.netty.channel.EventLoop;
  22 import java.net.SocketAddress;
  23 import org.apache.sshd.ClientChannel;
  24 import org.apache.sshd.ClientSession;
  25 import org.opendaylight.netconf.nettyutil.handler.ssh.client.AsyncSshHandlerReader;
  26 import org.opendaylight.netconf.nettyutil.handler.ssh.client.AsyncSshHandlerReader.ReadMsgHandler;
  27 import org.opendaylight.netconf.nettyutil.handler.ssh.client.AsyncSshHandlerWriter;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 class MinaSshNettyChannel extends AbstractServerChannel {
  32
  33     private static final Logger LOG = LoggerFactory.getLogger(MinaSshNettyChannel.class);
  34     private static final ChannelMetadata METADATA = new ChannelMetadata(false);
  35     private final ChannelConfig config = new DefaultChannelConfig(this);
  36     private final CallHomeSessionContext context;
  37     private final ClientSession session;
  38     private final ClientChannel sshChannel;
  39     private final AsyncSshHandlerReader  sshReadHandler;
  40     private final AsyncSshHandlerWriter sshWriteAsyncHandler;
  41
  42
  43     private volatile boolean nettyClosed = false;
  44
  45     MinaSshNettyChannel(CallHomeSessionContext context, ClientSession session, ClientChannel sshChannel) {
  46         this.context = Preconditions.checkNotNull(context);
  47         this.session = Preconditions.checkNotNull(session);
  48         this.sshChannel = Preconditions.checkNotNull(sshChannel);
  49         this.sshReadHandler = new AsyncSshHandlerReader(new ConnectionClosedDuringRead(), new FireReadMessage(), "netconf",
  50                 sshChannel.getAsyncOut());
  51         this.sshWriteAsyncHandler = new AsyncSshHandlerWriter(sshChannel.getAsyncIn());
  52         pipeline().addFirst(createChannelAdapter());
  53     }
  54
  55     private ChannelOutboundHandlerAdapter createChannelAdapter() {
  56         return new ChannelOutboundHandlerAdapter() {
  57
  58             @Override
  59             public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
  60                 sshWriteAsyncHandler.write(ctx, msg, promise);
  61             }
  62
  63         };
  64     }
  65
  66     @Override
  67     public ChannelConfig config() {
  68         return config;
  69     }
  70
  71     private boolean notClosing(org.apache.sshd.common.Closeable sshCloseable) {
  72         return !sshCloseable.isClosing() && !sshCloseable.isClosed();
  73     }
  74
  75
  76     @Override
  77     public boolean isOpen() {
  78         return notClosing(session);
  79     }
  80
  81     @Override
  82     public boolean isActive() {
  83         return notClosing(session);
  84     }
  85
  86     @Override
  87     public ChannelMetadata metadata() {
  88         return METADATA;
  89     }
  90
  91     @Override
  92     protected AbstractUnsafe newUnsafe() {
  93        return new SshUnsafe();
  94     }
  95
  96     @Override
  97     protected boolean isCompatible(EventLoop loop) {
  98         return true;
  99     }
 100
 101     @Override
 102     protected SocketAddress localAddress0() {
 103         return session.getIoSession().getLocalAddress();
 104     }
 105
 106     @Override
 107     protected SocketAddress remoteAddress0() {
 108         return context.getRemoteAddress();
 109     }
 110
 111     @Override
 112     protected void doBind(SocketAddress localAddress) throws Exception {
 113         throw new UnsupportedOperationException("Bind not supported.");
 114     }
 115
 116     void doMinaDisconnect(boolean blocking) {
 117         if(notClosing(session)) {
 118             sshChannel.close(blocking);
 119             session.close(blocking);
 120         }
 121     }
 122
 123     void doNettyDisconnect() {
 124         if(! nettyClosed) {
 125             nettyClosed = true;
 126             pipeline().fireChannelInactive();
 127             sshReadHandler.close();
 128             sshWriteAsyncHandler.close();
 129         }
 130     }
 131
 132     @Override
 133     protected void doDisconnect() throws Exception {
 134         LOG.info("Disconnect invoked");
 135         doNettyDisconnect();
 136         doMinaDisconnect(false);
 137     }
 138
 139     @Override
 140     protected void doClose() throws Exception {
 141         context.removeSelf();
 142         if(notClosing(session)) {
 143             session.close(true);
 144             sshChannel.close(true);
 145         }
 146     }
 147
 148     @Override
 149     protected void doBeginRead() throws Exception {
 150         // Intentional NOOP - read is started by AsyncSshHandlerReader
 151     }
 152
 153     @Override
 154     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
 155         throw new IllegalStateException("Outbound writes to SSH should be done by SSH Write handler");
 156     }
 157
 158     private final class FireReadMessage implements ReadMsgHandler {
 159
 160         @Override
 161         public void onMessageRead(ByteBuf msg) {
 162             pipeline().fireChannelRead(msg);
 163         }
 164
 165     }
 166
 167     private final class ConnectionClosedDuringRead implements AutoCloseable {
 168
 169         /**
 170          * Invoked when SSH session dropped during read using {@link AsyncSshHandlerReader}
 171          */
 172         @Override
 173         public void close() throws Exception {
 174             doNettyDisconnect();
 175         }
 176
 177     }
 178
 179     private class SshUnsafe extends AbstractUnsafe {
 180
 181         @Override
 182         public void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
 183             throw new UnsupportedOperationException("Unsafe is not supported.");
 184         }
 185
 186     }
 187 }