How to tail a file with rx-java?

Question

Using the rx-java package, how can I tail a file?

code:

package tail;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.logging.Logger;

public class FileUtils {

    public FileUtils() {
    }

    public void readFromInputStream(InputStream inputStream)
            throws IOException {
        StringBuilder resultStringBuilder = new StringBuilder();
        try (BufferedReader br
                = new BufferedReader(new InputStreamReader(inputStream))) {
            String line;
            while ((line = br.readLine()) != null) {
                resultStringBuilder.append(line).append("\n");
            }
        }
        System.out.println(resultStringBuilder);
    }

}

I'd want to parse each line as its written to the file. First, need an equivalent to tail -f foo.log from Java.

Did you look into using: https://commons.apache.org/proper/commons-io/apidocs/org/apache/commons/io/input/TailerListenerAdapter.html#handle-java.lang.String- — qxotk, Mar 01 '22 at 17:38

score 2 · Answer 1 · answered May 16 '20 at 11:54

I'm not sure about using RxJava specifically. But you can take a general Java solution and combine with RxJava. This of course depends on how you want to be informed.

One basic option is to tail the file using an open InputStream (or some stream wrapper classes or reader class). Using a running thread, you can wait on the file with InputStream.read or sample the amount of new bytes with InputStream.available. In RxJava you can use the IO scheduler. After finishing to read an entire line, just invoke some observers from the thread.

In core Java, it could look something like:

class ObserveTask implements Runnable {

    private File file;
    private Listener listener;

    @Override
    public void run() {
        try (InputStream stream = new FileInputStream(file); 
            BufferedReader reader = new BufferedReader(stream)) {
            while (!Thread.interrupted()) {
                String line = reader.readLine();
                listener.onNewLine(line);
            }
        }
    }
}

Now just run the task:

Scheduler scheduler = Schedulers.newThread();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(new ObserveTask());

Of course this is just a basic example.

You can split the task to parts, where each run, a check is made for the bytes, while the stream remains open. Of course this is dangerous, since leaving a stream open like that might end in leaking.

A more advanced solution can include using WatchService.

How to tail a file with rx-java?

1 Answers1