opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerWriter.java

   1 /*
   2  * Copyright (c) 2014 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
   9 package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client;
  10
  11 import com.google.common.base.Charsets;
  12 import com.google.common.base.Preconditions;
  13 import io.netty.buffer.ByteBuf;
  14 import io.netty.channel.ChannelHandlerContext;
  15 import io.netty.channel.ChannelPromise;
  16 import java.util.Deque;
  17 import java.util.LinkedList;
  18 import java.util.Queue;
  19 import org.apache.sshd.common.future.SshFutureListener;
  20 import org.apache.sshd.common.io.IoOutputStream;
  21 import org.apache.sshd.common.io.IoWriteFuture;
  22 import org.apache.sshd.common.io.WritePendingException;
  23 import org.apache.sshd.common.util.Buffer;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 /**
  28  * Async Ssh writer. Takes messages(byte arrays) and sends them encrypted to remote server.
  29  * Also handles pending writes by caching requests until pending state is over.
  30  */
  31 final class AsyncSshHandlerWriter implements AutoCloseable {
  32
  33     private static final Logger logger = LoggerFactory
  34             .getLogger(AsyncSshHandlerWriter.class);
  35
  36     // public static final int MAX_PENDING_WRITES = 1000;
  37     // TODO implement Limiting mechanism for pending writes
  38     // But there is a possible issue with limiting:
  39     // 1. What to do when queue is full ? Immediate Fail for every request ?
  40     // 2. At this level we might be dealing with Chunks of messages(not whole messages) and unexpected behavior might occur
  41     // when we send/queue 1 chunk and fail the other chunks
  42
  43     private IoOutputStream asyncIn;
  44
  45     // Order has to be preserved for queued writes
  46     private final Deque<PendingWriteRequest> pending = new LinkedList<>();
  47
  48     public AsyncSshHandlerWriter(final IoOutputStream asyncIn) {
  49         this.asyncIn = asyncIn;
  50     }
  51
  52     public synchronized void write(final ChannelHandlerContext ctx,
  53             final Object msg, final ChannelPromise promise) {
  54         // TODO check for isClosed, isClosing might be performed by mina SSH internally and is not required here
  55         // If we are closed/closing, set immediate fail
  56         if (asyncIn == null || asyncIn.isClosed() || asyncIn.isClosing()) {
  57             promise.setFailure(new IllegalStateException("Channel closed"));
  58         } else {
  59             final ByteBuf byteBufMsg = (ByteBuf) msg;
  60             if (pending.isEmpty() == false) {
  61                 queueRequest(ctx, byteBufMsg, promise);
  62                 return;
  63             }
  64
  65             writeWithPendingDetection(ctx, promise, byteBufMsg);
  66         }
  67     }
  68
  69     private void writeWithPendingDetection(final ChannelHandlerContext ctx, final ChannelPromise promise, final ByteBuf byteBufMsg) {
  70         try {
  71             if (logger.isTraceEnabled()) {
  72                 logger.trace("Writing request on channel: {}, message: {}", ctx.channel(), byteBufToString(byteBufMsg));
  73             }
  74             asyncIn.write(toBuffer(byteBufMsg)).addListener(new SshFutureListener<IoWriteFuture>() {
  75
  76                         @Override
  77                         public void operationComplete(final IoWriteFuture future) {
  78                             if (logger.isTraceEnabled()) {
  79                                 logger.trace("Ssh write request finished on channel: {} with result: {}: and ex:{}, message: {}",
  80                                         ctx.channel(), future.isWritten(), future.getException(), byteBufToString(byteBufMsg));
  81                             }
  82
  83                             // Notify success or failure
  84                             if (future.isWritten()) {
  85                                 promise.setSuccess();
  86                             } else {
  87                                 logger.warn("Ssh write request failed on channel: {} for message: {}", ctx.channel(), byteBufToString(byteBufMsg), future.getException());
  88                                 promise.setFailure(future.getException());
  89                             }
  90
  91                             // Not needed anymore, release
  92                             byteBufMsg.release();
  93
  94                             // Check pending queue and schedule next
  95                             // At this time we are guaranteed that we are not in pending state anymore so the next request should succeed
  96                             writePendingIfAny();
  97                         }
  98                     });
  99         } catch (final WritePendingException e) {
 100             queueRequest(ctx, byteBufMsg, promise);
 101         }
 102     }
 103
 104     private synchronized void writePendingIfAny() {
 105         if (pending.peek() == null) {
 106             return;
 107         }
 108
 109         // In case of pending, reschedule next message from queue
 110         final PendingWriteRequest pendingWrite = pending.poll();
 111         final ByteBuf msg = pendingWrite.msg;
 112         if (logger.isTraceEnabled()) {
 113             logger.trace("Writing pending request on channel: {}, message: {}", pendingWrite.ctx.channel(), byteBufToString(msg));
 114         }
 115
 116         writeWithPendingDetection(pendingWrite.ctx, pendingWrite.promise, msg);
 117     }
 118
 119     private static String byteBufToString(final ByteBuf msg) {
 120         msg.resetReaderIndex();
 121         final String s = msg.toString(Charsets.UTF_8);
 122         msg.resetReaderIndex();
 123         return s;
 124     }
 125
 126     private void queueRequest(final ChannelHandlerContext ctx, final ByteBuf msg, final ChannelPromise promise) {
 127 //        try {
 128         logger.debug("Write pending on channel: {}, queueing, current queue size: {}", ctx.channel(), pending.size());
 129         if (logger.isTraceEnabled()) {
 130             logger.trace("Queueing request due to pending: {}", byteBufToString(msg));
 131         }
 132         new PendingWriteRequest(ctx, msg, promise).pend(pending);
 133 //        } catch (final Exception ex) {
 134 //            logger.warn("Unable to queue write request on channel: {}. Setting fail for the request: {}", ctx.channel(), ex, byteBufToString(msg));
 135 //            msg.release();
 136 //            promise.setFailure(ex);
 137 //        }
 138     }
 139
 140     @Override
 141     public synchronized void close() {
 142         asyncIn = null;
 143     }
 144
 145     private Buffer toBuffer(final ByteBuf msg) {
 146         // TODO Buffer vs ByteBuf translate, Can we handle that better ?
 147         final byte[] temp = new byte[msg.readableBytes()];
 148         msg.readBytes(temp, 0, msg.readableBytes());
 149         return new Buffer(temp);
 150     }
 151
 152     private static final class PendingWriteRequest {
 153         private final ChannelHandlerContext ctx;
 154         private final ByteBuf msg;
 155         private final ChannelPromise promise;
 156
 157         public PendingWriteRequest(final ChannelHandlerContext ctx, final ByteBuf msg, final ChannelPromise promise) {
 158             this.ctx = ctx;
 159             // Reset reader index, last write (failed) attempt moved index to the end
 160             msg.resetReaderIndex();
 161             this.msg = msg;
 162             this.promise = promise;
 163         }
 164
 165         public void pend(final Queue<PendingWriteRequest> pending) {
 166             // Preconditions.checkState(pending.size() < MAX_PENDING_WRITES,
 167             // "Too much pending writes(%s) on channel: %s, remote window is not getting read or is too small",
 168             // pending.size(), ctx.channel());
 169             Preconditions.checkState(pending.offer(this), "Cannot pend another request write (pending count: %s) on channel: %s",
 170                     pending.size(), ctx.channel());
 171         }
 172     }
 173 }