davidmoten/rxjava2-file

rxjava2-file

Status: released to Maven Central

Requires Java 8+.

Flowable utilities for files:

• tail a file (either lines or byte[])
• trigger tail updates using Java 8 and later NIO WatchService events
• or trigger tail updates using any Flowable
• stream WatchEvents from a WatchService
• tested on Linux, OSX (notes), Windows (notes)
• Note that WatchService is problematic on OSX and Windows (see notes below) so your best bet is Linux!

Maven site reports are here including javadoc.

Getting started

Add this maven dependency to your pom.xml:

<dependency>
  <groupId>com.github.davidmoten</groupId>
  <artifactId>rxjava2-file</artifactId>
  <version>VERSION_HERE</version>
</dependency>

How to build

git clone https://github.com/davidmoten/rxjava2-file
cd rxjava2-file
mvn clean install 

Examples

Tail a text file with NIO

Tail the lines of the text log file /var/log/server.log as a Flowable<String>:

import com.github.davidmoten.rx2.file.Files;

Flowable<String> lines = 
     Files.tailLines("/var/log/server.log")
          .nonBlocking()
          .pollingInterval(500, TimeUnit.MILLISECONDS, Schedulers.io())
          // set a private sun modifier that improves OSX responsiveness
          .modifier(SensitivityWatchEventModifier.HIGH)
          .startPosition(0)
          .chunkSize(8192)
          .utf8()
          .build();

or, using defaults of startPosition 0, chunkSize 8192, charset UTF-8, scheduler Schedulers.io():

Flowable<String> items = 
     Files.tailLines("/var/log/server.log").nonBlocking().build();
	  

Tail a text file without NIO

The above example uses a WatchService to generate WatchEvents to prompt rereads of the end of the file to perform the tail.

To use polling without a WatchService (say every 5 seconds):

Flowable<String> items = 
  Files.tailLines("/var/log/server.log")
       .events(Flowable.interval(5, TimeUnit.SECONDS))
       .build();

Tail a binary file with NIO

Flowable<byte[]> items = 
  Files.tailBytes("/tmp/dump.bin").blocking().build();

Tail a binary file without NIO

Flowable<byte[]> items = 
  Files.tailBytes("/tmp/dump.bin")
       .events(Flowable.interval(5, TimeUnit.SECONDS))
       .build();

Stream WatchService events for a file

Flowable<WatchEvent<?>> events = 
  Files
    .watch(file)
    .nonBlocking()
    .scheduler(Schedulers.io())
    .pollInterval(1, TimeUnit.MINUTES)
    .build();

Backpressure

When tailLines or tailBytes is used a conversion to Flowable occurs on the WatchEvent stream. This is desirable to handle large amounts of data being tailed in combination with a slow processor (e.g. a network call). The default strategy is BUFFER but the strategy is specifiable in the tailLines and tailBytes builders.

Non-blocking and blocking

Two alternatives are supported by the library for getting file change events from a WatchService. The nonBlocking() builder methods configure the stream to use events via WatchService.poll which is a non-blocking call (but may involve some I/O?). The blocking() builder methods configure the stream to use events via WatchService.take which is a blocking call.

So when specify nonBlocking() you end up with a stream that is asynchronous and blocking() gives you a synchronous stream (everything happens on the current thread unless of course you add asynchrony to the returned Flowable).

OSX

Apparently the WatchService can be slow on OSX (see here). Note that the first example above shows how to pass a special WatchEvent.Modifier which some find has a beneficial effect. Without that the WatchService can take >10 seconds to detect changes to the file system.

Windows

Detecting changes to files on Windows also seems problematic. See https://stackoverflow.com/questions/24306875/watchservice-in-windows-7-does-not-work.

rxjava2-file's Issues

DON

Confusing behavior regarding start position 0

I'm currently using the following "tail" configuration

Files
        .tailLines(file)
        .blocking()
        .startPosition(startPosition)
        .backpressureStrategy(BackpressureStrategy.BUFFER)
        .build());

Where startPosition is either 0 or the length of the file. The problem I'm having is with startingPosition=0. I'd expect the Flowable to immediately begin emitting lines from the beginning of the file up into its end, and then start a "real" tail.

What instead happens is that as soon as the file changes (for example, lines are appended), lines from the beginning are emitted.

Isn't this a bit counterintuitive? If the file changes, I'd expect events that reflect this change, not lines from the beginning of the file.

Readme code seems to not run

Copy and pasting from the readme leads to this type error.
incompatible types: io.reactivex.Flowable<java.lang.Long> cannot be converted to io.reactivex.Observable<java.nio.file.WatchEvent<?>>

Am I missing something?

I was able to get it to work using .events by using Observable.interval and then mapping it to a custom WatchEvent. (it wasn't happy with returning StandardWatchEventKinds.* but still this somehow seems incongruent with your example.

Does this library handle backpressure?

I see you're using interval to poll for file changes (which, by the way, makes little to me: a file watcher is supposed to inform you (via push!) of changes, not the other way round). This does have implications on backpressure, as I've read. Does this library take care of this?