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                     throw new RuntimeException(e);
  61                 }
  62             }
  63             return this.bb.readByte() & 0xFF;
  64         }
  65     }
  66
  67     @Override
  68     public int available() throws IOException {
  69         synchronized (lock) {
  70             return this.bb.readableBytes();
  71         }
  72     }
  73
  74     public void channelRegistered(ChannelHandlerContext ctx)
  75             throws Exception {
  76         ctx.fireChannelRegistered();
  77     }
  78
  79     public void channelUnregistered(ChannelHandlerContext ctx)
  80             throws Exception {
  81         ctx.fireChannelUnregistered();
  82     }
  83
  84     public void channelActive(ChannelHandlerContext ctx)
  85             throws Exception {
  86         ctx.fireChannelActive();
  87     }
  88
  89     public void channelInactive(ChannelHandlerContext ctx)
  90             throws Exception {
  91         ctx.fireChannelInactive();
  92     }
  93
  94     public void channelRead(ChannelHandlerContext ctx, Object o)
  95             throws Exception {
  96         synchronized(lock) {
  97             this.bb.discardReadBytes();
  98             this.bb.writeBytes((ByteBuf) o);
  99             lock.notifyAll();
 100         }
 101     }
 102
 103     public void channelReadComplete(ChannelHandlerContext ctx)
 104             throws Exception {
 105         ctx.fireChannelReadComplete();
 106     }
 107
 108     public void userEventTriggered(ChannelHandlerContext ctx, Object o)
 109             throws Exception {
 110         ctx.fireUserEventTriggered(o);
 111     }
 112
 113     public void channelWritabilityChanged(ChannelHandlerContext ctx)
 114             throws Exception {
 115         ctx.fireChannelWritabilityChanged();
 116     }
 117
 118     public void handlerAdded(ChannelHandlerContext ctx)
 119             throws Exception {
 120     }
 121
 122     public void handlerRemoved(ChannelHandlerContext ctx)
 123             throws Exception {
 124     }
 125
 126     public void exceptionCaught(ChannelHandlerContext ctx, Throwable throwable)
 127             throws Exception {
 128         ctx.fireExceptionCaught(throwable);
 129     }
 130 }
 131