Java 类java.util.stream.Stream 实例源码

void initJar() throws IOException {
    String testClassPath = System.getProperty("test.class.path", "");
    Path jarsrc = Stream.of(testClassPath.split(File.pathSeparator))
            .map(Paths::get)
            .filter(e -> e.endsWith("jarsrc"))
            .findAny()
            .orElseThrow(() -> new InternalError("jarsrc not found"));
    jarPath = Paths.get("lib.jar");
    try (JarOutputStream out = new JarOutputStream(Files.newOutputStream(jarPath))) {
        String[] classNames = {"Lib1.class", "Lib2.class"};
        for (String cn : classNames) {
            out.putNextEntry(new JarEntry("lib/" + cn));
            Path libClass = jarsrc.resolve("lib").resolve(cn);
            out.write(Files.readAllBytes(libClass));
        }
    }
}

private Stream<File> searchDir(File dir, boolean warnOnBadDir) {
  if (dir == null || !dir.exists() || !dir.isDirectory()) {
    if (warnOnBadDir && dir != null) {
      getLog().warn("Directory does not exist or is not a directory: " + dir);
    }
    return Stream.empty();
  }
  getLog().debug("Adding directory " + dir);
  DirectoryScanner ds = new DirectoryScanner();
  ds.setBasedir(dir);
  ds.setIncludes(includes != null && includes.length > 0 ? includes : DEFAULT_INCLUDES);
  ds.setExcludes(excludes);
  ds.addDefaultExcludes();
  ds.setCaseSensitive(false);
  ds.setFollowSymlinks(false);
  ds.scan();
  return Stream.of(ds.getIncludedFiles()).map(filename -> new File(dir, filename)).parallel();
}

@Test
public void testMapPartialNull() throws Exception {
    repeat(3, i -> {
        Map<Object, Object> m1 = Stream.of("a" + i, "b" + i, "c" + i).collect(toMap(o -> o, o -> o + "test"));

        when(rpc1.map(or(eq(singletonMap(i + 1, i)), eq(singletonMap(i + 2, i))))).thenReturn(m1);
        when(rpc2.map(or(eq(singletonMap(i + 1, i)), eq(singletonMap(i + 2, i))))).thenReturn(null);

        Map<Object, Object> args = new HashMap<>();
        args.put(i + 1, i);
        args.put(i + 2, i);

        Map<Object, Object> result = client.map(args);

        assertEquals(m1, result);

        verify(rpc1).map(or(eq(singletonMap(i + 1, i)), eq(singletonMap(i + 2, i))));
        verify(rpc2).map(or(eq(singletonMap(i + 1, i)), eq(singletonMap(i + 2, i))));

        verifyNoMoreInteractions(rpc1, rpc2);
        reset(rpc1, rpc2);
    });
}

@Override
public void execute(Tuple tuple) {
    final String data = tuple.getString(0);
    LOGGER.debug("Processing datapoint", data);
    try {
        Datapoint datapoint = MAPPER.readValue(data, Datapoint.class);
        if (isUpdateRequired(datapoint)) {
            addDatapoint(datapoint);

            List<Object> stream = Stream.of(datapoint.getMetric(), datapoint.getTime(), datapoint.getValue(),
                    datapoint.getTags())
                    .collect(Collectors.toList());

            LOGGER.debug("emit: " + stream);
            collector.emit(stream);
        }
    } catch (IOException e) {
        LOGGER.error("Failed read datapoint", e);
    } finally {
        collector.ack(tuple);
    }
}

@Test(dataProvider = "styles")
public void testPlusScalar(DataFrameAlgebra.Lib lib, boolean parallel) {
    DataFrameAlgebra.LIBRARY.set(lib);
    Stream.of(int.class, long.class, double.class).forEach(type -> {
        Array.of(20, 77, 95, 135, 233, 245).forEach(count -> {
            final DataFrame<Integer,Integer> frame = random(120, count, parallel, type);
            final DataFrame<Integer,Integer> result = frame.plus(25);
            assertEquals(result, v -> {
                switch (ArrayType.of(type)) {
                    case INTEGER:   return frame.data().getInt(v.rowOrdinal(), v.colOrdinal()) + 25;
                    case LONG:      return frame.data().getLong(v.rowOrdinal(), v.colOrdinal()) + 25;
                    default:        return frame.data().getDouble(v.rowOrdinal(), v.colOrdinal()) + 25d;
                }
            });
        });
    });
}

@Override
public SharedInterProcessLock getLock(String application, EntityType entityType) {
    if (! connector.isConnected()) {
        throw new RedirectorDataSourceException("No connection to zookeeper");
    }

    String lockPath = Stream
        .of(REDIRECTOR_ZOOKEEPER_PATH, SERVICES_PATH, application, REDIRECTOR_LOCK_PATH, entityType.getPath())
        .collect(Collectors.joining(DELIMETER, DELIMETER, ""));

    if (lockMap.containsKey(lockPath)) {
        return lockMap.get(lockPath);
    } else {
        SharedInterProcessLock lock = connector.createLock(lockPath);
        lockMap.put(lockPath, lock);
        return lock;
    }
}

@Test
public void verifySaveAttributeReleasePolicyAllowedAttrRulesAndFilter() {
    final RegexRegisteredService r = new RegexRegisteredService();
    r.setName("testSaveAttributeReleasePolicyAllowedAttrRulesAndFilter");
    r.setServiceId(SERVICE_ID);
    r.setTheme("testtheme");
    r.setEvaluationOrder(1000);
    r.setAccessStrategy(new DefaultRegisteredServiceAccessStrategy(true, false));
    r.setProxyPolicy(new RegexMatchingRegisteredServiceProxyPolicy("https://.+"));
    r.setRequiredHandlers(Stream.of("h1", "h2").collect(Collectors.toSet()));

    final ReturnAllowedAttributeReleasePolicy policy = new ReturnAllowedAttributeReleasePolicy();
    policy.setAllowedAttributes(Arrays.asList("1", "2", "3"));
    r.setAttributeReleasePolicy(policy);
    r.getAttributeReleasePolicy().setAttributeFilter(new RegisteredServiceRegexAttributeFilter("\\w+"));

    final RegisteredService r2 = this.dao.save(r);
    final RegisteredService r3 = this.dao.findServiceById(r2.getId());

    assertEquals(r, r2);
    assertEquals(r2, r3);
    assertNotNull(r3.getAttributeReleasePolicy());
    assertEquals(r2.getAttributeReleasePolicy(), r3.getAttributeReleasePolicy());
}

public QuadBuilder mirror() {
    if(facingMap.containsKey(last)) {
        Stream<Vector3> stream = Arrays.stream(facingMap.get(last).getVectors());
        switch(last) {
            case DOWN:
            case UP:
                stream.forEach(vec -> {
                    vec.subtract(0.5D).rotate(EnumFacing.Axis.Y, 180).add(0.5D);
                });
                break;
            case NORTH:
            case SOUTH:
                stream.forEach(vec -> {
                    vec.subtract(0.5D).rotate(EnumFacing.Axis.X, 180).add(0.5D);
                });
                break;
            case EAST:
            case WEST:
                stream.forEach(vec -> {
                    vec.subtract(0.5D).rotate(EnumFacing.Axis.Z, 180).add(0.5D);
                });
                break;
        }
    }
    return this;
}

@Test
public void findsAllTheBeatles() {
    PerformanceFixed stub = new PerformanceFixed() {
        @Override
        public String getName() {
            throw new UnsupportedOperationException();
        }

        @Override
        public Stream<Artist> getMusicians() {
            return Stream.of(SampleData.theBeatles);
        }
    };

    List<Artist> allMusicians = stub.getAllMusicians().collect(toList());
    assertThat(allMusicians, hasItem(SampleData.theBeatles));
    // There really must be a better way than this
    assertThat(allMusicians, hasItems(SampleData.membersOfTheBeatles.toArray(new Artist[0])));
}

@Test
public void should_create_the_correct_stream_on_groupingBy_then_all_max_by_value_and_a_comparator_for_an_empty_stream() {

    // Given
    Stream<String> strings = Stream.empty();

    Collector<String, ?, Set<Map.Entry<String, Long>>> collector =
            CollectorsUtils.groupingByAndAllMaxByValue(
                    Function.<String>identity(),
                    counting(),
                    Comparator.reverseOrder()
            );

    // When
    Set<Map.Entry<String, Long>> result = strings.collect(collector);

    // Then
    assertThat(result).isEmpty();
}

private List<Label> getLabels(final Description result) {
    return Stream.of(
            getLabels(result, Epic.class, ResultsUtils::createLabel),
            getLabels(result, Feature.class, ResultsUtils::createLabel),
            getLabels(result, Story.class, ResultsUtils::createLabel),
            getLabels(result, Severity.class, ResultsUtils::createLabel),
            getLabels(result, Owner.class, ResultsUtils::createLabel),
            getLabels(result, Tag.class, this::createLabel)
    ).reduce(Stream::concat).orElseGet(Stream::empty).collect(Collectors.toList());
}

@Test
public void canCollectPairsIntoMap() {

    Map<String, String> expectedMap = new HashMap<>();
    expectedMap.put("hello", "world");
    expectedMap.put("six of one", "half a dozen of the other");

    Map<String, String> actualMap = Stream.of(
        entry("hello", "world"),
        entry("six of one", "half a dozen of the other")
    ).collect(toMap());

    assertThat(actualMap, is(expectedMap));
}

public void testToImmutableTableMergingNullColumnKey() {
  Collector<Cell<String, String, Integer>, ?, ImmutableTable<String, String, Integer>> collector =
      ImmutableTable.toImmutableTable(Cell::getRowKey, t -> null, Cell::getValue, Integer::sum);
  try {
    Stream.of(Tables.immutableCell("one", "uno", 1)).collect(collector);
    fail("Expected NullPointerException");
  } catch (NullPointerException expected) {
  }
}

private <A extends Annotation> void checkInjectedContext(InjectionDevice device, Class<A> annotationClass, boolean includeCommonContext, Class<?>... expectedContextClasses) {
    Set<Object> injectedContext = device.getMethodContext(annotationClass);
    assertNotNull(injectedContext);

    // stream the total set of expected classes, filter out those found in context - if the resulting stream isn't empty we're missing something
    final Predicate<Class<?>> contextHasInstance = klass -> injectedContext.stream().anyMatch(obj -> klass.isInstance(obj));
    Stream<Class<?>> expectedClassStream = Arrays.stream(expectedContextClasses);
    if (includeCommonContext) {
        expectedClassStream = Stream.concat(expectedClassStream, Arrays.stream(COMMON_CONTEXT_CLASSES));
    }
    Optional<Class<?>> missingExpectedContext = expectedClassStream.filter(klass -> !contextHasInstance.test(klass)).findFirst();
    if (missingExpectedContext.isPresent()) {
        Assert.fail("Context missing expected instance of: " + missingExpectedContext.get());
    }
}

static Stream<String> dumpResults(
    Stream<Map.Entry<int[], Map<String, String>>> results
) {
    return results
        .map(le ->
                 IntStream.of(le.getKey())
                          .mapToObj(String::valueOf)
                          .collect(joining(","))
                 + "#" +
                 le.getValue().entrySet().stream()
                   .map(e -> e.getKey() + "=" + e.getValue())
                   .collect(joining("|"))
        );
}

@Test
public void groupsOfTwoHangingMan() {
    final List<Pattern> patterns =
            PROCESSOR.scan(Stream.of(buildCandlestick(10.0, 2.0, 16.0, 2.0, BodyType.HAMMER, TrendDirection.UPPER),
                                     buildCandlestick(9.0, 8.0, 10, 1.0, BodyType.SHORT, TrendDirection.UPPER)))
                    .collect(toList());
    assertEquals(1, patterns.size());
    assertEquals(PatternType.HANGING_MAN, patterns.get(0).getPatternType());
}

@Override public Spliterator<T> trySplit() {
  Spliterator<? extends Stream<? extends T>> split = blocks.trySplit();
  if (split == null) return null;
  Spliterator<T> result = new FlattenedSpliterator<T>(split, currentBlock);
  currentBlock = null;
  return result;
}

public void testToImmutableTableConflict() {
  Collector<Cell<String, String, Integer>, ?, ImmutableTable<String, String, Integer>> collector =
      ImmutableTable.toImmutableTable(Cell::getRowKey, Cell::getColumnKey, Cell::getValue);
  try {
    Stream.of(Tables.immutableCell("one", "uno", 1), Tables.immutableCell("one", "uno", 2))
        .collect(collector);
    fail("Expected IllegalArgumentException");
  } catch (IllegalArgumentException expected) {
  }
}

/**
 * Remove this dungeon.
 */
public void remove() {
    if (getWorld() == null) {
        Dungeons.getDungeons().remove(this); // Somehow this didn't unload, do it again.
        return; // Don't unload dungeon twice.
    }

    // Remove players and unload puzzles.
    Core.logInfo("Removing dungeon " + getWorld().getName() + ".");
    getScheduler().cancelAll(); // Stop any dungeon-tasks.
    removePlayers(); // Remove all players.
    getPuzzles().forEach(Puzzle::onDungeonRemove); // Disable puzzles.

    // Handle saving.
    if (isEditMode()) {
        getWorld().getEntities().stream().filter(e -> e instanceof Monster || e instanceof Item).forEach(Entity::remove);
        Stream.of(getWorld().getLoadedChunks()).forEach(Chunk::unload);
        getWorld().save(); // Save data such as Spawn Location
    }

    Bukkit.unloadWorld(getWorld(), isEditMode()); // Unload the world.
    if (isEditMode()) { // Zip up and save the world.
        Core.alertStaff("Saving modified " + getType().name() + " dungeon.");
        ZipUtil.zip(getWorld().getWorldFolder(), getType().getWorld().getPath());
    }

    Utils.removeFile(getWorld().getName()); // Delete the world folder.
    Utils.removeFile("plugins/WorldGuard/worlds/" + getWorld().getName()); // Delete WorldGuard residue.
    Dungeons.getDungeons().remove(this); // Remove this dungeon.
    this.world = null; // Mark this dungeon as unloaded.
}

@Test
@SuppressWarnings("unchecked")
public void parseReturnsLeftIfInputStreamReturnsNullArray() {
    Stream<String> tokens = mock(Stream.class);
    when(tokens.toArray(any())).thenReturn(null);

    Either<String, Pair<Integer, String>> parseResult =
            parser().parse(tokens);

    assertNotNull(parseResult);
    assertTrue(parseResult.isLeft());
}

@Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)
public void testOps(String name, TestData.OfRef<Integer> data) {
    exerciseOpsInt(data,
                   s -> Stream.concat(s, data.stream()),
                   s -> IntStream.concat(s, data.stream().mapToInt(Integer::intValue)),
                   s -> LongStream.concat(s, data.stream().mapToLong(Integer::longValue)),
                   s -> DoubleStream.concat(s, data.stream().mapToDouble(Integer::doubleValue)));
}

@Test
public void handling_multiple_exceptions_with_result_example() {

    List<Integer> customerValues = Stream.of("Bob", "Bill")
        .map(tryTo(this::findCustomerByName))
        .peek(onSuccessDo(this::sendEmailUpdateTo))
        .map(onSuccessTry(Customer::calculateValue))
        .map(recover(ifType(NoCustomerWithThatName.class, error -> {
            log("Customer not found :(");
            return -1;
        })))
        .map(recover(ifType(IOException.class, error -> -2)))
        .map(ifFailed(__ -> -127))
        .collect(toList());
}

private List<SourceFile> getSourceFiles(Path basePath) throws IOException {
    try (Stream<Path> stream = Files.walk(basePath)) {
        return stream.filter(path -> path.getFileName().toString().endsWith(".js"))
            .map(path -> new FileSystemSourceFile(basePath, basePath.relativize(path)))
            .collect(Collectors.toList());
    }
}

public void testForEachPair_parallel() {
  Stream<String> streamA = IntStream.range(0, 100000).mapToObj(String::valueOf).parallel();
  Stream<Integer> streamB = IntStream.range(0, 100000).mapToObj(i -> i).parallel();

  AtomicInteger count = new AtomicInteger(0);
  Streams.forEachPair(
      streamA,
      streamB,
      (a, b) -> {
        count.incrementAndGet();
        Truth.assertThat(a.equals(String.valueOf(b))).isTrue();
      });
  Truth.assertThat(count.get()).isEqualTo(100000);
  // of course, this test doesn't prove that anything actually happened in parallel...
}

/**
 * Zips two streams together using the zipper function, resulting in a single stream of
 * items from each stream combined using the provided function.
 *
 * The resultant stream will have the length of the shorter of the two input streams.
 *
 * Sourced from https://stackoverflow.com/questions/17640754/zipping-streams-using-jdk8-with-lambda-java-util-stream-streams-zip
 */
static <A , B, C> Stream<C> zip(Stream<A> a, Stream<B> b, BiFunction<A, B, C> zipper) {
    Objects.requireNonNull(zipper);
    Spliterator<? extends A> aSpliterator = Objects.requireNonNull(a).spliterator();
    Spliterator<? extends B> bSpliterator = Objects.requireNonNull(b).spliterator();

    // Zipping looses DISTINCT and SORTED characteristics
    int characteristics = aSpliterator.characteristics() & bSpliterator.characteristics() &
        ~(Spliterator.DISTINCT | Spliterator.SORTED);

    long zipSize = ((characteristics & Spliterator.SIZED) != 0)
        ? Math.min(aSpliterator.getExactSizeIfKnown(), bSpliterator.getExactSizeIfKnown())
        : -1;

    Iterator<A> aIterator = Spliterators.iterator(aSpliterator);
    Iterator<B> bIterator = Spliterators.iterator(bSpliterator);
    Iterator<C> cIterator = new Iterator<C>() {
        @Override
        public boolean hasNext() {
            return aIterator.hasNext() && bIterator.hasNext();
        }

        @Override
        public C next() {
            return zipper.apply(aIterator.next(), bIterator.next());
        }
    };

    Spliterator<C> split = Spliterators.spliterator(cIterator, zipSize, characteristics);
    return (a.isParallel() || b.isParallel())
        ? StreamSupport.stream(split, true)
        : StreamSupport.stream(split, false);
}

public long countRecsInFile(String empFile) throws IOException{
long numWords = 0;
Stream<String> content = null;
Path path = Paths.get(empFile);
    content = Files.lines(path, Charset.defaultCharset());
numWords = content
        .map(line -> Arrays.stream(line.split(" ")))
        .count();
        content.close();
        return numWords;
  }

private void runClasses(Class<?>... classes) {
    final ClassSelector[] classSelectors = Stream.of(classes)
            .map(DiscoverySelectors::selectClass)
            .toArray(ClassSelector[]::new);
    final LauncherDiscoveryRequest request = LauncherDiscoveryRequestBuilder.request()
            .selectors(classSelectors)
            .build();

    AllureJunit5AnnotationProcessor.setLifecycle(lifecycle);
    final Launcher launcher = LauncherFactory.create();
    launcher.execute(request, new AllureJunit5(lifecycle));
}

@Override
public Stream<Tuple<String>> getInput() {
    List<Tuple<String>> inputTuples = Arrays.asList(
            createMapTuple(1, 2, 3),
            createMapTuple(4, 5, 6),
            createMapTuple(7, 8, 9));
    return inputTuples.stream();
}

private Field[] getAccessibleFields() {
    if (proxy.fields == null) {
        proxy.fields = ReflectionUtil.getFieldsUpToJpaOlingoEntity(getClass());
        Stream.of(proxy.fields).forEach(f -> f.setAccessible(true));
    }
    return proxy.fields;
}

@Test
public void returnedStreamNonTerminal() {
    Integer[] actual = zipIndex(Stream.of(1, 2))
                       .map(Pair::fst) // i.e. can use returned stream
                       .toArray(Integer[]::new);

    assertArrayEquals(array(0, 1), actual);
}

private Snapshot createSnapshot(long minGeneration) {
    try (ReleasableLock ignored = readLock.acquire()) {
        ensureOpen();
        Snapshot[] snapshots = Stream.concat(readers.stream(), Stream.of(current))
                .filter(reader -> reader.getGeneration() >= minGeneration)
                .map(BaseTranslogReader::newSnapshot).toArray(Snapshot[]::new);
        return new MultiSnapshot(snapshots);
    }
}

public double[][] kmExpectedCount(DepInstance inst) {
    int n = inst.length;

    double[][] ret = new double[n][n];
    for (int i = 0; i < n; i++) {
        if (i == 0) {
            for (int j = 1; j < n; j++) {
                ret[i][j] = 1. / (n - 1);
            }
        } else {
            for (int j = 1; j < n; j++) {
                if (i == j) {
                    ret[i][j] = 0;
                } else {
                    ret[i][j] = Math.abs(1. / (i - j));
                }
            }
        }
    }

    double sum = Stream.of(ret).mapToDouble((x -> DoubleStream.of(x).sum())).sum();

    for (int i = 0; i < n; i++) {
        for (int j = 0; j < n; j++) {
            ret[i][j] *= ((n - 1) / sum);
        }
    }
    return ret;
}

/**
 * Returns the containing string of the given file.
 * 
 * @param filePath the path to file
 * @return the content of the given file
 * @throws IOException if an I/O error occurs while opening the file
 */
public static String readStringFromFile(final String filePath) throws IOException {
   StringBuilder builder = new StringBuilder();
   try (Stream<String> stream = Files.lines(Paths.get(filePath), Charset.defaultCharset())) {
      for (Object line : stream.toArray()) {
         builder.append(line);
         builder.append(StringUtils.LINE_BREAK);
      }
   }
   return builder.toString();
}

@Test
public void removeAllOnlyDeletesListedProps() {
    JPropAccessor<String> s1 = makeString(key("s1"));
    JPropAccessor<String> s2 = makeString(key("s2"));                      

    props.setAll(Stream.of(s1, s2), "same");
    props.set(boolProp, true);
    props.removeAll(Stream.of(s1, s2));

    assertFalse(props.get(s1).isPresent());
    assertFalse(props.get(s2).isPresent());
    assertTrue(props.get(boolProp).isPresent());
}

public void testWalk() {
    try (Stream<Path> s = Files.walk(testFolder)) {
        Object[] actual = s.sorted().toArray();
        assertEquals(actual, all);
    } catch (IOException ioe) {
        fail("Unexpected IOException");
    }
}

protected static Stream<? extends Arguments> arguments(ArgumentsProvider provider, ExtensionContext context) {
    try {
        return provider.provideArguments(context);
    } catch (Exception e) {
        throw ExceptionUtils.throwAsUncheckedException(e);
    }
}

@BeforeEach
public void setUp() {
    initMocks(this);
    when(mockResource.getMementos()).thenReturn(emptyList());
    when(mockResource.getInteractionModel()).thenReturn(LDP.RDFSource);
    when(mockResource.getModified()).thenReturn(time);
    when(mockResource.getBinary()).thenReturn(empty());
    when(mockResource.isMemento()).thenReturn(false);
    when(mockResource.getExtraLinkRelations()).thenAnswer(inv -> Stream.empty());

    when(mockLdpRequest.getRequest()).thenReturn(mockRequest);
    when(mockLdpRequest.getPath()).thenReturn("");
    when(mockLdpRequest.getBaseUrl()).thenReturn(baseUrl);
    when(mockLdpRequest.getHeaders()).thenReturn(mockHeaders);
}

private LexerPatternImpl addStopRules(Stream<PositionedMatcher> rules) {
    return new LexerPatternImpl(skipFinders, concat(stopFinders, rules));
}

@Override
public Stream<Vote> getVotes(Stream<Transaction> others) {
    // TODO Auto-generated method stub
    return null;
}

@Override
protected Stream<Slot.Player> slots() {
    return slot.map(Stream::of)
               .orElseGet(Slot.Player::player);
}