Also emit a watch event at startup when using the builder source.

src/main/java/com/github/davidmoten/rx/FileObservable.java

@@ -51,15 +51,7 @@ public final class FileObservable {
      */
     public final static Observable<byte[]> tailFile(File file, long startPosition,
             long sampleTimeMs, int chunkSize) {
-        Observable<Object> events = from(file, StandardWatchEventKinds.ENTRY_CREATE,
-                StandardWatchEventKinds.ENTRY_MODIFY, StandardWatchEventKinds.OVERFLOW)
-                        // don't care about the event details, just that there
-                        // is one
-                        .cast(Object.class)
-                        // get lines once on subscription so we tail the lines
-                        // in the file at startup
-                        .startWith(new Object());
-        return tailFile(file, startPosition, sampleTimeMs, chunkSize, events);
+        return tailFile(file, startPosition, sampleTimeMs, chunkSize, getDefaultWatchEventSource(file, null));
     }
 
     /**
@@ -332,6 +324,17 @@ public Boolean call(Object event) {
         }
     };
 
+    private static Observable<Object> getDefaultWatchEventSource(File file, Action0 onWatchStarted) {
+        return from(file, onWatchStarted, StandardWatchEventKinds.ENTRY_CREATE,
+                StandardWatchEventKinds.ENTRY_MODIFY, StandardWatchEventKinds.OVERFLOW)
+                // don't care about the event details, just that there
+                // is one
+                .cast(Object.class)
+                // get lines once on subscription so we tail the lines
+                // in the file at startup
+                .startWith(new Object());
+    }
+
     public static Builder tailer() {
         return new Builder();
     }
@@ -452,12 +455,7 @@ public Observable<String> tailText() {
         }
 
         private Observable<?> getSource() {
-            if (source == null)
-                return from(file, onWatchStarted, StandardWatchEventKinds.ENTRY_CREATE,
-                        StandardWatchEventKinds.ENTRY_MODIFY, StandardWatchEventKinds.OVERFLOW);
-            else
-                return source;
-
+            return source != null ? source : getDefaultWatchEventSource(file, onWatchStarted);
         }
 
     }

src/test/java/com/github/davidmoten/rx/FileObservableTest.java

@@ -259,4 +259,29 @@ public void call(String line) {
         assertEquals("line 3", list.get(1).trim());
         sub.unsubscribe();
     }
+
+    @Test
+    public void testFileTailingExistingLines() throws InterruptedException, IOException {
+        final File log = new File("target/test.log");
+        log.delete();
+        log.createNewFile();
+        append(log, "a0");
+        append(log, "a1");
+        append(log, "a2");
+
+        Observable<String> tailer = FileObservable.tailer().file(log).sampleTimeMs(50).utf8().tailText();
+        final List<String> list = new ArrayList<String>();
+        final CountDownLatch latch = new CountDownLatch(3);
+        Subscription sub = tailer.subscribeOn(Schedulers.io()).subscribe(new Action1<String>() {
+            @Override
+            public void call(String line) {
+                System.out.println("received: '" + line + "'");
+                list.add(line);
+                latch.countDown();
+            }
+        });
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+        assertEquals(Arrays.asList("a0", "a1", "a2"), list);
+        sub.unsubscribe();
+    }
 }