public static void main(String[] args) throws Exception {
    PipedWriter out = new PipedWriter();
    PipedReader in = new PipedReader();
    // 将输出流和输入流进行连接，否则在使用时会抛出IOException
    out.connect(in);
    Thread printThread = new Thread(new Print(in), "PrintThread");
    printThread.start();
    int receive = 0;
    try {
        while ((receive = System.in.read()) != -1) {
            out.write(receive);
        }
    } finally {
        out.close();
    }
}
 

@Override
protected void assertConnect(String term) throws Exception {
  if (endpoint != null) {
    throw failure("Already a session");
  }
  final CountDownLatch latch = new CountDownLatch(1);
  final PipedWriter out = new PipedWriter();
  in = new PipedReader(out);
  endpoint = new Endpoint() {
    @Override
    public void onOpen(Session session, EndpointConfig endpointConfig) {
      session.addMessageHandler(new MessageHandler.Whole<String>() {
        @Override
        public void onMessage(String message) {
          try {
            out.write(message);
          } catch (IOException e) {
            e.printStackTrace();
          }
        }
      });
      latch.countDown();
    }
    @Override
    public void onClose(Session sess, CloseReason closeReason) {
      session = null;
      endpoint = null;
      in = null;
    }
    @Override
    public void onError(Session session, Throwable thr) {
    }
  };
  ClientEndpointConfig clientEndpointConfig = ClientEndpointConfig.Builder.create().build();
  WebSocketContainer webSocketContainer = ContainerProvider.getWebSocketContainer();
  session = webSocketContainer.connectToServer(endpoint, clientEndpointConfig, new URI("http://localhost:8080/ws"));
  latch.await();
}
 

private static PipedReader newPipedReader(String contents) {
    try (PipedWriter w = new PipedWriter()) {
        PipedReader r = new PipedReader(w);
        w.write(contents);
        return r;
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
}
 

@Override
protected void assertConnect(String term) throws Exception {
  if (endpoint != null) {
    throw failure("Already a session");
  }
  CountDownLatch latch = new CountDownLatch(1);
  PipedWriter out = new PipedWriter();
  in = new PipedReader(out);
  endpoint = new Endpoint() {
    @Override
    public void onOpen(Session session, EndpointConfig endpointConfig) {
      session.addMessageHandler(new MessageHandler.Whole<String>() {
        @Override
        public void onMessage(String message) {
          try {
            out.write(message);
          } catch (IOException e) {
            e.printStackTrace();
          }
        }
      });
      latch.countDown();
    }
    @Override
    public void onClose(Session sess, CloseReason closeReason) {
      session = null;
      endpoint = null;
      in = null;
    }
    @Override
    public void onError(Session session, Throwable thr) {
    }
  };
  ClientEndpointConfig clientEndpointConfig = ClientEndpointConfig.Builder.create().build();
  WebSocketContainer webSocketContainer = ContainerProvider.getWebSocketContainer();
  session = webSocketContainer.connectToServer(endpoint, clientEndpointConfig, new URI("http://localhost:8080/ws"));
  latch.await();
}
 

@Override
protected void assertConnect(String term) throws Exception {
  if (endpoint != null) {
    throw failure("Already a session");
  }
  CountDownLatch latch = new CountDownLatch(1);
  PipedWriter out = new PipedWriter();
  in = new PipedReader(out);
  endpoint = new Endpoint() {
    @Override
    public void onOpen(Session session, EndpointConfig endpointConfig) {
      session.addMessageHandler(new MessageHandler.Whole<String>() {
        @Override
        public void onMessage(String message) {
          try {
            out.write(message);
          } catch (IOException e) {
            e.printStackTrace();
          }
        }
      });
      latch.countDown();
    }
    @Override
    public void onClose(Session sess, CloseReason closeReason) {
      session = null;
      endpoint = null;
      in = null;
    }
    @Override
    public void onError(Session session, Throwable thr) {
    }
  };
  ClientEndpointConfig clientEndpointConfig = ClientEndpointConfig.Builder.create().build();
  WebSocketContainer webSocketContainer = ContainerProvider.getWebSocketContainer();
  session = webSocketContainer.connectToServer(endpoint, clientEndpointConfig, new URI("http://localhost:8080/ws"));
  latch.await();
}
 

private void testInterruptReader(final PipedReader reader) throws Exception {
    Thread thread = interruptMeLater();
    try {
        reader.read();
        fail();
    } catch (InterruptedIOException expected) {
    } finally {
        confirmInterrupted(thread);
    }
}
 

public PReader(PipedWriter pw) {
    try {
        pr = new PipedReader(pw);
    } catch (IOException e) {
        System.out.println("Exception setting up reader: "
                + e.toString());
    }
}
 

public void test_close() throws Exception {
    PipedReader rd = new PipedReader();
    pw = new PipedWriter(rd);
    reader = new PReader(rd);
    try {
        pw.close();
    } catch (IOException e) {
        fail("Test 1: Unexpected IOException: " + e.getMessage());
    }
}
 

Pipe(Term term) throws IOException {
    pipedReader = new PipedReader();
    pipedWriter = new PipedWriter(pipedReader);

    term.addInputListener(new TermListener());
}
 

Print(PipedReader in) {
    this.in = in;
}
 

public PipedHostTelService(AppContext appContext, Writer outDataWriter) throws IOException {
    super(appContext);
    super.initConstructor(new PipedReader(this.inDataWriter), outDataWriter);
}
 

public void testInterruptPipedReader() throws Exception {
    PipedWriter writer = new PipedWriter();
    PipedReader reader = new PipedReader(writer);
    testInterruptReader(reader);
}
 

public void testInterruptPipedWriter() throws Exception {
    final PipedWriter writer = new PipedWriter();
    new PipedReader(writer);
    testInterruptWriter(writer);
}
 

public PReader(PipedReader pr) {
    this.pr = pr;
}
 

/**
 * Test sysinfo.getInfo()
 *
 * Currently only tests getInfo() by comparing the first line with the
 * expected first line in English. Because so much of sysinfo changes from
 * machine-to-machine, writing a better test may be difficult.
 *
 * Test spawns a separate thread in which to call sysinfo and feed the
 * PipedWriter. Using PipedWriter and PipedReader from the same thread
 * can cause a deadlock.
 */
public void testGetInfo() throws IOException {
    sysinfo_api_helper sah = new sysinfo_api_helper();
    sah.start();
    PipedReader pipeR = new PipedReader(sah.getPipedWriter());
    BufferedReader br = new BufferedReader(pipeR);
    assertEquals("------------------ Java Information ------------------",
                 br.readLine());
    br.close();
    pipeR.close();
}
 

/**
 * Test sysinfo.getInfo()
 *
 * Currently only tests getInfo() by comparing the first line with the
 * expected first line in English. Because so much of sysinfo changes from
 * machine-to-machine, writing a better test may be difficult.
 *
 * Test spawns a separate thread in which to call sysinfo and feed the
 * PipedWriter. Using PipedWriter and PipedReader from the same thread
 * can cause a deadlock.
 */
public void testGetInfo() throws IOException {
    sysinfo_api_helper sah = new sysinfo_api_helper();
    sah.start();
    PipedReader pipeR = new PipedReader(sah.getPipedWriter());
    BufferedReader br = new BufferedReader(pipeR);
    assertEquals("------------------ Java Information ------------------",
                 br.readLine());
    br.close();
    pipeR.close();
}
 

/**
 * Instantiates the RecordReader.
 * @param format    RDF serialization format to parse.
 * @param charBufferSize    Number of input characters to hold in
 *                          memory; if exceeded, wait until the parser
 *                          thread consumes some text before proceeding
 *                          with reading input.
 * @param statementBufferSize   Number of output statements to hold in
 *                              memory; if exceeded, wait until the
 *                              client consumes data before proceeding
 *                              with parsing.
 * @param timeoutSeconds    Number of seconds to wait for the parser
 *                          thread to provide the next statement (or
 *                          state that there are none). If exceeded,
 *                          abort.
 */
RdfFileRecordReader(RDFFormat format, int charBufferSize, int statementBufferSize, int timeoutSeconds) {
    rdfParser = Rio.createParser(format);
    rdfParser.setRDFHandler(this);
    statementCache = new LinkedBlockingQueue<RyaStatementWritable>(statementBufferSize);
    pipeOut = new PipedWriter();
    pipeIn = new PipedReader(charBufferSize);
    this.timeoutSeconds = timeoutSeconds;
    logger.info("Initializing RecordReader with parameters:");
    logger.info("\tRDF serialization format = " + format.getName());
    logger.info("\tinput buffer size = " + charBufferSize + " characters");
    logger.info("\tstatement cache size = " + statementBufferSize);
    logger.info("\tparser timeout = " + timeoutSeconds + " seconds");
}
 

/**
 * Test sysinfo.getInfo()
 *
 * Currently only tests getInfo() by comparing the first line with the
 * expected first line in English. Because so much of sysinfo changes from
 * machine-to-machine, writing a better test may be difficult.
 *
 * Test spawns a separate thread in which to call sysinfo and feed the
 * PipedWriter. Using PipedWriter and PipedReader from the same thread
 * can cause a deadlock.
 */
public void testGetInfo() throws IOException {
    sysinfo_api_helper sah = new sysinfo_api_helper();
    sah.start();
    PipedReader pipeR = new PipedReader(sah.getPipedWriter());
    BufferedReader br = new BufferedReader(pipeR);
    assertEquals("------------------ Java Information ------------------",
                 br.readLine());
    br.close();
    pipeR.close();
}