Java 类java.util.function.BiConsumer 实例源码

/**
 * Writes the related collection's URL, using a {@code BiConsumer}.
 *
 * @param relatedCollection the related collection
 * @param parentEmbeddedPathElements the list of embedded path elements
 * @param biConsumer the {@code BiConsumer} that writes the related
 *        collection URL
 */
public <U> void writeRelatedCollection(
    RelatedCollection<T, U> relatedCollection, String resourceName,
    FunctionalList<String> parentEmbeddedPathElements,
    BiConsumer<String, FunctionalList<String>> biConsumer) {

    Predicate<String> fieldsPredicate = getFieldsPredicate();

    String key = relatedCollection.getKey();

    if (!fieldsPredicate.test(key)) {
        return;
    }

    String url = createNestedCollectionURL(
        _requestInfo.getServerURL(), _path, resourceName);

    FunctionalList<String> embeddedPathElements = new FunctionalList<>(
        parentEmbeddedPathElements, key);

    biConsumer.accept(url, embeddedPathElements);
}

/**
 * Test the alignment of the ticks
 * @throws InterruptedException
 * @throws IOException
 * @throws ParseException
 */
@Test(timeout=10000)
public void testTickAlignment1() throws InterruptedException, IOException, ParseException {

       final SimpleDateFormat parser = new SimpleDateFormat("HH:mm:ss");

    final CountDownLatch latch = new CountDownLatch(3);

    final BiConsumer<BitfinexCurrencyPair, Tick> tickConsumer = (s, t) -> {
        Assert.assertTrue(t.getEndTime().getSecond() == 59);
        latch.countDown();
    };

    final TickMerger tickMerger = new TickMerger(BitfinexCurrencyPair.BTC_USD, Timeframe.MINUTES_1, tickConsumer);
    tickMerger.addNewPrice(parser.parse("01:01:23").getTime(), 1.0, 5.0);
    tickMerger.addNewPrice(parser.parse("01:02:33").getTime(), 2.0, 5.0);
    tickMerger.addNewPrice(parser.parse("02:03:53").getTime(), 2.0, 5.0);
    tickMerger.addNewPrice(parser.parse("22:22:53").getTime(), 2.0, 5.0);

    tickMerger.close();

    latch.await();
}

void testDoubleBadOriginBound(BiConsumer<Double, Double> bi) {
        executeAndCatchIAE(() -> bi.accept(17.0, 2.0));
        executeAndCatchIAE(() -> bi.accept(0.0, 0.0));
        executeAndCatchIAE(() -> bi.accept(Double.NaN, FINITE));
        executeAndCatchIAE(() -> bi.accept(FINITE, Double.NaN));
        executeAndCatchIAE(() -> bi.accept(Double.NEGATIVE_INFINITY, Double.NEGATIVE_INFINITY));

        // Returns NaN
//        executeAndCatchIAE(() -> bi.accept(Double.NEGATIVE_INFINITY, FINITE));
//        executeAndCatchIAE(() -> bi.accept(Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY));

        executeAndCatchIAE(() -> bi.accept(FINITE, Double.NEGATIVE_INFINITY));

        // Returns Double.MAX_VALUE
//        executeAndCatchIAE(() -> bi.accept(FINITE, Double.POSITIVE_INFINITY));

        executeAndCatchIAE(() -> bi.accept(Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY));
        executeAndCatchIAE(() -> bi.accept(Double.POSITIVE_INFINITY, FINITE));
        executeAndCatchIAE(() -> bi.accept(Double.POSITIVE_INFINITY, Double.POSITIVE_INFINITY));
    }

@Override
protected <RS, R1 extends Request<RS>> void invokeEntity(E entity, R1 request, BiConsumer<RS, Throwable> callback) throws Exception {
    if (!(request instanceof InvocationRequest)) {
        callback.accept(null, new IllegalArgumentException("Someone managed to call internal dispatcher of "+ProxiedSyncEventSourcingRuntime.this.getClass().getSimpleName()+" directly."));        
    }
    InvocationRequest r = (InvocationRequest)request;
    R handler = entity.requestHandler();
    if (handler == null) {
        callback.accept(null, new NullPointerException("Entity "+entity.getIdentity()+" returned null command handler"));
    } else {
        try {
            callback.accept((RS) r.m.invoke(handler, r.arguments),null);
        } catch (InvocationTargetException ite) {
            logger.info("Caught", ite);
            callback.accept(null, ite.getTargetException());
        } catch (Throwable t) {
             logger.info("Caught", t);
            callback.accept(null, t);
        }
    }
}

public static void runAlert(BiConsumer<Stage, AlertWindowController> setup) {
    try {
        // JavaFX2 doesn't actually have a standard alert template. Instead the Scene Builder app will create FXML
        // files for an alert window for you, and then you customise it as you see fit. I guess it makes sense in
        // an odd sort of way.
        Stage dialogStage = new Stage();
        dialogStage.initModality(Modality.APPLICATION_MODAL);
        FXMLLoader loader = new FXMLLoader(GuiUtils.class.getResource("alert.fxml"));
        Pane pane = loader.load();
        AlertWindowController controller = loader.getController();
        setup.accept(dialogStage, controller);
        dialogStage.setScene(new Scene(pane));
        dialogStage.showAndWait();
    } catch (IOException e) {
        // We crashed whilst trying to show the alert dialog (this should never happen). Give up!
        throw new RuntimeException(e);
    }
}

private <T> void forEachNamedParamInfoParam(Map<T, List<NamedParamInfo>> paramInfo, int level, int subLevel, BiConsumer<T, StatsSet> consumer)
{
    paramInfo.forEach((scope, namedParamInfos) ->
    {
        namedParamInfos.forEach(namedParamInfo ->
        {
            if (((namedParamInfo.getFromLevel() == null) && (namedParamInfo.getToLevel() == null)) || ((namedParamInfo.getFromLevel() <= level) && (namedParamInfo.getToLevel() >= level)))
            {
                if (((namedParamInfo.getFromSubLevel() == null) && (namedParamInfo.getToSubLevel() == null)) || ((namedParamInfo.getFromSubLevel() <= subLevel) && (namedParamInfo.getToSubLevel() >= subLevel)))
                {
                    final StatsSet params = Optional.ofNullable(namedParamInfo.getInfo().getOrDefault(level, Collections.emptyMap()).get(subLevel)).orElseGet(() -> new StatsSet());
                    namedParamInfo.getGeneralInfo().getSet().forEach((k, v) -> params.getSet().putIfAbsent(k, v));
                    params.set(".name", namedParamInfo.getName());
                    consumer.accept(scope, params);
                }
            }
        });
    });
}

@Override
public void operate(MergeEngine set) {

    for (MethodMatchEntry match : set.getAllMethodMatches()) {
        if (match.getNewMethod() == null || match.isMerged()) {
            continue;
        }
        MethodEntry old = match.getOldMethod();
        MappingsSet old_set = set.getOldMappings();
        String key = "L" + old_set.mapTypeSafe(old.getOwnerName()) + ";"
                + old_set.mapMethodSafe(old.getOwnerName(), old.getName(), old.getDescription())
                + MappingsSet.MethodMapping.mapSig(old.getDescription(), old_set);
        BiConsumer<MethodMatchEntry, MergeEngine> merger = custom_mergers.get(key);
        if (merger != null) {
            merger.accept(match, set);
            match.setMerged();
        }
    }

}

private void initializeBackground() {
    final Component component = controller.getComponent();

    // Bind the background width and height to the values in the model
    controller.background.widthProperty().bindBidirectional(component.widthProperty());
    controller.background.heightProperty().bindBidirectional(component.heightProperty());

    final BiConsumer<Color, Color.Intensity> updateColor = (newColor, newIntensity) -> {
        // Set the background color to the lightest possible version of the color
        controller.background.setFill(newColor.getColor(newIntensity.next(-10).next(2)));
    };

    updateColorDelegates.add(updateColor);

    component.colorProperty().addListener(observable -> {
        updateColor.accept(component.getColor(), component.getColorIntensity());
    });

    updateColor.accept(component.getColor(), component.getColorIntensity());
}

@SuppressWarnings("unchecked")
@Override
public synchronized void forEach(BiConsumer<? super K, ? super V> action) {
    Objects.requireNonNull(action);     // explicit check required in case
                                        // table is empty.
    final int expectedModCount = modCount;

    Entry<?, ?>[] tab = table;
    for (Entry<?, ?> entry : tab) {
        while (entry != null) {
            action.accept((K)entry.key, (V)entry.value);
            entry = entry.next;

            if (expectedModCount != modCount) {
                throw new ConcurrentModificationException();
            }
        }
    }
}

public void setUp(final BiConsumer<Integer, Builder> consumer) throws Exception {
    clusterName = "es-idxproxy-" + System.currentTimeMillis();
    // create runner instance
    runner = new ElasticsearchClusterRunner();
    // create ES nodes
    runner.onBuild((number, settingsBuilder) -> {
        settingsBuilder.put("http.cors.enabled", true);
        settingsBuilder.put("http.cors.allow-origin", "*");
        settingsBuilder.putArray("discovery.zen.ping.unicast.hosts", "127.0.0.1:9301-9310");
        consumer.accept(number, settingsBuilder);
    }).build(newConfigs().clusterName(clusterName).numOfNode(numOfNode)
            .pluginTypes("org.codelibs.elasticsearch.idxproxy.IndexingProxyPlugin"));

    // wait for yellow status
    runner.ensureYellow();
}

/**
 * Applies action to all disrupted links between two sets of nodes.
 */
private void applyToNodes(String[] nodes1, String[] nodes2, BiConsumer<MockTransportService, MockTransportService> consumer) {
    for (String node1 : nodes1) {
        if (disruptedLinks.nodes().contains(node1)) {
            for (String node2 : nodes2) {
                if (disruptedLinks.nodes().contains(node2)) {
                    if (node1.equals(node2) == false) {
                        if (disruptedLinks.disrupt(node1, node2)) {
                            consumer.accept(transport(node1), transport(node2));
                        }
                    }
                }
            }
        }
    }
}

private void analyzeRootGroupPorts(NiFiFlow nifiFlow, ProcessGroupStatus rootProcessGroup) {
        BiConsumer<PortStatus, Boolean> portEntityCreator = (port, isInput) -> {
            final String typeName = isInput ? TYPE_NIFI_INPUT_PORT : TYPE_NIFI_OUTPUT_PORT;

            final AtlasEntity entity = new AtlasEntity(typeName);
            final String portName = port.getName();

            entity.setAttribute(ATTR_NIFI_FLOW, nifiFlow.getId());
            entity.setAttribute(ATTR_NAME, portName);
            entity.setAttribute(ATTR_QUALIFIED_NAME, port.getId());
            // TODO: do we have anything to set?
//            entity.setAttribute(ATTR_DESCRIPTION, port.getComponent().getComments());

            final AtlasObjectId portId = new AtlasObjectId(typeName, ATTR_QUALIFIED_NAME, port.getId());
            final Map<AtlasObjectId, AtlasEntity> ports = isInput ? nifiFlow.getRootInputPortEntities() : nifiFlow.getRootOutputPortEntities();
            ports.put(portId, entity);

            if (isInput) {
                nifiFlow.addRootInputPort(port);
            } else {
                nifiFlow.addRootOutputPort(port);
            }
        };

        rootProcessGroup.getInputPortStatus().forEach(port -> portEntityCreator.accept(port, true));
        rootProcessGroup.getOutputPortStatus().forEach(port -> portEntityCreator.accept(port, false));
    }

public static void fillOptionParams(String value, String option, boolean isToast,
        BiConsumer<String, String> c) {
    String quotedOption = PgDiffUtils.getQuotedName(option);
    if (isToast) {
        quotedOption = "toast."+ option;
    }
    c.accept(quotedOption, value);
}

/**
 * The callback is invoked when it finishes calculating the price.
 */
private static void calculatePrice(String product, BiConsumer<RuntimeException, Double> callback) {
    // !!!!! CUIDADO não fazer isto
    Thread th = new Thread(() -> {
        delay(3000);
        if(product.length() > 4 ) callback.accept(new RuntimeException("Illegal Product " + product), null);
        double res = random.nextDouble() * product.charAt(0) + product.charAt(1);
        double price = ((int) (res * 100)) / 100.0;
        callback.accept(null, price);
    });
    th.start();
}

public void forEach(BiConsumer<String, String> consumer) {
    Map<String, Integer> currentIndex = new HashMap<>(valuesByCapitalizedName.size());
    for (String name : headerNames) {
        String key = name.toUpperCase();
        currentIndex.merge(key, 0, (a, b) -> a + 1);
        String value = valuesByCapitalizedName.get(key).get(currentIndex.get(key));
        consumer.accept(name, value);
    }
}

private <T> void testProfilesInSessionAfter(final CompletableFuture<T> initialSetupFuture,
                                           final Function<T, CompletableFuture<Void>> operation,
                                           final BiConsumer<CompletableFuture<List<CommonProfile>>, Async> assertions,
                                           final TestContext testContext) {
    final Async async = testContext.async();
    final CompletableFuture<List<CommonProfile>> profilesInSessionFuture = initialSetupFuture
            .thenCompose(operation)
            .thenCompose(v -> profileManager.getAll(true));
    assertions.accept(profilesInSessionFuture, async);

}

@Override
public void forEach(BiConsumer<? super K, ? super V> action) {
  checkNotNull(action);
  for (Entry<K, V> entry : entries) {
    action.accept(entry.getKey(), entry.getValue());
  }
}

@Override
protected <RS, R extends Request<RS>> void invokeEntity(E entity, R request, BiConsumer<RS, Throwable> callback)
        throws Exception {
    CompletionStage<RS> invocation = entity.execute(request);
    if (invocation == null) {
        logger.error("Entity with ID {} did not return any invocation for request {}. Entity: {}",
            entity.getIdentity(), request, entity);
        callback.accept(null, new IllegalStateException("Entity " + entity.getIdentity()
                + " did not handle the request"));
    } else {
        invocation.whenComplete(callback);
    }
}

@Override
public void onPoll(BiConsumer<String, Float> consumer) {
  if (this.pollConsumer.contains(consumer)) {
    return;
  }

  this.pollConsumer.add(consumer);
}

/**
 * Writes binary resources. This method uses a {@code BiConsumer} so each
 * {@code javax.ws.rs.ext.MessageBodyWriter} can write each binary
 * differently.
 *
 * @param biConsumer the {@code BiConsumer} called to write each binary
 */
public void writeBinaries(BiConsumer<String, String> biConsumer) {
    Function<String, String> urlFunction = binaryId -> createBinaryURL(
        _requestInfo.getServerURL(), binaryId, _path);

    writeFields(
        Representor::getBinaryFunctions,
        entry -> biConsumer.accept(
            entry.getKey(), urlFunction.apply(entry.getKey())));
}

private ProductSyncOptions buildSyncOptions() {
    final BiConsumer<String, Throwable> errorCallBack = (errorMessage, exception) -> {
        errorCallBackMessages.add(errorMessage);
        errorCallBackExceptions.add(exception);
    };
    final Consumer<String> warningCallBack = warningMessage -> warningCallBackMessages.add(warningMessage);

    return ProductSyncOptionsBuilder.of(CTP_TARGET_CLIENT)
                                    .errorCallback(errorCallBack)
                                    .warningCallback(warningCallBack)
                                    .build();
}

@SafeVarargs
static SizeInBytes getSizeInBytes(
    BiFunction<String, SizeInBytes, SizeInBytes> getter,
    String key, SizeInBytes defaultValue, BiConsumer<String, SizeInBytes>... assertions) {
  final SizeInBytes value = get(getter, key, defaultValue, assertions);
  requireMin(0L).accept(key, value.getSize());
  return value;
}

@Override
public void onPressed(BiConsumer<String, Float> consumer) {
  if (this.pressedConsumer.contains(consumer)) {
    return;
  }

  this.pressedConsumer.add(consumer);
}

@Override
protected Collection<EsHighlightedField<EsShopItem, Long>> getFields() {
    List<EsHighlightedField<EsShopItem, Long>> fieldsList = Arrays.asList(HighlightedFields.values());

    fieldsList.add(new EsHighlightedField<EsShopItem, Long>() {
        @Override
        public String getName() {
            return "colors";
        }
        @Override
        public BiConsumer<EsShopItem, String> getConsumer() {
            return (item, content) -> item.setColors(highlightArray(item.getColors(), content));
        }
    });

    fieldsList.add(new EsHighlightedField<EsShopItem, Long>() {
        @Override
        public String getName() {
            return "sizes";
        }
        @Override
        public BiConsumer<EsShopItem, String> getConsumer() {
            return (item, content) -> item.setSizes(highlightArray(item.getSizes(), content));
        }
    });

    return fieldsList;
}

@Override
public void forEach(BiConsumer<? super K, ? super V> action) {
  checkNotNull(action);
  ImmutableList<K> keyList = keySet.asList();
  for (int i = 0; i < size(); i++) {
    action.accept(keyList.get(i), valueList.get(i));
  }
}

private void callbackUnifier() {//@formatter:off
    // how many callbacks exits in your app? did you include errors in all of them, did you even have listeners
    // (means, and handler with more than one success path, like the old MouseListener with onMove, onClick, etc.
    // in just one "callback")?
    Runnable javaThreadTarget = () -> L.log("no arguments, no error path");
    // new Thread(javaThreadTarget)
    Consumer<String> java8Action = success -> L.log("from the sdk, and looks good, but and standard at all");
    Stream.of("a","b","c").forEach(java8Action);
    BiConsumer<String, Throwable> java8Callback = (success, error) -> L.log("nice, but also no one use this");
}

/**
 * @param data
 * @param fnEach
 * @param <T>
 */
public static <T> void itJObject(
        final JsonObject data,
        final BiConsumer<T, String> fnEach
) {
    Congregation.exec(data, fnEach);
}

private <U> CompletableFuture<Void> biAcceptStage(
    Executor e, CompletionStage<U> o,
    BiConsumer<? super T,? super U> f) {
    CompletableFuture<U> b;
    if (f == null || (b = o.toCompletableFuture()) == null)
        throw new NullPointerException();
    CompletableFuture<Void> d = new CompletableFuture<Void>();
    if (e != null || !d.biAccept(this, b, f, null)) {
        BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
        bipush(b, c);
        c.tryFire(SYNC);
    }
    return d;
}

/**
 * Writes a related model. This method uses three consumers: one that writes
 * the model's info, one that writes its URL if it's a linked related model,
 * and one that writes its URL if it's an embedded related model. Therefore,
 * each {@code javax.ws.rs.ext.MessageBodyWriter} can write the related
 * model differently.
 *
 * @param relatedModel the related model instance
 * @param pathFunction the function that gets a single model's path
 * @param modelBiConsumer the consumer that writes the related model's
 *        information
 * @param linkedURLBiConsumer the consumer that writes a linked related
 *        model's URL
 * @param embeddedURLBiConsumer the consumer that writes an embedded related
 *        model's url
 */
public <U> void writeRelatedModel(
    RelatedModel<T, U> relatedModel,
    Function<SingleModel<?>, Optional<Path>> pathFunction,
    BiConsumer<SingleModel<?>, FunctionalList<String>> modelBiConsumer,
    BiConsumer<String, FunctionalList<String>> linkedURLBiConsumer,
    BiConsumer<String, FunctionalList<String>> embeddedURLBiConsumer) {

    writeRelatedModel(
        relatedModel, pathFunction,
        (url, embeddedPathElements) -> {
            Optional<SingleModel<U>> singleModelOptional = getSingleModel(
                relatedModel, _singleModel);

            if (!singleModelOptional.isPresent()) {
                return;
            }

            Predicate<String> embedded = _requestInfo.getEmbedded();

            SingleModel<U> singleModel = singleModelOptional.get();

            Stream<String> stream = Stream.concat(
                Stream.of(embeddedPathElements.head()),
                embeddedPathElements.tailStream());

            String embeddedPath = String.join(
                ".", stream.collect(Collectors.toList()));

            if (embedded.test(embeddedPath)) {
                embeddedURLBiConsumer.accept(url, embeddedPathElements);
                modelBiConsumer.accept(singleModel, embeddedPathElements);
            }
            else {
                linkedURLBiConsumer.accept(url, embeddedPathElements);
            }
        });
}

@SuppressWarnings("unchecked")
public void forEach(BiConsumer<? super Register, ? super T> consumer) {
    for (int i = 0; i < values.length; ++i) {
        T value = (T) values[i];
        if (value != null) {
            consumer.accept(architecture.getRegisters().get(i), value);
        }
    }
}

public final void compute() {
    final BiConsumer<? super K, ? super V> action;
    if ((action = this.action) != null) {
        for (int i = baseIndex, f, h; batch > 0 &&
                 (h = ((f = baseLimit) + i) >>> 1) > i;) {
            addToPendingCount(1);
            new ForEachMappingTask<K,V>
                (this, batch >>>= 1, baseLimit = h, f, tab,
                 action).fork();
        }
        for (Node<K,V> p; (p = advance()) != null; )
            action.accept(p.key, p.val);
        propagateCompletion();
    }
}

/**
 * Returns a collector that collects items in a Morpheus array
 * @param type              the array type
 * @param expectedLength    an estimate of the expected length, does not have to be exact
 * @param <T>               the array element type
 * @return                  the newly created collector
 */
public static <T> Collector<T,ArrayBuilder<T>,Array<T>> toArray(Class<T> type, int expectedLength) {
    final Supplier<ArrayBuilder<T>> supplier = () -> ArrayBuilder.of(expectedLength, type);
    final BinaryOperator<ArrayBuilder<T>> combiner = ArrayBuilder::addAll;
    final BiConsumer<ArrayBuilder<T>,T> accumulator = ArrayBuilder::add;
    final Function<ArrayBuilder<T>,Array<T>> finisher = ArrayBuilder::toArray;
    return Collector.of(supplier, accumulator, combiner, finisher);
}

/**
 * 游戏记录
 *
 * @return
 */
private GameBase.RecordResponse gameRecord() {
    GameBase.RecordResponse.Builder recordResponse = GameBase.RecordResponse.newBuilder();
    jsonObject.clear();
    jsonObject.put("userId", userId);
    ApiResponse<List<GameRecordRepresentation>> gameRecordResponse = JSON.parseObject(HttpUtil.urlConnectionByRsa(Constant.apiUrl + Constant.gamerecordListUrl, jsonObject.toJSONString()),
            new TypeReference<ApiResponse<List<GameRecordRepresentation>>>() {
            });
    Map<GameType, GameBase.GameRecord.Builder> gameRecords = new HashMap<>();
    if (0 == gameRecordResponse.getCode()) {
        for (GameRecordRepresentation gameRecordRepresentation : gameRecordResponse.getData()) {
            if (!gameRecords.containsKey(gameRecordRepresentation.getGameType())) {
                gameRecords.put(gameRecordRepresentation.getGameType(), GameBase.GameRecord.newBuilder()
                        .setGameType(GameBase.GameType.forNumber(gameRecordRepresentation.getGameType().ordinal())));
            }
            GameBase.Record.Builder record = GameBase.Record.newBuilder();
            record.setRecordId(gameRecordRepresentation.getId());
            record.setRoomNo(gameRecordRepresentation.getRoomNo() + "");
            record.setGameCount(gameRecordRepresentation.getGameCount());
            record.setDateTime(gameRecordRepresentation.getCreateDate().getTime());
            if (null != gameRecordRepresentation.getsData()) {
                List<TotalScore> totalScores = JSON.parseArray(new String(gameRecordRepresentation.getsData(), Charset.forName("utf-8")), TotalScore.class);
                for (TotalScore totalScore : totalScores) {
                    record.addUserRecord(GameBase.UserRecord.newBuilder().setNickname(totalScore.getNickname())
                            .setHead(totalScore.getHead()).setID(totalScore.getUserId()).setScore(totalScore.getScore()));
                }
            }
            gameRecords.get(gameRecordRepresentation.getGameType()).addRecords(record);
        }
        gameRecords.forEach(new BiConsumer<GameType, GameBase.GameRecord.Builder>() {
            @Override
            public void accept(GameType gameType, GameBase.GameRecord.Builder builder) {
                recordResponse.addGameRecords(builder);
            }
        });
    }
    return recordResponse.build();
}

private <U> CompletableFuture<Void> biAcceptStage(
    Executor e, CompletionStage<U> o,
    BiConsumer<? super T,? super U> f) {
    CompletableFuture<U> b;
    if (f == null || (b = o.toCompletableFuture()) == null)
        throw new NullPointerException();
    CompletableFuture<Void> d = new CompletableFuture<Void>();
    if (e != null || !d.biAccept(this, b, f, null)) {
        BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
        bipush(b, c);
        c.tryFire(SYNC);
    }
    return d;
}

private void test(BenchmarkState sbState, ThreadState tState, BiConsumer<StringBuilder, Double> formatter)
{
    sbState.stringBuilders.forEach(sb -> {
               IntStream.range(0, tState.numRecordings).forEach(
                       i -> {
                           formatter.accept(sb, tState.rand.nextDouble());
                       }
               );
               String s = sb.toString();
           }
    );
}

/**
 * Implements a unary -> stream call as {@link Mono} -> {@link Flux}, where the server responds with a
 * stream of messages.
 */
public static <TRequest, TResponse> Flux<TResponse> oneToMany(
        Mono<TRequest> rxRequest,
        BiConsumer<TRequest, StreamObserver<TResponse>> delegate) {
    try {
        ReactorConsumerStreamObserver<TRequest, TResponse> consumerStreamObserver = new ReactorConsumerStreamObserver<>();
        rxRequest.subscribe(request -> delegate.accept(request, consumerStreamObserver));
        return ((Flux<TResponse>) consumerStreamObserver.getRxConsumer())
                .transform(Operators.lift(new SubscribeOnlyOnceLifter<TResponse>()));
    } catch (Throwable throwable) {
        return Flux.error(throwable);
    }
}

public static <T> CompletableSubscriber<T> pullEach(BiConsumer<T, Subscription> consumer) {
  return new AbstractCompletableSubscriber<T>() {

    @Override
    public void onSubscribe(Subscription subscription) {
      super.onSubscribe(subscription);
      subscription.request(1);
    }

    @Override
    public void onNext(T item) {
      consumer.accept(item, subscription);
    }
  };
}

static void assertAssembledMessage(final BiConsumer<Object, ActorRef> mockAssembledMessageCallback,
        final BytesMessage message, final ActorRef sender) {
    ArgumentCaptor<Object> assembledMessageCaptor = ArgumentCaptor.forClass(Object.class);
    ArgumentCaptor<ActorRef> senderActorRefCaptor = ArgumentCaptor.forClass(ActorRef.class);
    verify(mockAssembledMessageCallback).accept(assembledMessageCaptor.capture(), senderActorRefCaptor.capture());
    assertEquals("Assembled message", message, assembledMessageCaptor.getValue());
    assertEquals("Sender ActorRef", sender, senderActorRefCaptor.getValue());
}

@SuppressWarnings("unchecked")
HelperEventListener(SingleBuilder<T> builder, List<BiConsumer<SingleSubscription<T>, ? super T>> handlers) {
    this.eventClass = builder.eventClass;
    this.priority = builder.priority;
    this.exceptionConsumer = builder.exceptionConsumer;

    this.filters = builder.filters.toArray(new Predicate[builder.filters.size()]);
    this.preExpiryTests = builder.preExpiryTests.toArray(new BiPredicate[builder.preExpiryTests.size()]);
    this.midExpiryTests = builder.midExpiryTests.toArray(new BiPredicate[builder.midExpiryTests.size()]);
    this.postExpiryTests = builder.postExpiryTests.toArray(new BiPredicate[builder.postExpiryTests.size()]);
    this.handlers = handlers.toArray(new BiConsumer[handlers.size()]);

    this.timing = Timings.of("helper-events: " + handlers.stream().map(handler -> Delegate.resolve(handler).getClass().getName()).collect(Collectors.joining(" | ")));
}

public void forEach(BiConsumer<? super K, ? super V> action) {
    if (action == null) throw new NullPointerException();
    V v;
    for (Node<K,V> n = findFirst(); n != null; n = n.next) {
        if ((v = n.getValidValue()) != null)
            action.accept(n.key, v);
    }
}