下面列出了怎么用java.util.stream.StreamSupport的API类实例代码及写法,或者点击链接到github查看源代码。
@Override
public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<TripDuration> iterable, Collector<AverageTripDuration> collector) {
if (Iterables.size(iterable) > 1) {
String location = Iterables.get(iterable, 0).pickupGeoHash;
String airportCode = Iterables.get(iterable, 0).airportCode;
long sumDuration = StreamSupport
.stream(iterable.spliterator(), false)
.mapToLong(trip -> trip.tripDuration)
.sum();
double avgDuration = (double) sumDuration / Iterables.size(iterable);
collector.collect(new AverageTripDuration(location, airportCode, sumDuration, avgDuration, timeWindow.getEnd()));
}
}
/**
* Two step process, first given the list of instructors,
* find the coressponding rmp-id. Then using the new rmp-id,
* query the rating.
* E.g:
* For professor "Victor Shoup":
* 1. We first find the rmp-id on RMP using getLinkAsync
* -> rmp-id: 1134872
* 2. We then use the rmp-id to query the rating itself
* -> https://www.ratemyprofessors.com/ShowRatings.jsp?tid=1134872;
* @param names
* @param batchSizeNullable
* @return
*/
public static Stream<Rating> getRatings(Iterator<Instructor> names,
Integer batchSizeNullable) {
int batchSize = batchSizeNullable != null
? batchSizeNullable
: 50; // @Performance what should this number be?
// @TODO Change this to actually be correct in terms of types used
SimpleBatchedFutureEngine<Instructor, Instructor> instructorResults =
new SimpleBatchedFutureEngine<>(
names, batchSize, (instructor, __) -> getLinkAsync(instructor));
SimpleBatchedFutureEngine<Instructor, Rating> engine =
new SimpleBatchedFutureEngine<>(
instructorResults, batchSize, (instructor, __) -> {
try {
return queryRatingAsync(instructor.name, instructor.id);
} catch (Exception e) {
throw new RuntimeException(e);
}
});
return StreamSupport.stream(engine.spliterator(), false)
.filter(i -> i != null);
}
@SuppressWarnings("rawtypes")
@EventListener
public void handleContextRefresh(ContextRefreshedEvent event) {
final Environment environment = event.getApplicationContext().getEnvironment();
LOGGER.info("====== Environment and configuration ======");
LOGGER.info("Active profiles: {}", Arrays.toString(environment.getActiveProfiles()));
final MutablePropertySources sources = ((AbstractEnvironment) environment).getPropertySources();
StreamSupport.stream(sources.spliterator(), false).filter(ps -> ps instanceof EnumerablePropertySource)
.map(ps -> ((EnumerablePropertySource) ps).getPropertyNames()).flatMap(Arrays::stream).distinct()
.forEach(prop -> {
Object resolved = environment.getProperty(prop, Object.class);
if (resolved instanceof String) {
LOGGER.info("{} - {}", prop, environment.getProperty(prop));
} else {
LOGGER.info("{} - {}", prop, "NON-STRING-VALUE");
}
});
LOGGER.debug("===========================================");
}
private BlockPos argsToEntityPos(String name) throws CmdError
{
LivingEntity entity = StreamSupport
.stream(MC.world.getEntities().spliterator(), true)
.filter(e -> e instanceof LivingEntity).map(e -> (LivingEntity)e)
.filter(e -> !e.removed && e.getHealth() > 0)
.filter(e -> e != MC.player)
.filter(e -> !(e instanceof FakePlayerEntity))
.filter(e -> name.equalsIgnoreCase(e.getDisplayName().getString()))
.min(
Comparator.comparingDouble(e -> MC.player.squaredDistanceTo(e)))
.orElse(null);
if(entity == null)
throw new CmdError("Entity \"" + name + "\" could not be found.");
return new BlockPos(entity.getPos());
}
@Test
void idGenerationWithNewEntitiesShouldWork(@Autowired ThingWithGeneratedIdRepository repository) {
List<ThingWithGeneratedId> things = IntStream.rangeClosed(1, 10)
.mapToObj(i -> new ThingWithGeneratedId("name" + i))
.collect(toList());
Iterable<ThingWithGeneratedId> savedThings = repository.saveAll(things);
assertThat(savedThings)
.hasSize(things.size())
.extracting(ThingWithGeneratedId::getTheId)
.allMatch(s -> s.matches("thingWithGeneratedId-\\d+"));
Set<String> distinctIds = StreamSupport.stream(savedThings.spliterator(), false)
.map(ThingWithGeneratedId::getTheId).collect(toSet());
assertThat(distinctIds).hasSize(things.size());
}
protected void verifyDatabase(Iterable<ThingWithAssignedId> expectedValues) {
List<String> ids = StreamSupport.stream(expectedValues.spliterator(), false)
.map(ThingWithAssignedId::getTheId).collect(toList());
List<String> names = StreamSupport.stream(expectedValues.spliterator(), false)
.map(ThingWithAssignedId::getName).collect(toList());
try (Session session = driver.session()) {
Record record = session
.run("MATCH (n:Thing) WHERE n.theId in $ids RETURN COLLECT(n) as things", Values.parameters("ids", ids))
.single();
List<Node> nodes = record.get("things").asList(Value::asNode);
assertThat(nodes).extracting(n -> n.get("theId").asString()).containsAll(ids);
assertThat(nodes).extracting(n -> n.get("name").asString())
.containsAll(names);
}
}
@Test
public void contextLoads() {
LoggerContext context = (LoggerContext) LoggerFactory.getILoggerFactory();
String val = context.getProperty(SpacedLogbackSystem.KEY_ENABLED);
Assert.assertEquals("true", val);
Logger logger = context.getLogger("com.baidu");
Assert.assertEquals(Level.DEBUG, logger.getLevel());
logger = context.getLogger("com.baidu.ebiz");
Assert.assertEquals(Level.WARN, logger.getLevel());
logger = context.getLogger("ROOT");
List<Appender<ILoggingEvent>> list = StreamSupport.stream(
Spliterators.spliteratorUnknownSize(
logger.iteratorForAppenders(), Spliterator.ORDERED), false)
.collect(Collectors.toList());
Assert.assertThat(list.size(), Matchers.greaterThan(1));
LoggerFactory.getLogger("com.baidu").info("info for root");
StatusPrinter.print(context);
}
@Test // gh-22740
public void registerMultipleModulesWithNullTypeId() {
Jackson2ObjectMapperBuilder builder = new Jackson2ObjectMapperBuilder();
SimpleModule fooModule = new SimpleModule();
fooModule.addSerializer(new FooSerializer());
SimpleModule barModule = new SimpleModule();
barModule.addSerializer(new BarSerializer());
builder.modulesToInstall(fooModule, barModule);
ObjectMapper objectMapper = builder.build();
assertEquals(1, StreamSupport
.stream(getSerializerFactoryConfig(objectMapper).serializers().spliterator(), false)
.filter(s -> s.findSerializer(null, SimpleType.construct(Foo.class), null) != null)
.count());
assertEquals(1, StreamSupport
.stream(getSerializerFactoryConfig(objectMapper).serializers().spliterator(), false)
.filter(s -> s.findSerializer(null, SimpleType.construct(Bar.class), null) != null)
.count());
}
@SuppressWarnings("rawtypes")
@EventListener
public void handleContextRefresh(ContextRefreshedEvent event) {
final Environment environment = event.getApplicationContext().getEnvironment();
LOGGER.info("====== Environment and configuration ======");
LOGGER.info("Active profiles: {}", Arrays.toString(environment.getActiveProfiles()));
final MutablePropertySources sources = ((AbstractEnvironment) environment).getPropertySources();
StreamSupport.stream(sources.spliterator(), false).filter(ps -> ps instanceof EnumerablePropertySource)
.map(ps -> ((EnumerablePropertySource) ps).getPropertyNames()).flatMap(Arrays::stream).distinct()
.forEach(prop -> {
Object resolved = environment.getProperty(prop, Object.class);
if (resolved instanceof String) {
LOGGER.info("{} - {}", prop, environment.getProperty(prop));
} else {
LOGGER.info("{} - {}", prop, "NON-STRING-VALUE");
}
});
LOGGER.debug("===========================================");
}
@SuppressWarnings("rawtypes")
@EventListener
public void handleContextRefresh(ContextRefreshedEvent event) {
final Environment environment = event.getApplicationContext().getEnvironment();
LOGGER.info("====== Environment and configuration ======");
LOGGER.info("Active profiles: {}", Arrays.toString(environment.getActiveProfiles()));
final MutablePropertySources sources = ((AbstractEnvironment) environment).getPropertySources();
StreamSupport.stream(sources.spliterator(), false).filter(ps -> ps instanceof EnumerablePropertySource)
.map(ps -> ((EnumerablePropertySource) ps).getPropertyNames()).flatMap(Arrays::stream).distinct()
.forEach(prop -> {
LOGGER.info("{}", prop);
// Object resolved = environment.getProperty(prop, Object.class);
// if (resolved instanceof String) {
// LOGGER.info("{}", environment.getProperty(prop));
// }
});
LOGGER.info("===========================================");
}
@SuppressWarnings("rawtypes")
@EventListener
public void handleContextRefresh(ContextRefreshedEvent event) {
final Environment environment = event.getApplicationContext().getEnvironment();
LOGGER.info("====== Environment and configuration ======");
LOGGER.info("Active profiles: {}", Arrays.toString(environment.getActiveProfiles()));
final MutablePropertySources sources = ((AbstractEnvironment) environment).getPropertySources();
StreamSupport.stream(sources.spliterator(), false).filter(ps -> ps instanceof EnumerablePropertySource)
.map(ps -> ((EnumerablePropertySource) ps).getPropertyNames()).flatMap(Arrays::stream).distinct()
.forEach(prop -> {
LOGGER.info("{}", prop);
// Object resolved = environment.getProperty(prop, Object.class);
// if (resolved instanceof String) {
// LOGGER.info("{}", environment.getProperty(prop));
// }
});
LOGGER.info("===========================================");
}
@Test
void serializeExceptionAsArray() throws IOException {
Exception exception = new Exception("foo");
StringBuilder jsonBuilder = new StringBuilder();
jsonBuilder.append('{');
EcsJsonSerializer.serializeException(jsonBuilder, exception, true);
jsonBuilder.append('}');
System.out.println(jsonBuilder);
JsonNode jsonNode = new ObjectMapper().readTree(jsonBuilder.toString());
assertThat(jsonNode.get("error.type").textValue()).isEqualTo(exception.getClass().getName());
assertThat(jsonNode.get("error.message").textValue()).isEqualTo("foo");
StringWriter stringWriter = new StringWriter();
exception.printStackTrace(new PrintWriter(stringWriter));
assertThat(StreamSupport.stream(jsonNode.get("error.stack_trace").spliterator(), false)
.map(JsonNode::textValue)
.collect(Collectors.joining(System.lineSeparator(), "", System.lineSeparator())))
.isEqualTo(stringWriter.toString());
}
private static Stream<String> getSuggestedValues(PositionalParamSpec paramSpec) {
Iterable<String> completionCandidates = paramSpec.completionCandidates();
if (completionCandidates != null) {
return StreamSupport.stream(Spliterators.spliteratorUnknownSize(completionCandidates.iterator(), 0), false);
} else {
return getSuggestedValues(paramSpec.type());
}
}
@Override
public long approximateNumEntries() {
StreamObserverSpliterator<io.apicurio.registry.streams.distore.proto.Size> observer = new StreamObserverSpliterator<>();
stub.approximateNumEntries(
VoidReq.newBuilder()
.setStoreName(storeName)
.build(),
observer
);
return StreamSupport
.stream(observer, false)
.mapToLong(Size::getSize)
.findFirst()
.getAsLong();
}
@SuppressWarnings("unchecked")
public synchronized static <S> S load(Class<? super S> service, Class<S> implClass) {
return (S) singletonServices.getOrDefault(implClass.getCanonicalName(),
StreamSupport.
stream(ServiceLoader.load(service).spliterator(), false)
.filter(implClass::isInstance)
.map(ServiceSupport::singletonFilter)
.findFirst().orElseThrow(() -> new ServiceLoadException(implClass)));
}
Iterable<Table> listTables(DatasetId datasetId, TableDefinition.Type... types) {
Set<TableDefinition.Type> allowedTypes = ImmutableSet.copyOf(types);
Iterable<Table> allTables = bigQuery.listTables(datasetId).iterateAll();
return StreamSupport.stream(allTables.spliterator(), false)
.filter(table -> allowedTypes.contains(table.getDefinition().getType()))
.collect(toImmutableList());
}
static int countSplits(ForkJoinPool fjp) throws Exception {
// The number of splits will be equivalent to the number of leaf nodes
// and will be a power of 2
ForkJoinTask<Integer> fInteger = fjp.submit(() -> {
Spliterator<Integer> s = IntStream.range(0, 1024).boxed().parallel().spliterator();
SplitCountingSpliterator<Integer> cs = new SplitCountingSpliterator<>(s);
StreamSupport.stream(cs, true).forEach(e -> {});
return cs.splits();
});
return fInteger.get();
}
@Override
public Stream<VideoMetadata> streamVideoMetadata(VideoQuery query) {
Preconditions.checkNotNull(query.getPlaceId(), "Must specify placeid");
UUID start = start(query);
UUID end = end(query);
logger.debug("Querying recordings by: types: {} deleted: {} tags: {} cameras: {} in range [{} - {}] limit [{}]", query.getRecordingType(), query.isListDeleted(), query.getTags(), query.getCameras(), start, end, query.getLimit());
Predicate<VideoMetadata> predicate = queryPredicate(query);
Iterator<VideoRecordingSize> recordingIds = queryPlan(query, start, end);
Iterator<VideoMetadata> result = Iterators.transform(recordingIds, (r) -> fetchVideoMetadata(r, predicate));
Spliterator<VideoMetadata> stream = Spliterators.spliteratorUnknownSize(result, Spliterator.IMMUTABLE | Spliterator.NONNULL);
return StreamSupport.stream(stream, false);
}
/**
* Regex allows the following paths:
* /storage.json
* /storage/{group}-plugins.json
* Note: for the second case the group involves the leading slash, therefore it should be removed then
*/
@GET
@Path("/storage{group: (/[^/]+?)*}-plugins.json")
@Produces(MediaType.APPLICATION_JSON)
public List<PluginConfigWrapper> getConfigsFor(@PathParam("group") String pluginGroup) {
pluginGroup = StringUtils.isNotEmpty(pluginGroup) ? pluginGroup.replace("/", "") : ALL_PLUGINS;
return StreamSupport.stream(
Spliterators.spliteratorUnknownSize(storage.getStore().getAll(), Spliterator.ORDERED), false)
.filter(byPluginGroup(pluginGroup))
.map(entry -> new PluginConfigWrapper(entry.getKey(), entry.getValue()))
.sorted(PLUGIN_COMPARATOR)
.collect(Collectors.toList());
}
protected static List<Trace> loadTraces() {
Iterable<Trace> iterable = Plugins.TRACE.extensions();
if (iterable != null) {
return Arrays.asList(StreamSupport.stream(iterable.spliterator(), false).toArray(Trace[]::new));
}
return Collections.emptyList();
}
SortedTestData(List<T> coll) {
super("SortedTestData", coll,
c -> StreamSupport.stream(Spliterators.spliterator(c.toArray(), Spliterator.ORDERED | Spliterator.SORTED), false),
c -> StreamSupport.stream(Spliterators.spliterator(c.toArray(), Spliterator.ORDERED | Spliterator.SORTED), true),
c -> Spliterators.spliterator(c.toArray(), Spliterator.ORDERED | Spliterator.SORTED),
List::size);
}
@Override
public List<CellValue> getDataFromRange(String range)
{
return StreamSupport.stream(CellRangeAddress.valueOf(range).spliterator(), false)
.map(CellAddress::formatAsString)
.map(addr -> new CellValue(getDataFromCell(addr), addr))
.collect(Collectors.toList());
}
@NonNull
@Override
public <T> Stream<T> entityStream(@NonNull ResultSet resultSet, @Nullable String prefix, @NonNull Class<T> rootEntity) {
ArgumentUtils.requireNonNull("resultSet", resultSet);
ArgumentUtils.requireNonNull("rootEntity", rootEntity);
TypeMapper<ResultSet, T> mapper = new SqlResultEntityTypeMapper<>(prefix, getEntity(rootEntity), columnNameResultSetReader, jsonCodec);
Iterable<T> iterable = () -> new Iterator<T>() {
boolean nextCalled = false;
@Override
public boolean hasNext() {
try {
if (!nextCalled) {
nextCalled = true;
return resultSet.next();
} else {
return nextCalled;
}
} catch (SQLException e) {
throw new DataAccessException("Error retrieving next JDBC result: " + e.getMessage(), e);
}
}
@Override
public T next() {
nextCalled = false;
return mapper.map(resultSet, rootEntity);
}
};
return StreamSupport.stream(iterable.spliterator(), false);
}
@Override
public void call(String[] args) throws CmdException
{
if(args.length != 1)
throw new CmdSyntaxError();
FollowHack followHack = WURST.getHax().followHack;
if(followHack.isEnabled())
followHack.setEnabled(false);
Entity entity = StreamSupport
.stream(MC.world.getEntities().spliterator(), true)
.filter(e -> e instanceof LivingEntity)
.filter(e -> !e.removed && ((LivingEntity)e).getHealth() > 0)
.filter(e -> e != MC.player)
.filter(e -> !(e instanceof FakePlayerEntity))
.filter(e -> args[0].equalsIgnoreCase(e.getName().getString()))
.min(
Comparator.comparingDouble(e -> MC.player.squaredDistanceTo(e)))
.orElse(null);
if(entity == null)
throw new CmdError(
"Entity \"" + args[0] + "\" could not be found.");
followHack.setEntity(entity);
followHack.setEnabled(true);
}
/**
* Returns a stream of {@code int} zero-extending the {@code char} values
* from this sequence. Any char which maps to a <a
* href="{@docRoot}/java/lang/Character.html#unicode">surrogate code
* point</a> is passed through uninterpreted.
*
* <p>If the sequence is mutated while the stream is being read, the
* result is undefined.
*
* @return an IntStream of char values from this sequence
* @since 1.8
*/
public default IntStream chars() {
class CharIterator implements PrimitiveIterator.OfInt {
int cur = 0;
public boolean hasNext() {
return cur < length();
}
public int nextInt() {
if (hasNext()) {
return charAt(cur++);
} else {
throw new NoSuchElementException();
}
}
@Override
public void forEachRemaining(IntConsumer block) {
for (; cur < length(); cur++) {
block.accept(charAt(cur));
}
}
}
return StreamSupport.intStream(() ->
Spliterators.spliterator(
new CharIterator(),
length(),
Spliterator.ORDERED),
Spliterator.SUBSIZED | Spliterator.SIZED | Spliterator.ORDERED,
false);
}
private void assertEntities() {
var expected = expectedEntityNum.toArray();
var actual = StreamSupport.stream(entityRepository.findAll().spliterator(), false)
.map(e -> e.getEntityNum())
.toArray();
Arrays.sort(expected);
Arrays.sort(actual);
assertArrayEquals(expected, actual);
}
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
if (context.isRestored()) {
restoredState = true;
unionListState = context.getOperatorStateStore().getUnionListState(unionStateDescriptor);
Set<Integer> actualIndices = StreamSupport.stream(unionListState.get().spliterator(), false).collect(Collectors.toSet());
if (getRuntimeContext().getTaskName().contains(SINGLE_REGION_SOURCE_NAME)) {
Assert.assertTrue(CollectionUtils.isEqualCollection(EXPECTED_INDICES_SINGLE_REGION, actualIndices));
} else {
Assert.assertTrue(CollectionUtils.isEqualCollection(EXPECTED_INDICES_MULTI_REGION, actualIndices));
}
if (indexOfThisSubtask == 0) {
listState = context.getOperatorStateStore().getListState(stateDescriptor);
Assert.assertTrue("list state should be empty for subtask-0",
((List<Integer>) listState.get()).isEmpty());
} else {
listState = context.getOperatorStateStore().getListState(stateDescriptor);
Assert.assertTrue("list state should not be empty for subtask-" + indexOfThisSubtask,
((List<Integer>) listState.get()).size() > 0);
if (indexOfThisSubtask == NUM_OF_REGIONS - 1) {
index = listState.get().iterator().next();
if (index != snapshotIndicesOfSubTask.get(lastCompletedCheckpointId.get())) {
throw new RuntimeException("Test failed due to unexpected recovered index: " + index +
", while last completed checkpoint record index: " + snapshotIndicesOfSubTask.get(lastCompletedCheckpointId.get()));
}
}
}
} else {
unionListState = context.getOperatorStateStore().getUnionListState(unionStateDescriptor);
if (indexOfThisSubtask != 0) {
listState = context.getOperatorStateStore().getListState(stateDescriptor);
}
}
}
@Override
public void onUpdate()
{
mobs.clear();
Stream<MobEntity> stream =
StreamSupport.stream(MC.world.getEntities().spliterator(), false)
.filter(e -> e instanceof MobEntity).map(e -> (MobEntity)e)
.filter(e -> !e.removed && e.getHealth() > 0);
if(filterInvisible.isChecked())
stream = stream.filter(e -> !e.isInvisible());
mobs.addAll(stream.collect(Collectors.toList()));
}
@Override
public <A extends Annotation> Stream<MergedAnnotation<A>> stream(String annotationType) {
if (this.annotationFilter == FILTER_ALL) {
return Stream.empty();
}
return StreamSupport.stream(spliterator(annotationType), false);
}
/**
* Returns a stream of {@code int} zero-extending the {@code char} values
* from this sequence. Any char which maps to a <a
* href="{@docRoot}/java/lang/Character.html#unicode">surrogate code
* point</a> is passed through uninterpreted.
*
* <p>If the sequence is mutated while the stream is being read, the
* result is undefined.
*
* @return an IntStream of char values from this sequence
* @since 1.8
*/
public default IntStream chars() {
class CharIterator implements PrimitiveIterator.OfInt {
int cur = 0;
public boolean hasNext() {
return cur < length();
}
public int nextInt() {
if (hasNext()) {
return charAt(cur++);
} else {
throw new NoSuchElementException();
}
}
@Override
public void forEachRemaining(IntConsumer block) {
for (; cur < length(); cur++) {
block.accept(charAt(cur));
}
}
}
return StreamSupport.intStream(() ->
Spliterators.spliterator(
new CharIterator(),
length(),
Spliterator.ORDERED),
Spliterator.SUBSIZED | Spliterator.SIZED | Spliterator.ORDERED,
false);
}