xeij/NamedPipeInputStream.java- Source code ソースコード - XEiJ - STUDIO KAMADA
xeij/NamedPipeInputStream.java

    //========================================================================================
    //  NamedPipeInputStream.java
    //    en:Input from named pipe
    //    ja:名前付きパイプから入力します
    //  Copyright (C) 2003-2025 Makoto Kamada
    //
    //  This file is part of the XEiJ (X68000 Emulator in Java).
    //  You can use, modify and redistribute the XEiJ if the conditions are met.
    //  Read the XEiJ License for more details.
    //  https://stdkmd.net/xeij/
    //========================================================================================

    package xeij;

    import java.io.*;
    import java.lang.foreign.*;  //Arena,FunctionDescriptor,Linker,MemorySegment,SymbolLookup,ValueLayout
    import java.lang.invoke.*;  //MethodHandle
    import java.util.*;  //NoSuchElementException

    //class NamedPipeInputStream
    //  名前付きパイプInputStream
    public abstract class NamedPipeInputStream extends InputStream {

      //npis = NamedPipeInputStream.create (name)
      //  コンストラクタ
      //  npis  名前付きパイプInputStream
      //  name  名前付きパイプの名前。プレフィックスを含まない
      public static NamedPipeInputStream create (String name) throws IOException {
        return System.getProperty ("os.name").indexOf ("Windows") < 0 ? new Gen (name) : new Win (name);
      }  //open

      protected String path;  //パス。プレフィックスを含む
      protected volatile boolean closed;  //true=すでに閉じた
      protected volatile boolean connecting;  //true=接続待ちブロック中
      protected volatile boolean reading;  //true=受信待ちブロック中
      protected Thread thread;  //受信スレッド
      protected ByteQueue queue;  //受信キュー

      //npis = new NamedPipeInputStream (name)
      //  コンストラクタ
      //  npis  名前付きパイプInputStream
      //  name  名前付きパイプの名前。プレフィックスを含まない
      private NamedPipeInputStream (String name) throws IOException {
        //開始
        osStart (name);
        //受信キューを作る
        queue = new ByteQueue ();
        //受信スレッドを開始する
        thread = new Thread () {
          byte[] b = new byte[1024];
          @Override public void run () {
            try {
              //開いて接続する
              connecting = true;
              osOpenAndConnect ();  //送信側が開くまでブロックする
              connecting = false;
              while (!closed) {
                //受信する
                reading = true;
                int k = osRead (b);  //少なくとも1バイト受信するまでブロックする
                reading = false;
                if (closed) {
                  break;
                }
                if (k == -1) {  //EOFのとき
                  osClose ();  //直ちに接続し直す
                  connecting = true;
                  osOpenAndConnect ();  //送信側が開くまでブロックする
                  connecting = false;
                  continue;
                }
                //キューに書き込む
                queue.write (b, 0, k);
              }
            } catch (IOException ioe) {
              connecting = false;
              reading = false;
            }
            //閉じる
            osClose ();
          }
        };
        thread.start ();
        if (false) {
          System.out.println ("named pipe " + path + " opened");
        }
      }  //NamedPipeInputStream

      //------------------------------------------------------------------------
      //n = available ()
      //  ブロックせずに受信できる長さを返す
      //  n  ブロックせずに受信できる長さ
      @Override
      public int available () throws IOException {
        //閉じていたら失敗
        if (closed) {
          throw new IOException ("named pipe " + path + " closed");
        }
        //ブロックせずに受信できる長さを返す
        return queue.used ();
      }  //available

      //------------------------------------------------------------------------
      //close ()
      //  名前付きパイプを閉じる
      @Override
      public void close () {
        //閉じていたら何もしない
        if (closed) {
          return;
        }
        closed = true;
        //ブロックを解除する
        osUnblock ();
        //閉じる
        osClose ();
        //受信スレッドの終了を待つ
        try {
          thread.join (100L);
        } catch (InterruptedException ie) {
        }
        //終了
        osEnd ();
        //キューの読み出しをキャンセルする
        queue.cancel ();  //waitAndReadのブロックを解く
        if (false) {
          System.out.println ("named pipe " + path + " closed");
        }
      }  //close

      //------------------------------------------------------------------------
      //d = read ()
      //  1バイト受信する。1バイト受信するまでブロックする
      //  d  受信したデータ。0～255。-1=キャンセルされた
      @Override
      public int read () throws IOException {
        //閉じていたら失敗
        if (closed) {
          throw new IOException ("named pipe " + path + " closed");
        }
        //キューから1バイト読み出す
        return queue.waitAndRead ();  //1バイト読み出すまでブロックする
      }  //read

      //------------------------------------------------------------------------
      //k = read (b)
      //  配列に受信する。少なくとも1バイト受信するまでブロックする
      //  k  受信した長さ。-1=キャンセルされた
      //  b  配列
      @Override
      public int read (byte[] b) throws IOException {
        //閉じていたら失敗
        if (closed) {
          throw new IOException ("named pipe " + path + " closed");
        }
        //キューから配列に読み出す
        return queue.waitAndRead (b, 0, b.length);  //少なくとも1バイト読み出すまでブロックする
      }  //read

      //------------------------------------------------------------------------
      //k = read (b, o, n)
      //  配列に受信する。少なくとも1バイト受信するまでブロックする
      //  k  受信した長さ。-1=キャンセルされた
      //  b  配列
      //  o  位置
      //  n  長さ
      @Override
      public int read (byte[] b, int o, int n) throws IOException {
        //閉じていたら失敗
        if (closed) {
          throw new IOException ("named pipe " + path + " closed");
        }
        //引数を確認する
        if (o < 0 || n < 0 || b.length < o + n) {  //範囲外
          throw new IndexOutOfBoundsException ("b.length=" + b.length + ", o=" + o + ", n=" + n);
        }
        //キューから配列に読み出す。少なくとも1バイト読み出すまでブロックする
        return queue.waitAndRead (b, o, n);
      }  //read

      //------------------------------------------------------------------------
      //k = readNBytes (b, o, n)
      //  配列に受信する。nバイト受信するまでブロックする
      //  k  受信した長さ。-1=キャンセルされた
      //  b  配列
      //  o  位置
      //  n  長さ
      @Override
      public int readNBytes (byte[] b, int o, int n) throws IOException {
        //閉じていたら失敗
        if (closed) {
          throw new IOException ("named pipe " + path + " closed");
        }
        //引数を確認する
        if (o < 0 || n < 0 || b.length < o + n) {  //範囲外
          throw new IndexOutOfBoundsException ("b.length=" + b.length + ", o=" + o + ", n=" + n);
        }
        //キューから配列に読み出す
        int k = 0;
        while (k < n) {
          int t = queue.waitAndRead (b, o + k, n - k);  //少なくとも1バイト読み出すまでブロックする
          if (t == -1) {  //キャンセルされた
            return -1;
          }
          k += t;
        }
        //受信した長さを返す
        return k;
      }  //readNBytes

      protected abstract void osStart (String name) throws IOException;
      protected abstract void osOpenAndConnect () throws IOException;
      protected abstract int osRead (byte[] b) throws IOException;
      protected abstract void osUnblock ();
      protected abstract void osClose ();
      protected abstract void osEnd ();

      //class Gen
      //  UNIX系OS用
      private static class Gen extends NamedPipeInputStream {

        private File file;  //ファイル
        private FileInputStream stream;  //FileInputStream。null=まだ開いていないまたはすでに閉じた

        //npis = new Gen (name)
        //  コンストラクタ
        protected Gen (String name) throws IOException {
          super (name);
        }  //Gen

        //osStart (name)
        //  開始
        @Override
        protected void osStart (String name) throws IOException {
          //パス
          path = System.getProperty ("java.io.tmpdir") + "/" + name;  //通常は/tmp/～
          //ファイル
          file = new File (path);
          //mkfifoでFIFOを作る
          file.delete ();
          Process process;
          try {
            process = new ProcessBuilder ("mkfifo", path).inheritIO ().start ();
          } catch (IOException ioe) {  //start失敗。mkfifoを開始できなかった
            throw new IOException ("mkfifo " + path + " not started");
          }
          try {
            int exitCode = process.waitFor ();
            if (exitCode != 0) {  //mkfifoがエラー終了した
              file.delete ();
              throw new IOException ("mkfifo " + path + " terminated with exit code " + exitCode);
            }
          } catch (InterruptedException ie) {  //waitFor中断。mkfifoの終了を確認できなかった
            file.delete ();
            throw new IOException ("mkfifo " + path + " interrupted");
          }
        }  //osStart

        //osOpenAndConnect ()
        //  開いて接続する。送信側が開くまでブロックする
        @Override
        protected void osOpenAndConnect () throws IOException {
          if (stream == null) {
            stream = new FileInputStream (file);  //送信側が開くまでブロックする
          }
        }  //osOpenAndConnect

        //k = osRead (b)
        //  受信する。少なくとも1バイト受信するまでブロックする
        @Override
        protected int osRead (byte[] b) throws IOException {
          return stream.read (b);
        }  //osRead

        //osUnblock ()
        //  ブロックを解除する
        @Override
        protected void osUnblock () {
          //new FileInputStream()のブロックを解除する
          //  new FileInputStream()は送信側が開くまでブロックする。new FileOutputStream()でダミーの送信側を開くことで解除できる
          //  new FileInputStream()でブロックしていないのにnew FileOutputStream()を行うとnew FileOutputStream()がブロックしてしまう可能性がある
          if (connecting) {
            try {
              Thread.sleep (50L);
            } catch (InterruptedException ie) {
            }
            if (connecting) {
              try {
                new FileOutputStream (file).close ();
              } catch (IOException ioe) {
              }
              for (int i = 0; i < 100 && connecting; i++) {
                try {
                  Thread.sleep (50L);
                } catch (InterruptedException ie) {
                }
              }
              if (connecting) {
                System.out.println ("named pipe " + path + " unblocking timeout");
              }
            }
          }
          //FileInputStream.read()のブロックが解除されるまで待つ
          //  FileInputStream.read()は送信側が開いたまま送信しないとブロックする。送信側が送信するか閉じるまでブロックを解除できない
          if (reading) {
            for (int i = 0; i < 100 && reading; i++) {
              try {
                Thread.sleep (50L);
              } catch (InterruptedException ie) {
              }
            }
            if (reading) {
              System.out.println ("user operation required to unblock named pipe " + path);
              while (reading) {
                try {
                  Thread.sleep (1000L);
                } catch (InterruptedException ie) {
                }
              }
            }
          }
        }  //osUnblock

        //osClose ()
        //  閉じる
        @Override
        protected void osClose () {
          //開いていたら閉じる
          if (stream != null) {
            try {
              stream.close ();
            } catch (IOException ioe) {
            }
            stream = null;
          }
        }  //osClose

        //osEnd ()
        //  終了
        @Override
        protected void osEnd () {
          file.delete ();
        }  //osEnd

      }  //class Gen

      //class Win
      //  Windows用
      private static class Win extends NamedPipeInputStream {

        //エラーコード
        //  https://learn.microsoft.com/en-us/windows/win32/debug/system-error-codes
        private static final int ERROR_BROKEN_PIPE = 109;  //パイプが終了した。誰かが開いて閉じた
        private static final int ERROR_BAD_PIPE = 230;  //パイプの状態が無効。誰も開いていない
        private static final int ERROR_NO_DATA = 232;  //既に切断された
        private static final int ERROR_PIPE_CONNECTED = 535;  //既に接続している
        private static final int ERROR_PIPE_LISTENING = 536;  //誰も開いていない
        private static final int ERROR_OPERATION_ABORTED = 995;  //スレッドが終了したか操作が取り消された
        private static final int ERROR_NOT_FOUND = 1168;  //取り消す操作がない
        private static final long INVALID_HANDLE_VALUE = -1L;

        //リンカ
        private Linker linker;
        private MethodHandle downcallHandle (MemorySegment address, FunctionDescriptor function) {
          return linker.downcallHandle (address, function);
        }

        //アリーナ
        private Arena arena;

        //メソッド
        private MethodHandle CancelIoEx;
        private MethodHandle CloseHandle;
        private MethodHandle ConnectNamedPipe;
        private static final int PIPE_ACCESS_INBOUND = 0x00000001;
        private static final int PIPE_TYPE_BYTE = 0x00000000;
        private static final int PIPE_WAIT = 0x00000000;
        private static final int PIPE_UNLIMITED_INSTANCES = 255;
        private static final int BUFFER_SIZE = 8192;
        private MethodHandle CreateNamedPipeA;
        private MethodHandle GetLastError;
        private MethodHandle ReadFile;

        private MemorySegment handle;  //HANDLE。null=まだ開いていないまたはすでに閉じた

        //npis = new Win (name)
        //  コンストラクタ
        protected Win (String name) throws IOException {
          super (name);
        }  //Win

        //osStart (name)
        //  開始
        @Override
        protected void osStart (String name) throws IOException {
          //リンカ
          linker = Linker.nativeLinker ();
          //アリーナ
          arena = Arena.ofAuto ();
          //ライブラリ
          SymbolLookup kernel32 = SymbolLookup.libraryLookup ("kernel32", arena);
          //メソッド
          try {
            //CancelIoEx関数
            //  https://learn.microsoft.com/ja-jp/windows/win32/fileio/cancelioex-func
            CancelIoEx = downcallHandle (
              kernel32.findOrThrow ("CancelIoEx"),
              FunctionDescriptor.of (
                ValueLayout.JAVA_INT,  //BOOL
                ValueLayout.ADDRESS,  //HANDLE hFile
                ValueLayout.ADDRESS));  //LPOVERLAPPED lpOverlapped
            //CloseHandle関数
            //  https://learn.microsoft.com/ja-jp/windows/win32/api/handleapi/nf-handleapi-closehandle
            CloseHandle = downcallHandle (
              kernel32.findOrThrow ("CloseHandle"),
              FunctionDescriptor.of (
                ValueLayout.JAVA_INT,  //BOOL
                ValueLayout.ADDRESS));  //HANDLE hObject
            //ConnectNamedPipe関数
            //  https://learn.microsoft.com/ja-jp/windows/win32/api/namedpipeapi/nf-namedpipeapi-connectnamedpipe
            ConnectNamedPipe = downcallHandle (
              kernel32.findOrThrow ("ConnectNamedPipe"),
              FunctionDescriptor.of (
                ValueLayout.JAVA_INT,  //BOOL
                ValueLayout.ADDRESS,  //HANDLE hNamedPipe
                ValueLayout.ADDRESS));  //LPOVERLAPPED lpOverlapped
            //CreateNamedPipeA関数
            //  https://learn.microsoft.com/ja-jp/windows/win32/api/winbase/nf-winbase-createnamedpipea
            CreateNamedPipeA = downcallHandle (
              kernel32.findOrThrow ("CreateNamedPipeA"),
              FunctionDescriptor.of (
                ValueLayout.ADDRESS,  //HANDLE
                ValueLayout.ADDRESS,  //LPCSTR lpName
                ValueLayout.JAVA_INT,  //DWORD dwOpenMode
                ValueLayout.JAVA_INT,  //DWORD dwPipeMode
                ValueLayout.JAVA_INT,  //DWORD nMaxInstances
                ValueLayout.JAVA_INT,  //DWORD nOutBufferSize
                ValueLayout.JAVA_INT,  //DWORD nInBufferSize
                ValueLayout.JAVA_INT,  //DWORD nDefaultTimeOut
                ValueLayout.ADDRESS));  //LPSECURITY_ATTRIBUTES lpSecurityAttributes
            //GetLastError関数
            //  https://learn.microsoft.com/ja-jp/windows/win32/api/errhandlingapi/nf-errhandlingapi-getlasterror
            GetLastError = downcallHandle (
              kernel32.findOrThrow ("GetLastError"),
              FunctionDescriptor.of (
                ValueLayout.JAVA_INT));  //DWORD
            //ReadFile関数
            //  https://learn.microsoft.com/ja-jp/windows/win32/api/fileapi/nf-fileapi-readfilef
            ReadFile = downcallHandle (
              kernel32.findOrThrow ("ReadFile"),
              FunctionDescriptor.of (
                ValueLayout.JAVA_INT,  //BOOL
                ValueLayout.ADDRESS,  //HANDLE hFile
                ValueLayout.ADDRESS,  //LPVOID lpBuffer
                ValueLayout.JAVA_INT,  //DWORD nNumberOfBytesToRead
                ValueLayout.ADDRESS,  //LPDWORD lpNumberOfBytesRead
                ValueLayout.ADDRESS));  //LPOVERLAPPED lpOverlapped
          } catch (NoSuchElementException nsee) {
            nsee.printStackTrace ();
          }
          //パス
          path = "\\\\.\\pipe\\" + name;
        }  //osStart

        //osOpenAndConnect ()
        //  開いて接続する。送信側が開くまでブロックする
        @Override
        protected void osOpenAndConnect () throws IOException {
          //閉じていたら開く
          if (handle == null) {
            try {
              int error;
              //  https://learn.microsoft.com/ja-jp/windows/win32/api/winbase/nf-winbase-createnamedpipea
              if ((handle = (MemorySegment) CreateNamedPipeA.invoke (
                arena.allocateFrom (path),  //LPCSTR lpName
                PIPE_ACCESS_INBOUND,  //DWORD dwOpenMode
                PIPE_TYPE_BYTE | PIPE_WAIT,  //DWORD dwPipeMode
                PIPE_UNLIMITED_INSTANCES,  //DWORD nMaxInstances
                0,  //DWORD nOutBufferSize
                BUFFER_SIZE,  //DWORD nInBufferSize
                0,  //DWORD nDefaultTimeOut
                MemorySegment.NULL  //LPSECURITY_ATTRIBUTES lpSecurityAttributes
                )).address () == INVALID_HANDLE_VALUE &&
                  (error = (int) GetLastError.invoke ()) != -1) {
                handle = null;
                //開けなかったら失敗
                throw new IOException ("CreateNamedPipeA returned error code " + error);
              }
            } catch (IOException ioe) {
              throw ioe;
            } catch (Throwable e) {
              e.printStackTrace ();
              throw new IOException ("CreateNamedPipeA invocation failed");
            }
          }
          //接続を待つ。送信側が開くまでブロックする
          try {
            int error;
            //  https://learn.microsoft.com/ja-jp/windows/win32/api/namedpipeapi/nf-namedpipeapi-connectnamedpipe
            if ((int) ConnectNamedPipe.invoke (
              handle,  //HANDLE hNamedPipe
              MemorySegment.NULL) == 0 &&  //LPOVERLAPPED lpOverlapped
                (error = (int) GetLastError.invoke ()) != -1) {
              if (error == 0 ||  //誰かがエラーコードを上書きした
                  error == ERROR_NO_DATA ||  //232 既に切断された
                  error == ERROR_PIPE_CONNECTED) {  //535 既に接続している
              } else if (error == ERROR_OPERATION_ABORTED) {  //995 スレッドが終了したか操作が取り消された
                throw new InterruptedIOException ("ConnectNamedPipe aborted");
              } else {
                throw new IOException ("ConnectNamedPipe returned error code " + error);
              }
            }
          } catch (IOException ioe) {
            throw ioe;
          } catch (Throwable e) {
            e.printStackTrace ();
            throw new IOException ("ConnectNamedPipe invocation failed");
          }
        }  //osOpenAndConnect

        //k = osRead (b)
        //  受信する。少なくとも1バイト受信するまでブロックする
        @Override
        protected int osRead (byte[] b) throws IOException {
          int n = b.length;
          MemorySegment buf = arena.allocate ((long) n);  //MemorySegment.ofArray(b)はasSlice(o+k)できない
          int k = 0;
          MemorySegment t = arena.allocate (ValueLayout.JAVA_INT);
          t.set (ValueLayout.JAVA_INT, 0L, 0);
          try {
            while (k < 1) {
              t.set (ValueLayout.JAVA_INT, 0L, 0);
              int error;
              //  https://learn.microsoft.com/ja-jp/windows/win32/api/fileapi/nf-fileapi-readfile
              if ((int) ReadFile.invoke (
                handle,  //HANDLE hFile
                buf.asSlice ((long) k),  //LPVOID lpBuffer
                n - k,  //DWORD nNumberOfBytesToRead
                t,  //LPDWORD lpNumberOfBytesRead
                MemorySegment.NULL) == 0 &&  //LPOVERLAPPED lpOverlapped
                  (error = (int) GetLastError.invoke ()) != -1) {
                if (closed) {
                  throw new IOException ("named pipe " + path + " closed");
                }
                if (error == 0 ||  //誰かがエラーコードを上書きした
                    error == ERROR_BROKEN_PIPE ||  //109 パイプが終了した。誰かが開いて閉じた
                    error == ERROR_PIPE_LISTENING) {  //536 誰も開いていない
                  //0のとき閉じたのか取り消されたのかわからないが、後者はclosedなのでここには来ない
                  return -1;  //EOF
                } else if (error == ERROR_OPERATION_ABORTED) {  //995 スレッドが終了したか操作が取り消された
                  throw new InterruptedIOException ("ReadFile aborted");
                } else {
                  throw new IOException ("ReadFile returned error code " + error);
                }
              }
              if (closed) {
                throw new IOException ("named pipe " + path + " closed");
              }
              k += t.get (ValueLayout.JAVA_INT, 0L);
            }
          } catch (IOException ioe) {
            throw ioe;
          } catch (Throwable e) {
            e.printStackTrace ();
            throw new IOException ("ReadFile invocation failed");
          }
          System.arraycopy (buf.toArray (ValueLayout.JAVA_BYTE), 0, b, 0, k);
          return k;
        }  //osRead

        //osUnblock ()
        //  ブロックを解除する
        @Override
        protected void osUnblock () {
          //ConnectNamedPipeまたはReadFileのブロックを解除する
          //  CancelIoExで取り消された操作はERROR_OPERATION_ABORTEDになる
          //  MethodHandle.invoke()の問題で、取り消された操作がERROR_OPERATION_ABORTEDでなく0を受け取る可能性がある
          if (connecting || reading) {
            try {
              Thread.sleep (50L);
            } catch (InterruptedException ie) {
            }
            if (connecting) {
              try {
                int error;
                //  https://learn.microsoft.com/ja-jp/windows/win32/fileio/cancelioex-func
                if ((int) CancelIoEx.invoke (
                  handle,  //HANDLE hFile
                  MemorySegment.NULL) == 0 &&  //LPOVERLAPPED lpOverlapped
                    (error = (int) GetLastError.invoke ()) != -1) {
                  //if (error == ERROR_NOT_FOUND) {  //1168 取り消す操作がない
                  //}
                }
              } catch (Throwable e) {
                //e.printStackTrace ();
                //throw new IOException ("CancelIoEx invocation failed");
              }
              for (int i = 0; i < 100 && (connecting || reading); i++) {
                try {
                  Thread.sleep (50L);
                } catch (InterruptedException ie) {
                }
              }
              if (connecting || reading) {
                System.out.println ("named pipe " + path + " unblocking timeout");
              }
            }
          }
        }  //osUnblock

        //osClose ()
        //  閉じる
        @Override
        protected void osClose () {
          //開いていたら閉じる
          if (handle != null) {
            try {
              int error;
              //  https://learn.microsoft.com/ja-jp/windows/win32/api/handleapi/nf-handleapi-closehandle
              if ((int) CloseHandle.invoke (
                handle) == 0 &&  //HANDLE hObject
                  (error = (int) GetLastError.invoke ()) != -1) {
                //throw new IOException ("CloseHandle returned error code " + error);
              }
              //} catch (IOException ioe) {
              //  throw ioe;
            } catch (Throwable e) {
              e.printStackTrace ();
              //throw new IOException ("CloseHandle invocation failed");
            }
            handle = null;
          }
        }  //osClose

        //osEnd ()
        //  終了
        @Override
        protected void osEnd () {
        }  //osEnd

      }  //class Win

    }  //class NamedPipeInputStream

STUDIO KAMADA
Copyright © 1999-2026 Makoto Kamada