code: public static <T extends Serializable> Observable<T> read(final ObjectInputStream ois) {
          return Observable.create(new SyncOnSubscribe<ObjectInputStream,T>() {
  
              @Override
              protected ObjectInputStream generateState() {
                 return ois;
              }
  
              @Override
              protected ObjectInputStream next(ObjectInputStream ois, Observer<? super T> observer) {
                  try {
                      @SuppressWarnings("unchecked")
                      T t = (T) ois.readObject();
                      observer.onNext(t);
                  } catch (EOFException e) {
                      observer.onCompleted();
                  } catch (ClassNotFoundException e) {
                      observer.onError(e);
                  } catch (IOException e) {
                      observer.onError(e);
                  }
                  return ois;
              }
          });
      }
docs: Returns the deserialized objects from the given {@link InputStream} as an
  {@link Observable} stream.
  
  @param ois
  the {@link ObjectInputStream}
  @param <T>
  the generic type of the returned stream
  @return the stream of deserialized objects from the {@link InputStream}
  as an {@link Observable}.