下面列出了io.reactivex.Flowable#generate ( ) 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
public static <T> Flowable<T> create(Callable<Connection> connectionFactory, List<Object> parameters, String sql,
Function<? super ResultSet, T> mapper) {
Callable<ResultSet> initialState = () -> {
Connection con = connectionFactory.call();
PreparedStatement ps = con.prepareStatement(sql);
// TODO set parameters
ResultSet rs = ps.executeQuery();
return rs;
};
BiConsumer<ResultSet, Emitter<T>> generator = (rs, emitter) -> {
if (rs.next()) {
emitter.onNext(mapper.apply(rs));
} else {
emitter.onComplete();
}
};
Consumer<ResultSet> disposeState = FlowableSelect::closeSilently;
return Flowable.generate(initialState, generator, disposeState);
}
@Override
public Flowable<String> getRepos(String user) {
Callable<Iterator<String>> initialState =
() -> gitHbubRepos.getRepos(user)
.stream()
.map(Repository::getName).iterator();
BiConsumer<Iterator<String>, Emitter<String>> generator =
(iterator, emitter) -> {
if (iterator.hasNext()) {
emitter.onNext(iterator.next() + " ");
} else {
emitter.onComplete();
}
};
return Flowable.generate(initialState, generator);
}
@Override
public Flowable<String> getRepos(String user) {
Callable<Iterator<String>> initialState =
() -> gitHbubRepos.getRepos(user)
.stream()
.map(Repository::getName).iterator();
BiConsumer<Iterator<String>, Emitter<String>> generator =
(iterator, emitter) -> {
if (iterator.hasNext()) {
emitter.onNext(iterator.next() + " ");
} else {
emitter.onComplete();
}
};
return Flowable.generate(initialState, generator);
}
private static <T> Flowable<T> create(NamedPreparedStatement ps, List<Object> parameters,
Function<? super ResultSet, T> mapper) {
Callable<ResultSet> initialState = () -> {
Util.convertAndSetParameters(ps.ps, parameters, ps.names);
ps.ps.execute();
return ps.ps.getGeneratedKeys();
};
BiConsumer<ResultSet, Emitter<T>> generator = (rs, emitter) -> {
if (rs.next()) {
emitter.onNext(mapper.apply(rs));
} else {
emitter.onComplete();
}
};
Consumer<ResultSet> disposer = Util::closeSilently;
return Flowable.generate(initialState, generator, disposer);
}
private static <T> Flowable<T> createFlowable(NamedCallableStatement stmt,
Function<? super ResultSet, ? extends T> f) throws SQLException {
ResultSet rsActual = stmt.stmt.getResultSet();
Callable<ResultSet> initialState = () -> rsActual;
BiConsumer<ResultSet, Emitter<T>> generator = (rs, emitter) -> {
log.debug("getting row from ps={}, rs={}", stmt.stmt, rs);
if (rs.next()) {
T v = f.apply(rs);
log.debug("emitting {}", v);
emitter.onNext(v);
} else {
log.debug("completed");
emitter.onComplete();
}
};
Consumer<ResultSet> disposeState = Util::closeSilently;
return Flowable.generate(initialState, generator, disposeState);
}
/**
* Returns a Flowable stream of byte arrays from the given
* {@link InputStream} between 1 and {@code bufferSize} bytes.
*
* @param is
* input stream of bytes
* @param bufferSize
* max emitted byte array size
* @return a stream of byte arrays
*/
public static Flowable<byte[]> from(final InputStream is, final int bufferSize) {
return Flowable.generate(new Consumer<Emitter<byte[]>>() {
@Override
public void accept(Emitter<byte[]> emitter) throws Exception {
byte[] buffer = new byte[bufferSize];
int count = is.read(buffer);
if (count == -1) {
emitter.onComplete();
} else if (count < bufferSize) {
emitter.onNext(Arrays.copyOf(buffer, count));
} else {
emitter.onNext(buffer);
}
}
});
}
public static Flowable<ZippedEntry> unzip(final ZipInputStream zis) {
return Flowable.generate(new Consumer<Emitter<ZippedEntry>>() {
@Override
public void accept(Emitter<ZippedEntry> emitter) throws IOException {
ZipEntry zipEntry = zis.getNextEntry();
if (zipEntry != null) {
emitter.onNext(new ZippedEntry(zipEntry, zis));
} else {
// end of stream so eagerly close the stream (might not be a
// good idea since this method did not create the zis
zis.close();
emitter.onComplete();
}
}
});
}
public <T> Flowable<T> read(final Class<T> cls, final Input input) {
return Flowable.generate(new Consumer<Emitter<T>>() {
@Override
public void accept(Emitter<T> emitter) throws Exception {
if (input.eof()) {
emitter.onComplete();
} else {
T t = kryo.readObject(input, cls);
emitter.onNext(t);
}
}
});
}
private static <T> Flowable<T> create(PreparedStatement ps, Function<? super ResultSet, T> mapper) {
Callable<ResultSet> initialState = () -> {
ps.execute();
return ps.getGeneratedKeys();
};
BiConsumer<ResultSet, Emitter<T>> generator = (rs, emitter) -> {
if (rs.next()) {
emitter.onNext(mapper.apply(rs));
} else {
emitter.onComplete();
}
};
Consumer<ResultSet> disposer = Update::closeAll;
return Flowable.generate(initialState, generator, disposer);
}
@Override
public Flowable<Repository> getRepos0(String user) {
Callable<Iterator<Repository>> initialState =
gitHbubRepos.getRepos(user)
.stream()::iterator;
BiConsumer<Iterator<Repository>, Emitter<Repository>> generator =
(iterator, emitter) -> {
if (iterator.hasNext()) {
emitter.onNext(iterator.next());
} else {
emitter.onComplete();
}
};
return Flowable.generate(initialState, generator);
}
@Outgoing("temperature-values")
public Flowable<String> generate() {
return Flowable.generate(e -> {
int i = counter.getAndIncrement();
if (i == 3) {
e.onComplete();
} else {
e.onNext(Integer.toString(i));
}
});
}
@Override
public Flowable<Repository> getRepos0(String user) {
Callable<Iterator<Repository>> initialState =
gitHbubRepos.getRepos(user)
.stream()::iterator;
BiConsumer<Iterator<Repository>, Emitter<Repository>> generator =
(iterator, emitter) -> {
if (iterator.hasNext()) {
emitter.onNext(iterator.next());
} else {
emitter.onComplete();
}
};
return Flowable.generate(initialState, generator);
}
private static Flowable<Flowable<byte[]>> createServerSocketFlowable(final ServerSocket serverSocket,
final long timeoutMs, final int bufferSize, final Action preAcceptAction,
final Predicate<? super Socket> acceptSocket) {
return Flowable.generate( //
new Consumer<Emitter<Flowable<byte[]>>>() {
@Override
public void accept(Emitter<Flowable<byte[]>> emitter) throws Exception {
acceptConnection(timeoutMs, bufferSize, serverSocket, emitter, preAcceptAction, acceptSocket);
}
});
}
public static Flowable<String> from(final Reader reader, final int bufferSize) {
return Flowable.generate(new Consumer<Emitter<String>>() {
final char[] buffer = new char[bufferSize];
@Override
public void accept(Emitter<String> emitter) throws Exception {
int count = reader.read(buffer);
if (count == -1) {
emitter.onComplete();
} else {
emitter.onNext(String.valueOf(buffer, 0, count));
}
}
});
}
@Override
public Publisher<? extends Integer> call() throws Exception {
return Flowable.generate(new Consumer<Emitter<Integer>>() {
final int[] count = new int[1];
@Override
public void accept(Emitter<Integer> emitter) throws Exception {
count[0]++;
emitter.onNext(count[0]);
if (count[0] == 1000) {
emitter.onComplete();
}
}
});
}
private Flowable<EntityStateMachine<?, Id>> process(ClassId<?, Id> classId, Event<?> event,
Worker worker) {
EntityStateMachine<?, Id> machine = getStateMachine(classId.cls(), classId.id());
TransitionHandler handler = new TransitionHandler(classId, event, worker, machine);
return Flowable.generate(handler, handler);
}
@Override
public <X> Flowable fromCompletionStage(CompletionStage<X> cs) {
return Flowable.generate(emitter -> ObservableConverter.toStreamEvents(cs, emitter));
}
private static <T> Flowable<? extends T> create(PreparedStatement ps, List<Object> parameters,
Function<? super ResultSet, T> mapper, List<String> names, String sql, int fetchSize, int queryTimeoutSec) {
log.debug("parameters={}", parameters);
log.debug("names={}", names);
Callable<ResultSet> initialState = () -> {
List<Parameter> params = Util.toParameters(parameters);
boolean hasCollection = params.stream().anyMatch(x -> x.isCollection());
final PreparedStatement ps2;
if (hasCollection) {
// create a new prepared statement with the collection ? substituted with
// ?s to match the size of the collection parameter
ps2 = Util.prepare(ps.getConnection(), fetchSize, sql, params, queryTimeoutSec);
// now wrap the rs to auto close ps2 because it is single use (the next
// collection parameter may have a different ordinality so we need to build
// a new PreparedStatement with a different number of question marks
// substituted
return new ResultSetAutoClosesStatement(Util //
.setParameters(ps2, params, names) //
.executeQuery(), ps2);
} else {
// use the current prepared statement (normal re-use)
ps2 = ps;
return Util //
.setParameters(ps2, params, names) //
.executeQuery();
}
};
BiConsumer<ResultSet, Emitter<T>> generator = (rs, emitter) -> {
log.debug("getting row from ps={}, rs={}", rs.getStatement(), rs);
if (rs.next()) {
T v = mapper.apply(rs);
log.debug("emitting {}", v);
emitter.onNext(v);
} else {
log.debug("completed");
emitter.onComplete();
}
};
Consumer<ResultSet> disposeState = Util::closeSilently;
return Flowable.generate(initialState, generator, disposeState);
}