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         private final String id;
  24
  25         public SocketPipeMirror(String id) throws IOException {
  26                 this.id = id;
  27                 inPipe = Pipe.open();
  28                 outPipe = Pipe.open();
  29                 in = Channels.newInputStream(inPipe.source());
  30                 out = Channels.newOutputStream(outPipe.sink());
  31         }
  32
  33         public void open(SocketChannel channel) {
  34                 readInThread = new Thread(() -> {
  35
  36                         try {
  37                                 ByteBuffer buffer = ByteBuffer.allocate(1024);
  38                                 while (!readInThread.isInterrupted() && channel.isConnected()) {
  39                                         if (channel.read(buffer) < 0) {
  40                                                 in.close();
  41                                                 break;
  42                                         }
  43                                         buffer.flip();
  44                                         inPipe.sink().write(buffer);
  45                                         buffer.rewind();
  46                                 }
  47                         } catch (AsynchronousCloseException e) {
  48                                 // ignore
  49                         } catch (IOException e) {
  50                                 e.printStackTrace();
  51                         }
  52                 }, "JShell read " + id);
  53                 readInThread.start();
  54
  55                 writeOutThread = new Thread(() -> {
  56
  57                         try {
  58                                 ByteBuffer buffer = ByteBuffer.allocate(1024);
  59                                 while (!writeOutThread.isInterrupted() && channel.isConnected()) {
  60                                         if (outPipe.source().read(buffer) < 0) {
  61                                                 out.close();
  62                                                 break;
  63                                         }
  64                                         buffer.flip();
  65                                         channel.write(buffer);
  66                                         buffer.rewind();
  67                                 }
  68                         } catch (IOException e) {
  69                                 e.printStackTrace();
  70                         }
  71                 }, "JShell write " + id);
  72                 writeOutThread.start();
  73
  74         }
  75
  76         @Override
  77         public void close() throws IOException {
  78                 // TODO make it more robust
  79 //              readInThread.interrupt();
  80 //              writeOutThread.interrupt();
  81                 in.close();
  82                 out.close();
  83                 try {
  84                         readInThread.join();
  85                 } catch (InterruptedException e) {
  86                         // silent
  87                 }
  88                 try {
  89                         writeOutThread.join();
  90                 } catch (InterruptedException e) {
  91                         // silent
  92                 }
  93         }
  94
  95         public InputStream getInputStream() {
  96                 return in;
  97         }
  98
  99         public OutputStream getOutputStream() {
 100                 return out;
 101         }
 102 }