<commit_before>package net.ihiroky.niotty.nio;

import net.ihiroky.niotty.DefaultTransportStateEvent;
import net.ihiroky.niotty.StageContext;
import net.ihiroky.niotty.StoreStage;
import net.ihiroky.niotty.TaskLoop;
import net.ihiroky.niotty.TransportState;
import net.ihiroky.niotty.TransportStateEvent;
import net.ihiroky.niotty.buffer.BufferSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * Created on 13/01/10, 17:56
 *
 * @author <NAME>
 */
public abstract class AbstractSelector extends TaskLoop implements StoreStage<BufferSink, Void> {

    private Selector selector_;

    private Logger logger_ = LoggerFactory.getLogger(AbstractSelector.class);

    /**
     * Creates a new instance.
     */
    protected AbstractSelector() {
        super();
    }

    @Override
    protected void onOpen() {
        try {
            selector_ = Selector.open();
        } catch (IOException e) {
            if (selector_ != null) {
                try {
                    selector_.close();
                } catch (Exception e1) {
                    e1.printStackTrace();
                }
                selector_ = null;
            }
            throw new RuntimeException("failed to open selector.", e);
        }
    }

    @Override
    protected void poll(long timeout, TimeUnit timeUnit) throws Exception {
        int selected = (timeout == 0)
                ? selector_.selectNow()
                : selector_.select(Math.max(timeUnit.toMillis(timeout), 1));

        if (selected > 0) {
            processSelectedKeys(selector_.selectedKeys());
        }
    }

    @Override
    protected void wakeUp() {
        selector_.wakeup();
    }

    protected abstract void processSelectedKeys(Set<SelectionKey> selectedKeys) throws Exception;

    @Override
    protected void onClose() {
        for (SelectionKey key : selector_.keys()) {
            SelectableChannel channel = key.channel();
            try {
                key.cancel();
                channel.close();
            } catch (IOException ioe) {
                if (logger_.isDebugEnabled()) {
                    logger_.debug("failed to close registered channel : " + key, ioe);
                }
            }
        }

        try {
            selector_.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    void register(SelectableChannel channel, int ops, NioSocketTransport<?> transport) {
        try {<commit_msg>Add guard not to call wakeup() more than enough<commit_after>package net.ihiroky.niotty.nio;

import net.ihiroky.niotty.DefaultTransportStateEvent;
import net.ihiroky.niotty.StageContext;
import net.ihiroky.niotty.StoreStage;
import net.ihiroky.niotty.TaskLoop;
import net.ihiroky.niotty.TransportState;
import net.ihiroky.niotty.TransportStateEvent;
import net.ihiroky.niotty.buffer.BufferSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created on 13/01/10, 17:56
 *
 * @author <NAME>
 */
public abstract class AbstractSelector extends TaskLoop implements StoreStage<BufferSink, Void> {

    private Selector selector_;
    private AtomicBoolean wakenUp_;

    private Logger logger_ = LoggerFactory.getLogger(AbstractSelector.class);

    /**
     * Creates a new instance.
     */
    protected AbstractSelector() {
        wakenUp_ = new AtomicBoolean();
    }

    @Override
    protected void onOpen() {
        try {
            selector_ = Selector.open();
        } catch (IOException e) {
            if (selector_ != null) {
                try {
                    selector_.close();
                } catch (Exception e1) {
                    e1.printStackTrace();
                }
                selector_ = null;
            }
            throw new RuntimeException("failed to open selector.", e);
        }
    }

    @Override
    protected void poll(long timeout, TimeUnit timeUnit) throws Exception {
        int selected = (timeout == 0)
                ? selector_.selectNow()
                : selector_.select(Math.max(timeUnit.toMillis(timeout), 1));
        wakenUp_.set(false);
        if (selected > 0) {
            processSelectedKeys(selector_.selectedKeys());
        }
    }

    @Override
    protected void wakeUp() {
        if (wakenUp_.compareAndSet(false, true)) {
            selector_.wakeup();
        }
    }

    protected abstract void processSelectedKeys(Set<SelectionKey> selectedKeys) throws Exception;

    @Override
    protected void onClose() {
        for (SelectionKey key : selector_.keys()) {
            SelectableChannel channel = key.channel();
            try {
                key.cancel();
                channel.close();
            } catch (IOException ioe) {
                if (logger_.isDebugEnabled()) {
                    logger_.debug("failed to close registered channel : " + key, ioe);
                }
            }
        }

        try {
            selector_.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    void register(SelectableChannel channel, int ops, NioSocketTransport<?> transport) {
        try {
{'fraction_non_alphanumeric': 0.1009649749821301, 'fraction_numerical': 0.006075768406004289, 'mean_word_length': 2.14438202247191, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, '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': '2140113', 'n_tokens_mistral': 1554, 'n_tokens_neox': 1506, 'n_words': 341}