opendaylight/netconf/netconf-util/src/main/java/org/opendaylight/controller/netconf/util/handler/ssh/virtualsocket/ChannelInputStream.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
   9 package org.opendaylight.controller.netconf.util.handler.ssh.virtualsocket;
  10
  11 import io.netty.buffer.ByteBuf;
  12 import io.netty.buffer.Unpooled;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelInboundHandler;
  15
  16 import java.io.IOException;
  17 import java.io.InputStream;
  18
  19 /**
  20  * Class provides {@link InputStream} functionality to users of virtual socket.
  21  */
  22 public class ChannelInputStream extends InputStream implements ChannelInboundHandler {
  23     private final Object lock = new Object();
  24     private final ByteBuf bb = Unpooled.buffer();
  25
  26     @Override
  27     public int read(byte b[], int off, int len) throws IOException {
  28         if (b == null) {
  29             throw new NullPointerException();
  30         } else if (off < 0 || len < 0 || len > b.length - off) {
  31             throw new IndexOutOfBoundsException();
  32         } else if (len == 0) {
  33             return 0;
  34         }
  35
  36         int bytesRead = 1;
  37         synchronized (lock) {
  38             int c = read();
  39
  40             b[off] = (byte)c;
  41
  42             if(this.bb.readableBytes() == 0) return bytesRead;
  43
  44             int ltr = len-1;
  45             ltr = (ltr <= bb.readableBytes()) ? ltr : bb.readableBytes();
  46
  47             bb.readBytes(b, 1, ltr);
  48             bytesRead += ltr;
  49         }
  50         return bytesRead;
  51     }
  52
  53     @Override
  54     public int read() throws IOException {
  55         synchronized (lock) {
  56             while (this.bb.readableBytes() == 0) {
  57                 try {
  58                     lock.wait();
  59                 } catch (InterruptedException e) {
  60                     Thread.currentThread().interrupt();
  61                     throw new RuntimeException(e);
  62                 }
  63             }
  64             return this.bb.readByte() & 0xFF;
  65         }
  66     }
  67
  68     @Override
  69     public int available() throws IOException {
  70         synchronized (lock) {
  71             return this.bb.readableBytes();
  72         }
  73     }
  74
  75     public void channelRegistered(ChannelHandlerContext ctx)
  76             throws Exception {
  77         ctx.fireChannelRegistered();
  78     }
  79
  80     public void channelUnregistered(ChannelHandlerContext ctx)
  81             throws Exception {
  82         ctx.fireChannelUnregistered();
  83     }
  84
  85     public void channelActive(ChannelHandlerContext ctx)
  86             throws Exception {
  87         ctx.fireChannelActive();
  88     }
  89
  90     public void channelInactive(ChannelHandlerContext ctx)
  91             throws Exception {
  92         ctx.fireChannelInactive();
  93     }
  94
  95     public void channelRead(ChannelHandlerContext ctx, Object o)
  96             throws Exception {
  97         synchronized(lock) {
  98             this.bb.discardReadBytes();
  99             this.bb.writeBytes((ByteBuf) o);
 100             lock.notifyAll();
 101         }
 102     }
 103
 104     public void channelReadComplete(ChannelHandlerContext ctx)
 105             throws Exception {
 106         ctx.fireChannelReadComplete();
 107     }
 108
 109     public void userEventTriggered(ChannelHandlerContext ctx, Object o)
 110             throws Exception {
 111         ctx.fireUserEventTriggered(o);
 112     }
 113
 114     public void channelWritabilityChanged(ChannelHandlerContext ctx)
 115             throws Exception {
 116         ctx.fireChannelWritabilityChanged();
 117     }
 118
 119     public void handlerAdded(ChannelHandlerContext ctx)
 120             throws Exception {
 121     }
 122
 123     public void handlerRemoved(ChannelHandlerContext ctx)
 124             throws Exception {
 125     }
 126
 127     public void exceptionCaught(ChannelHandlerContext ctx, Throwable throwable)
 128             throws Exception {
 129         ctx.fireExceptionCaught(throwable);
 130     }
 131 }
 132