close input stream with Observable.using