org.argeo.cms.jshell/src/org/argeo/cms/jshell/SocketPipeMirror.java

   1 package org.argeo.cms.jshell;
   2
   3 import java.io.Closeable;
   4 import java.io.IOException;
   5 import java.io.InputStream;
   6 import java.io.OutputStream;
   7 import java.nio.ByteBuffer;
   8 import java.nio.channels.AsynchronousCloseException;
   9 import java.nio.channels.Channels;
  10 import java.nio.channels.Pipe;
  11 import java.nio.channels.SocketChannel;
  12
  13 class SocketPipeMirror implements Closeable {
  14         private final Pipe inPipe;
  15         private final Pipe outPipe;
  16
  17         private final InputStream in;
  18         private final OutputStream out;
  19
  20         private Thread readInThread;
  21         private Thread writeOutThread;
  22
  23         public SocketPipeMirror() throws IOException {
  24                 inPipe = Pipe.open();
  25                 outPipe = Pipe.open();
  26                 in = Channels.newInputStream(inPipe.source());
  27                 out = Channels.newOutputStream(outPipe.sink());
  28         }
  29
  30         public void open(SocketChannel channel) {
  31                 readInThread = new Thread(() -> {
  32
  33                         try {
  34                                 ByteBuffer buffer = ByteBuffer.allocate(1024);
  35                                 while (!readInThread.isInterrupted() && channel.isConnected()) {
  36                                         if (channel.read(buffer) < 0)
  37                                                 break;
  38                                         buffer.flip();
  39                                         inPipe.sink().write(buffer);
  40                                         buffer.rewind();
  41                                 }
  42                         } catch (AsynchronousCloseException e) {
  43                                 // ignore
  44                                 // TODO make it cleaner
  45                         } catch (IOException e) {
  46                                 e.printStackTrace();
  47                         }
  48                 }, "Read in");
  49                 readInThread.start();
  50
  51                 writeOutThread = new Thread(() -> {
  52
  53                         try {
  54                                 ByteBuffer buffer = ByteBuffer.allocate(1024);
  55                                 while (!writeOutThread.isInterrupted() && channel.isConnected()) {
  56                                         if (outPipe.source().read(buffer) < 0)
  57                                                 break;
  58                                         buffer.flip();
  59                                         channel.write(buffer);
  60                                         buffer.rewind();
  61                                 }
  62                         } catch (IOException e) {
  63                                 e.printStackTrace();
  64                         }
  65                 }, "Write out");
  66                 writeOutThread.start();
  67
  68         }
  69
  70         @Override
  71         public void close() throws IOException {
  72                 // TODO make it more robust
  73                 readInThread.interrupt();
  74                 writeOutThread.interrupt();
  75                 in.close();
  76                 out.close();
  77         }
  78
  79         public InputStream getInputStream() {
  80                 return in;
  81         }
  82
  83         public OutputStream getOutputStream() {
  84                 return out;
  85         }
  86 }