import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
/**
 * This service is used to read one or more attributes of one or more nodes.
 *
 * @param maxAge             the requested max age of the value, in milliseconds. If maxAge is set to 0, the Server
 *                           shall attempt to read a new value from the data source. If maxAge is set to the max
 *                           Int32 value or greater, the Server shall attempt to get a cached value. Negative values
 *                           are invalid for maxAge.
 * @param timestampsToReturn the requested {@link TimestampsToReturn}.
 * @param nodeIds            the {@link NodeId}s identifying the nodes to read.
 * @param attributeIds       the attribute ids to read, the size and order matching the provided {@link NodeId}s.
 * @return a {@link CompletableFuture} containing a list of {@link DataValue}s, the size and order matching the
 * provided {@link NodeId}s.
 */
default CompletableFuture<List<DataValue>> read(double maxAge,
                                                TimestampsToReturn timestampsToReturn,
                                                List<NodeId> nodeIds,
                                                List<UInteger> attributeIds) {

    if (nodeIds.size() != attributeIds.size()) {
        CompletableFuture<List<DataValue>> failed = new CompletableFuture<>();
        failed.completeExceptionally(new IllegalArgumentException("nodeIds.size() != attributeIds.size()"));
        return failed;
    } else {
        Stream<ReadValueId> stream = StreamUtils.zip(
            nodeIds.stream(), attributeIds.stream(),
            (nId, aId) -> new ReadValueId(nId, aId, null, QualifiedName.NULL_VALUE));

        return read(maxAge, timestampsToReturn, stream.collect(Collectors.toList()))
            .thenApply(r -> l(r.getResults()));
    }
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
/**
 *
 * @param p_action action name
 * @param p_values value stream
 * @param p_possibility stream with possibility of the action
 * @param p_executer function for calling on possibility match
 * @return stream with function results
 */
private static <T> boolean actionconsumer( @Nonnull final String p_action, @Nonnull final Stream<T> p_values,
                                           @Nonnull final Stream<String> p_possibility, @Nonnull final Stream<Consumer<Stream<T>>> p_executer )
{
    return StreamUtils.zip(
        p_possibility,
        p_executer,
        ( i, j ) ->
        {
            if ( p_action.equalsIgnoreCase( i ) )
            {
                j.accept( p_values );
                return true;
            }
            return false;
        }
    ).filter( i -> i ).findFirst().orElseGet( () -> false );
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
/**
 * Take triplet NA and triplet AA then generate the mutation control string.
 *
 * For example: given nas="TTT" and aas="Asn"; the generated result is
 * "  ." since the shortest path from TTT to Asn is via AAT.
 *
 * This method assume that nas and aas are always valid (length%3=0).
 *
 * @param allNAs
 * @param allAAs
 * @return Control string
 */
public static String generateControlString(String allNAs, String allAAs) {
	Iterable<String> nasList = Splitter.fixedLength(3).split(allNAs);
	Iterable<String> aasList = Splitter.fixedLength(3).split(allAAs);
	return StreamUtils.zip(
		StreamSupport.stream(nasList.spliterator(), false),
		StreamSupport.stream(aasList.spliterator(), false),
		(nas, aas) -> {
			if (nas.isEmpty() || aas.isEmpty()) {
				return "";
			}
			return generateTripletControl(nas, aas);
		})
	.collect(Collectors.joining());
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
/**
 * This service is used to write to the value attribute of one or more nodes.
 *
 * @param nodeIds the {@link NodeId}s identifying the nodes to write to.
 * @param values  the {@link DataValue}s to write.
 * @return a {@link CompletableFuture} containing a list of results for the writes.
 */
default CompletableFuture<List<StatusCode>> writeValues(List<NodeId> nodeIds, List<DataValue> values) {
    if (nodeIds.size() != values.size()) {
        CompletableFuture<List<StatusCode>> failed = new CompletableFuture<>();
        failed.completeExceptionally(new IllegalArgumentException("nodeIds.size() != values.size()"));
        return failed;
    } else {
        Stream<WriteValue> stream = StreamUtils.zip(
            nodeIds.stream(), values.stream(),
            (nodeId, value) -> new WriteValue(nodeId, uint(13), null, value));

        return write(stream.collect(Collectors.toList()))
            .thenApply(response -> l(response.getResults()));
    }
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
/**
 * multiple access checks
 *
 * @param p_firstinput first input
 * @param p_secondinput second input
 * @param p_access access collection
 * @param p_accessitem stream with triples of checks
 * @return check result
 */
public static boolean accesscheck( final String p_firstinput, final String p_secondinput, final Collection<EForbiddenAccess> p_access, final Stream<?> p_accessitem )
{
    return StreamUtils.windowed( p_accessitem, 3, 3 )
                      .anyMatch(
                          i -> p_firstinput.equalsIgnoreCase( i.get( 0 ).toString() )
                               && p_secondinput.equalsIgnoreCase( i.get( 1 ).toString() )
                               && p_access.contains( i.get( 2 ) )
                      );
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
/**
 *
 * @param p_action action name
 * @param p_values value stream
 * @param p_possibility stream with possibility of the action
 * @param p_executer function for calling on possibility match
 * @return stream with function results
 */
@Nonnull
private static <R, T> Stream<R> actionfunction( @Nonnull final String p_action, @Nonnull final Stream<T> p_values,
                                                @Nonnull final Stream<String> p_possibility, @Nonnull final Stream<Function<T, R>> p_executer )
{
    return StreamUtils.zip(
        p_possibility,
        p_executer,
        ( i, j ) ->  p_action.equalsIgnoreCase( i )
                     ? p_values.parallel().map( j )
                     : null
    ).flatMap( i -> i );
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
/**
 * Compares a configured route with the given path which comes from the client call.
 *
 * @param route configured route
 * @param request incoming model
 * @return returns true if all parts are equal
 **/
@Override
public boolean test(InternalRoute route, InternalRequest<?> request) {
    List<String> pathParts = request.getPathParts();
    List<RoutePart<?>> routeParts = route.getRouteParts();

    if (routeParts.size() != pathParts.size()) {
        return false;
    }

    request.setSelectedRoute(route);
    return StreamUtils.zip(routeParts.stream(), pathParts.stream(), PathComparator::compareParts)
        .allMatch(result -> result);
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
public String format(Collector<CharSequence, ?, String> collector, FormattingOption...options) {
    FormattingOption process = Stream.of(options).reduce(
            (s, i) -> s,
            (f1, f2) -> (s, i) -> f2.apply(f1.apply(s, i), i));
    return StreamUtils.zipWithIndex(parts.stream())
            .map(indexed -> process.apply(indexed.getValue(), indexed.getIndex()))
            .collect(collector);
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
public static Map<String, PathParam> getPathParams(InternalRoute route, List<String> pathParts) {
    return StreamUtils
        .zip(route.getRouteParts().stream(), pathParts.stream(), pathParamExtractor)
        .filter(Objects::nonNull)
        .collect(toMap(PathParam::getName, identity()));
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
public static Stream<CdxRecord> records(ArchiveReader warcReader, String filename, long warcLength) {
    Stream<CdxRecord> stream = Stream.generate(new CdxRecordProducer(warcReader, filename, warcLength)::next);
    return StreamUtils.takeWhile(stream, (record) -> record != null);
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
public static <T> Comparator<? super List<T>> toListComparator(Comparator<? super T> itemComparator) {
    return (o1, o2) -> StreamUtils.zip(o1.stream(), o2.stream(), itemComparator::compare)
            .filter(c -> c != 0)
            .findFirst()
            .orElseGet(() -> Integer.compare(o1.size(), o2.size()));
}
 

import com.codepoetics.protonpack.StreamUtils; //导入依赖的package包/类
@Override
public QueryResult<ChannelSchedule> execute(ScheduleQuery query)
        throws QueryExecutionException {


    Iterable<Channel> channels = resolveChannels(query);

    ImmutableSet<Publisher> selectedSources = selectedSources(query);

    Ordering<ChannelSchedule> ordering = getQueryIdOrdering(query);

    ImmutableList<ChannelSchedule> orderedChannelSchedules;
    ImmutableList<ChannelSchedule> orderedOverrideSchedules = ImmutableList.of();

    orderedChannelSchedules = ImmutableList.copyOf(
            ordering.sortedCopy(
                    getChannelSchedules(channels, query, selectedSources)
            )
    );

    if(query.getOverride().isPresent()) {
        orderedOverrideSchedules = ImmutableList.copyOf(
                ordering.sortedCopy(
                        getChannelSchedules(channels, query, selectedSources)
                )
        );
    }

    if (query.isMultiChannel()) {
        List<ChannelSchedule> channelSchedules;

        if (query.getOverride().isPresent()) {

            if (orderedChannelSchedules.size() != orderedOverrideSchedules.size()) {
                throw new IllegalStateException(
                        String.format("Original schedule should have same number "
                                + "of items as override: %d vs. %d, %s | %s",
                                orderedChannelSchedules.size(), orderedOverrideSchedules.size(),
                                query.getChannelIds(), query.getOverride()
                        ));
            }

            channelSchedules = StreamUtils
                    .zip(orderedChannelSchedules.stream(), orderedOverrideSchedules.stream(),
                            scheduleMerger::merge)
                    .collect(MoreCollectors.toImmutableList());
        } else {
            channelSchedules = orderedChannelSchedules;
        }
        return QueryResult.listResult(
                channelSchedules,
                query.getContext(),
                channelSchedules.size()
        );
    } else {
        ChannelSchedule originalSchedule =
                Iterables.getOnlyElement(orderedChannelSchedules);
        if (!orderedOverrideSchedules.isEmpty()) {
            ChannelSchedule override = Iterables.getOnlyElement(orderedOverrideSchedules);
            return QueryResult.singleResult(
                    scheduleMerger.merge(originalSchedule, override),
                    query.getContext()
            );
        } else {
            return QueryResult.singleResult(originalSchedule, query.getContext());
        }
    }
}