<commit_before>package org.rocket.network.netty;

import io.netty.util.concurrent.EventExecutor;
import org.fungsi.Unit;
import org.fungsi.concurrent.Future;
import org.fungsi.concurrent.Futures;
import org.fungsi.concurrent.Timer;
import org.rocket.network.NetworkCommand;

import java.time.Duration;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public final class BroadcastCommand implements NetworkCommand {
	private final EventExecutor worker;
	private final Stream<? extends NettyClient> clients;
	private final Object o;
    private final Supplier<Timer> timer;

	public BroadcastCommand(EventExecutor worker, Stream<? extends NettyClient> clients, Object o, Supplier<Timer> timer) {
		this.worker = worker;
		this.clients = clients;
		this.o = o;
        this.timer = timer;
    }

	@Override
	public void now() {
		clients.map(x -> x.channel)
			   .collect(NettyCollectors.collect(worker, x -> x.writeAndFlush(o)))
			   .awaitUninterruptibly();
	}

	@Override
	public void now(Duration max) {
		clients.map(x -> x.channel)
			   .collect(NettyCollectors.collect(worker, x -> x.writeAndFlush(o)))
			   .awaitUninterruptibly(max.toMillis());
	}

	@Override
	public Future<Unit> async() {
        List<Future<Unit>> futures = clients
                .map(x -> x.channel.writeAndFlush(o))
                .map(ChannelFutures::toFungsi)
                .collect(Collectors.toList());

        return Futures.collect(futures).map(it -> Unit.instance());
	}

	@Override
	public Future<Unit> async(Duration max) {
		return async().within(max, timer.get());
	}
}<commit_msg>much cuter that way <3<commit_after>package org.rocket.network.netty;

import io.netty.util.concurrent.EventExecutor;
import org.fungsi.Unit;
import org.fungsi.concurrent.Future;
import org.fungsi.concurrent.Futures;
import org.fungsi.concurrent.Timer;
import org.rocket.network.NetworkCommand;

import java.time.Duration;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public final class BroadcastCommand implements NetworkCommand {
	private final EventExecutor worker;
	private final Stream<? extends NettyClient> clients;
	private final Object o;
    private final Supplier<Timer> timer;

	public BroadcastCommand(EventExecutor worker, Stream<? extends NettyClient> clients, Object o, Supplier<Timer> timer) {
		this.worker = worker;
		this.clients = clients;
		this.o = o;
        this.timer = timer;
    }

	@Override
	public void now() {
		clients.map(x -> x.channel)
			   .collect(NettyCollectors.collect(worker, x -> x.writeAndFlush(o)))
			   .awaitUninterruptibly();
	}

	@Override
	public void now(Duration max) {
		clients.map(x -> x.channel)
			   .collect(NettyCollectors.collect(worker, x -> x.writeAndFlush(o)))
			   .awaitUninterruptibly(max.toMillis());
	}

	@Override
	public Future<Unit> async() {
        List<Future<Unit>> futures = clients
                .map(x -> x.channel.writeAndFlush(o))
                .map(ChannelFutures::toFungsi)
                .collect(Collectors.toList());

        return Futures.collect(futures).toUnit();
	}

	@Override
	public Future<Unit> async(Duration max) {
		return async().within(max, timer.get());
	}
}
{'fraction_non_alphanumeric': 0.1214868540344515, 'fraction_numerical': 0.0003022061045633122, 'mean_word_length': 3.8462664714494874, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6541803', 'n_tokens_mistral': 1093, 'n_tokens_neox': 1038, 'n_words': 223}