<commit_before>/*
 * Copyright 2022 Asynchronous Game Query Library
 *
 * 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 com.ibasco.agql.protocols.valve.source.query.logger;

import com.ibasco.agql.core.transport.enums.TransportType;
import com.ibasco.agql.core.util.Platform;
import com.ibasco.agql.core.util.TransportOptions;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.DatagramChannel;
import org.jetbrains.annotations.ApiStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

/**
 * <p>Listens for log messages produced by a Source Server.</p>
 * <br />
 *
 * <p>
 * <strong>NOTE:</strong> Ensure that the address of this service is registered to the source server, do this by registering it with  <code>'logaddress_add [ip:port]'</code> to receive log
 * messages from the game server.
 * Make sure that the port specified is not being blocked by an external service or firewall
 * </p>
 * <br />
 * <h3>Example Usage</h3>
 * <pre>
 * {@code
 *  SourceLogListenService service = new SourceLogListenService(new InetSocketAddress(0));
 *
 *     service.setLogEventCallback(SourceLogEventHandler::handleLogMessages);
 *     service.listen(); //non-blocking
 *     service.listen().join(); //blocking (will wait until close() is called or due to interruption)
 *
 *     public class SourceLogEventPrinter {
 *         public static void handleLogMessages(SourceLogEntry logEntry) {
 *             System.out.println(logEntry.getMessage());
 *         }
 *     }
 * }
 * </pre>
 *
 * @author <NAME>
 */
public class SourceLogListenService implements Closeable {

    private static final Logger log = LoggerFactory.getLogger(SourceLogListenService.class);

    private InetSocketAddress listenAddress;

    private final Bootstrap bootstrap;

    private final EventLoopGroup group;

    private volatile boolean bindInProgress;

    private volatile boolean started;

    private final AtomicReference<Consumer<SourceLogEntry>> callbackRef = new AtomicReference<>();

    /**
     * <p>Creates a new service that will listen to any ip address
     * and bind to a random local port number (Similar to 0.0.0.0)</p>
     */
    public SourceLogListenService() {
        this(new InetSocketAddress(0));
    }

    /**
     * <p>Creates a new service using the specified {@link InetSocketAddress} to listen on.</p>
     *
     * @param listenAddress
     *         An {@link InetSocketAddress} where the listen service will bind or listen on
     */
    public SourceLogListenService(InetSocketAddress listenAddress) {
        this(listenAddress, null);
    }

    /**
     * <p>Creates a new service using the specified {@link InetSocketAddress} to listen on and utilizing
     * the callback specified to notify listeners of source log events</p>
     *
     * @param listenAddress
     *         The {@link InetSocketAddress} to listen on
     * @param callback
     *         The {@link Consumer} callback that will be notified once a log event has been received
     */
    public SourceLogListenService(InetSocketAddress listenAddress, Consumer<SourceLogEntry> callback) {
        this(listenAddress, callback, null, 0, true);
    }

    /**
     * <p>Creates a new service using the specified {@link InetSocketAddress} to listen on and utilizing
     * the callback specified to notify listeners of source log events</p>
     *
     * @param listenAddress
     *         The {@link InetSocketAddress} to listen on
     * @param callback
     *         The {@link Consumer} callback that will be notified once a log event has been received
     * @param executorService
     *         The {@link ExecutorService} that will be used by the service. {@code null} to use the global executor provided by the library.
     * @param nThreads
     *         The number of threads that will be used by the underlying {@link EventLoopGroup} (a special executor servicee used by netty). The value should normally be less than or equals to the core pool size of the executor service.
     * @param useNative
     *         {@code true} if you prefer to use netty's <a href="https://netty.io/wiki/native-transports.html">native transports</a> over java's NIO (e.g. epoll on linux, kqueue on osx)
     */
    public SourceLogListenService(InetSocketAddress listenAddress, Consumer<SourceLogEntry> callback, ExecutorService executorService, int nThreads, boolean useNative) {
        EventLoopGroup group;
        if (executorService != null) {
            //ensure arguments are available
            if (nThreads < 0) {
                log.debug("LOG SERVICE => Using default nThreads parameter = 0");
                nThreads = 0;
            }
            group = Platform.createEventLoopGroup(executorService, nThreads, useNative);
        } else {
            group = Platform.getDefaultEventLoopGroup();
        }
        final Class<? extends Channel> channelClass = Platform.getChannelClass(TransportType.UDP, group);
        log.debug("LOG SERVICE => Executor: {}, Event loop group: {}, Channel class: {}, nThreads: {}, useNative: {}", executorService, group, channelClass, nThreads, useNative);
        this.listenAddress = listenAddress;
        this.group = group;
        setLogEventCallback(callback);
        bootstrap = new Bootstrap()
                .localAddress(listenAddress)
                .channel(channelClass)
                .attr(SourceLogListenHandler.CALLBACK, callbackRef)<commit_msg>Use poold byte buffer allocator by default<commit_after>/*
 * Copyright 2022 Asynchronous Game Query Library
 *
 * 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 com.ibasco.agql.protocols.valve.source.query.logger;

import com.ibasco.agql.core.transport.enums.TransportType;
import com.ibasco.agql.core.util.Platform;
import com.ibasco.agql.core.util.TransportOptions;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.DatagramChannel;
import org.jetbrains.annotations.ApiStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

/**
 * <p>Listens for log messages produced by a Source Server.</p>
 * <br />
 *
 * <p>
 * <strong>NOTE:</strong> Ensure that the address of this service is registered to the source server, do this by registering it with  <code>'logaddress_add [ip:port]'</code> to receive log
 * messages from the game server.
 * Make sure that the port specified is not being blocked by an external service or firewall
 * </p>
 * <br />
 * <h3>Example Usage</h3>
 * <pre>
 * {@code
 *  SourceLogListenService service = new SourceLogListenService(new InetSocketAddress(0));
 *
 *     service.setLogEventCallback(SourceLogEventHandler::handleLogMessages);
 *     service.listen(); //non-blocking
 *     service.listen().join(); //blocking (will wait until close() is called or due to interruption)
 *
 *     public class SourceLogEventPrinter {
 *         public static void handleLogMessages(SourceLogEntry logEntry) {
 *             System.out.println(logEntry.getMessage());
 *         }
 *     }
 * }
 * </pre>
 *
 * @author <NAME>
 */
public class SourceLogListenService implements Closeable {

    private static final Logger log = LoggerFactory.getLogger(SourceLogListenService.class);

    private InetSocketAddress listenAddress;

    private final Bootstrap bootstrap;

    private final EventLoopGroup group;

    private volatile boolean bindInProgress;

    private volatile boolean started;

    private final AtomicReference<Consumer<SourceLogEntry>> callbackRef = new AtomicReference<>();

    /**
     * <p>Creates a new service that will listen to any ip address
     * and bind to a random local port number (Similar to 0.0.0.0)</p>
     */
    public SourceLogListenService() {
        this(new InetSocketAddress(0));
    }

    /**
     * <p>Creates a new service using the specified {@link InetSocketAddress} to listen on.</p>
     *
     * @param listenAddress
     *         An {@link InetSocketAddress} where the listen service will bind or listen on
     */
    public SourceLogListenService(InetSocketAddress listenAddress) {
        this(listenAddress, null);
    }

    /**
     * <p>Creates a new service using the specified {@link InetSocketAddress} to listen on and utilizing
     * the callback specified to notify listeners of source log events</p>
     *
     * @param listenAddress
     *         The {@link InetSocketAddress} to listen on
     * @param callback
     *         The {@link Consumer} callback that will be notified once a log event has been received
     */
    public SourceLogListenService(InetSocketAddress listenAddress, Consumer<SourceLogEntry> callback) {
        this(listenAddress, callback, null, 0, true);
    }

    /**
     * <p>Creates a new service using the specified {@link InetSocketAddress} to listen on and utilizing
     * the callback specified to notify listeners of source log events</p>
     *
     * @param listenAddress
     *         The {@link InetSocketAddress} to listen on
     * @param callback
     *         The {@link Consumer} callback that will be notified once a log event has been received
     * @param executorService
     *         The {@link ExecutorService} that will be used by the service. {@code null} to use the global executor provided by the library.
     * @param nThreads
     *         The number of threads that will be used by the underlying {@link EventLoopGroup} (a special executor servicee used by netty). The value should normally be less than or equals to the core pool size of the executor service.
     * @param useNative
     *         {@code true} if you prefer to use netty's <a href="https://netty.io/wiki/native-transports.html">native transports</a> over java's NIO (e.g. epoll on linux, kqueue on osx)
     */
    public SourceLogListenService(InetSocketAddress listenAddress, Consumer<SourceLogEntry> callback, ExecutorService executorService, int nThreads, boolean useNative) {
        EventLoopGroup group;
        if (executorService != null) {
            //ensure arguments are available
            if (nThreads < 0) {
                log.debug("LOG SERVICE => Using default nThreads parameter = 0");
                nThreads = 0;
            }
            group = Platform.createEventLoopGroup(executorService, nThreads, useNative);
        } else {
            group = Platform.getDefaultEventLoopGroup();
        }
        final Class<? extends Channel> channelClass = Platform.getChannelClass(TransportType.UDP, group);
        log.debug("LOG SERVICE => Executor: {}, Event loop group: {}, Channel class: {}, nThreads: {}, useNative: {}", executorService, group, channelClass, nThreads, useNative);
        this.listenAddress = listenAddress;
        this.group = group;
        setLogEventCallback(callback);
        bootstrap = new Bootstrap()
                .localAddress(listenAddress)
                .channel(channelClass)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .attr(SourceLogListenHandler.CALLBACK, callbackRef)
{'fraction_non_alphanumeric': 0.08960825709594182, 'fraction_numerical': 0.0034404566424270856, 'mean_word_length': 3.267600934267601, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 69, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, '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': '3640015', 'n_tokens_mistral': 3442, 'n_tokens_neox': 3351, 'n_words': 1165}