public static Observable<Fix> from(final File file, BinaryFixesFormat format) {

        Func0<InputStream> resourceFactory = new Func0<InputStream>() {

            @Override
            public InputStream call() {
                try {
                    if (file.getName().endsWith(".gz"))
                        return new GZIPInputStream(new FileInputStream(file));
                    else
                        return new FileInputStream(file);
                } catch (FileNotFoundException e) {
                    throw new RuntimeException(e);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        };

        Func1<InputStream, Observable<Fix>> obsFactory = new Func1<InputStream, Observable<Fix>>() {

            @Override
            public Observable<Fix> call(InputStream is) {
                Optional<Integer> mmsi;
                if (format == BinaryFixesFormat.WITH_MMSI)
                    mmsi = Optional.absent();
                else
                    mmsi = Optional.of(BinaryFixesUtil.getMmsi(file));

                return Observable.create(new BinaryFixesOnSubscribeWithBackp(is, mmsi, format));
            }
        };
        Action1<InputStream> disposeAction = new Action1<InputStream>() {

            @Override
            public void call(InputStream is) {
                try {
                    is.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        };
        return Observable.using(resourceFactory, obsFactory, disposeAction, true);

    }
public static Observable < Fix > from ( final File file , BinaryFixesFormat format ) { Func0 < InputStream > resourceFactory = new Func0 < InputStream > ( ) { @ Override public InputStream call ( ) { try { if ( file . getName ( ) . endsWith ( ".gz" ) ) return new GZIPInputStream ( new FileInputStream ( file ) ) ; else return new FileInputStream ( file ) ; } catch ( FileNotFoundException e ) { throw new RuntimeException ( e ) ; } catch ( IOException e ) { throw new RuntimeException ( e ) ; } } } ; Func1 < InputStream , Observable < Fix > > obsFactory = new Func1 < InputStream , Observable < Fix > > ( ) { @ Override public Observable < Fix > call ( InputStream is ) { Optional < Integer > mmsi ; if ( format == BinaryFixesFormat . WITH_MMSI ) mmsi = Optional . absent ( ) ; else mmsi = Optional . of ( BinaryFixesUtil . getMmsi ( file ) ) ; return Observable . create ( new BinaryFixesOnSubscribeWithBackp ( is , mmsi , format ) ) ; } } ; Action1 < InputStream > disposeAction = new Action1 < InputStream > ( ) { @ Override public void call ( InputStream is ) { try { is . close ( ) ; } catch ( IOException e ) { throw new RuntimeException ( e ) ; } } } ; return Observable . using ( resourceFactory , obsFactory , disposeAction , true ) ; }
Returns stream of fixes from the given file. If the file name ends in
'.gz' then the file is unzipped before being read.

@param file
@return fixes stream