### Question:
Streams { public static Observable<String> nmeaFrom(final File file) { return Observable.using( Checked.f0(() -> new FileInputStream(file)), is -> nmeaFrom(is), is -> { try { is.close(); } catch (IOException e) { } } , true); }  static Observable<String> connect(String host, int port); static Observable<TimestampedAndLine<AisMessage>> connectAndExtract(String host,
            int port); static Observable<TimestampedAndLine<AisMessage>> extract(
            Observable<String> rawAisNmea); static Observable<TimestampedAndLines<AisMessage>> extractWithLines(
            Observable<String> rawAisNmea); static Observable<Timestamped<AisMessage>> extractMessages(
            Observable<String> rawAisNmea); static Func1<Optional<T>, Boolean> isPresent(); static Func1<Optional<T>, T> toValue(); static Transformer<Optional<T>, T> valueIfPresent(); static Observable<Fix> extractFixes(Observable<String> rawAisNmea); static Observable<String> nmeaFrom(final File file); static Observable<String> nmeaFrom(InputStream is); static Observable<String> nmeaFromGzip(String filename); static Observable<Observable<String>> nmeasFromGzip(Observable<File> files); static Observable<String> nmeaFromGzip(final File file); static void print(Observable<?> stream, final PrintStream out); static void print(Observable<?> stream); static final Func1<String, Observable<LineAndTime>> toLineAndTime(); static final Transformer<NmeaMessage, Optional<List<NmeaMessage>>> addToBuffer(
            int bufferSize); static final Transformer<NmeaMessage, NmeaMessage> aggregateMultiLineNmea(
            int bufferSize); static final Transformer<NmeaMessage, NmeaMessage> aggregateMultiLineNmeaWithLines(
            int bufferSize); static Observable<String> connectOnce(final HostPort hostPort); static Func1<List<File>, Observable<Integer>> extractFixesFromNmeaGzAndAppendToFile(
            final int linesPerProcessor, final Scheduler scheduler,
            final Func1<Fix, String> fileMapper, final int writeBufferSize,
            final Action1<File> logger); static Observable<Integer> writeFixesFromNmeaGz(File input, Pattern inputPattern,
            File output, int logEvery, int writeBufferSize, Scheduler scheduler,
            int linesPerProcessor, long downSampleIntervalMs, Func1<Fix, String> fileMapper); static void main(String[] args); static final int BUFFER_SIZE; static final Func1<String, Optional<NmeaMessage>> LINE_TO_NMEA_MESSAGE; static final Func1<NmeaMessage, Optional<Timestamped<AisMessage>>> TO_AIS_MESSAGE; static final Func1<NmeaMessage, TimestampedAndLine<AisMessage>> TO_AIS_MESSAGE_AND_LINE; static final Func1<Optional<List<NmeaMessage>>, TimestampedAndLines<AisMessage>> TO_AIS_MESSAGE_AND_LINES; }### Answer:
@Test public void testNmeaFromFile() { assertEquals(100, (int) Streams.nmeaFrom(new File("src/test/resources/nmea-timestamped.txt")).count() .toBlocking().single()); }