vertx - how to read stream from executable program async

Question

I have generator g , once I run it from console , it starts writing to console output(stdout) sleep for x seconds and continue , a stream of data .

I would like my program to run g and bind to his output to java vertx application as stream input.

I would like all the reading to be done async , how can I achieve it ?

this is what I am doing :

public class InputHandler extends AbstractVerticle {

    final String command = "path";

    @Override
    public void start() throws Exception {
        Runtime r = Runtime.getRuntime();
        Process p;     // Process tracks one external native process
        BufferedReader is;  // reader for output of process
        String line;
        p = r.exec(command);
        System.out.println("In Main after exec");
         is = new BufferedReader(new InputStreamReader(p.getInputStream()));

        while ((line = is.readLine()) != null)
           try {
               System.out.println(line);
           }catch (Exception ex){
               System.out.println(ex);
           }
    }
}

and this is the exception which is thrown :

io.vertx.core.VertxException: Thread blocked
    at java.io.FileInputStream.readBytes(Native Method)
    at java.io.FileInputStream.read(FileInputStream.java:255)
    at java.io.BufferedInputStream.read1(BufferedInputStream.java:284)
    at java.io.BufferedInputStream.read(BufferedInputStream.java:345)
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284)
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178)
    at java.io.InputStreamReader.read(InputStreamReader.java:184)
    at java.io.BufferedReader.fill(BufferedReader.java:161)
    at java.io.BufferedReader.readLine(BufferedReader.java:324)
    at java.io.BufferedReader.readLine(BufferedReader.java:389)
    at io.vertx.example.InputHandler.start(InputHandler.java:27)
    at io.vertx.core.AbstractVerticle.start(AbstractVerticle.java:106)
    at io.vertx.core.impl.DeploymentManager.lambda$doDeploy$8(DeploymentManager.java:483)
    at io.vertx.core.impl.DeploymentManager$$Lambda$8/884603232.handle(Unknown Source)
    at io.vertx.core.impl.ContextImpl.lambda$wrapTask$2(ContextImpl.java:339)
    at io.vertx.core.impl.ContextImpl$$Lambda$9/154173878.run(Unknown Source)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(Thread.java:745)

Answer 1

You should be using a Worker verticle. As stated here https://vertx.io/preview/docs/vertx-core/java/#blocking_code they can handle blocking code.

For example you should deploy your verticle this way:

vertx.deployVerticle(YourVerticle.class.getName(), new DeploymentOptions().setWorker(true));

Answer 2

So, I found out how to solve this problem:

 public void start() throws Exception {
        Runtime r = Runtime.getRuntime();
        Process p = r.exec(command);
        AsyncInputStream asyncInputStream = new AsyncInputStream(vertx, vertx.nettyEventLoopGroup(), p.getInputStream());
        asyncInputStream.handler(buffer ->
                lineHandler(buffer.getString(0, buffer.length() - 1))
        );
    }

I needed to dive in and find out how to use org.wisdom.framework.vertx.AsyncInputStream, so you would need this in pom XML:

<wisdomVertxEngin.version>0.10.0</wisdomVertxEngin.version>

<dependency>
  <groupId>org.wisdom-framework</groupId>
  <artifactId>wisdom-vertx-engine</artifactId>
  <version>${wisdomVertxEngin.version}</version>
</dependency>