<commit_before>/*
 * Copyright 2014 Higher Frequency Trading
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.openhft.chronicle.tcp;

import net.openhft.chronicle.*;
import net.openhft.chronicle.tools.WrappedExcerpt;
import net.openhft.lang.model.constraints.NotNull;
import net.openhft.lang.thread.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.EOFException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ChronicleSource implements Chronicle {
    @NotNull
    private final Chronicle chronicle;
    private final ChronicleSourceConfig config;
    private final ServerSocketChannel server;
    private final Selector selector;
    @NotNull
    private final String name;
    @NotNull
    private final ExecutorService service;
    private final Logger logger;
    private final Object notifier = new Object();
    private static final long busyWaitTimeNS = 100 * 1000;
    private volatile boolean closed;
    private long lastUnpausedNS = 0;
    private int maxMessages;
    private List<ChronicleSourceSocketHandler> handlers;

    public ChronicleSource(@NotNull final Chronicle chronicle, final int port) throws IOException {
        this(chronicle, ChronicleSourceConfig.DEFAULT, new InetSocketAddress(port));
    }

    public ChronicleSource(@NotNull final Chronicle chronicle, @NotNull final InetSocketAddress address) throws IOException {
        this(chronicle, ChronicleSourceConfig.DEFAULT, address);
    }

    public ChronicleSource(@NotNull final Chronicle chronicle, @NotNull final ChronicleSourceConfig config, final int port) throws IOException {
        this(chronicle, config, new InetSocketAddress(port));
    }

    public ChronicleSource(@NotNull final Chronicle chronicle, @NotNull final ChronicleSourceConfig config, @NotNull final InetSocketAddress address) throws IOException {
        this.closed = false;
        this.chronicle = chronicle;
        this.config = config;
        this.server = ServerSocketChannel.open();
        this.server.socket().setReuseAddress(true);
        this.server.socket().bind(address);
        this.server.configureBlocking(false);
        this.selector = Selector.open();
        this.server.register(selector, SelectionKey.OP_ACCEPT);
        this.name = chronicle.name() + "@" + address.getPort();
        this.logger = LoggerFactory.getLogger(getClass().getName() + "." + name);<commit_msg>CHRON-13 Check that resources are cleanup on windows<commit_after>/*
 * Copyright 2014 Higher Frequency Trading
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.openhft.chronicle.tcp;

import net.openhft.chronicle.*;
import net.openhft.chronicle.tools.WrappedExcerpt;
import net.openhft.lang.model.constraints.NotNull;
import net.openhft.lang.thread.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.EOFException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ChronicleSource implements Chronicle {
    @NotNull
    private final Chronicle chronicle;
    private final ChronicleSourceConfig config;
    private final ServerSocketChannel server;
    private final Selector selector;
    @NotNull
    private final String name;
    @NotNull
    private final ExecutorService service;
    private final Logger logger;
    private final Object notifier = new Object();
    private static final long busyWaitTimeNS = 100 * 1000;
    private volatile boolean closed;
    private long lastUnpausedNS = 0;
    private int maxMessages;

    public ChronicleSource(@NotNull final Chronicle chronicle, final int port) throws IOException {
        this(chronicle, ChronicleSourceConfig.DEFAULT, new InetSocketAddress(port));
    }

    public ChronicleSource(@NotNull final Chronicle chronicle, @NotNull final InetSocketAddress address) throws IOException {
        this(chronicle, ChronicleSourceConfig.DEFAULT, address);
    }

    public ChronicleSource(@NotNull final Chronicle chronicle, @NotNull final ChronicleSourceConfig config, final int port) throws IOException {
        this(chronicle, config, new InetSocketAddress(port));
    }

    public ChronicleSource(@NotNull final Chronicle chronicle, @NotNull final ChronicleSourceConfig config, @NotNull final InetSocketAddress address) throws IOException {
        this.closed = false;
        this.chronicle = chronicle;
        this.config = config;
        this.server = ServerSocketChannel.open();
        this.server.socket().setReuseAddress(true);
        this.server.socket().bind(address);
        this.server.configureBlocking(false);
        this.selector = Selector.open();
        this.server.register(selector, SelectionKey.OP_ACCEPT);
        this.name = chronicle.name() + "@" + address.getPort();
        this.logger = LoggerFactory.getLogger(getClass().getName() + "." + name);
{'fraction_non_alphanumeric': 0.08284928156527056, 'fraction_numerical': 0.005808621216753286, 'mean_word_length': 4.226038338658147, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '983071', 'n_tokens_mistral': 1726, 'n_tokens_neox': 1636, 'n_words': 537}