<commit_before>/*
 * The MIT License (MIT)
 *
 * Copyright (c) Open Application Platform Authors
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

package oap.logstream.net;

import lombok.extern.slf4j.Slf4j;
import oap.concurrent.scheduler.Scheduled;
import oap.concurrent.scheduler.Scheduler;
import oap.io.Closeables;
import oap.logstream.LoggingBackend;
import oap.metrics.Metrics;

import java.nio.file.Path;
import java.util.concurrent.TimeUnit;

@Slf4j
public class SocketLoggingBackend implements LoggingBackend {

    private final String host;
    private final int port;
    private final Scheduled scheduled;
    protected int maxBuffers = 5000;
    private Connection connection;
    private Buffers buffers;
    protected long flushInterval = 5000;
    private boolean loggingAvailable = true;
    private boolean closed = false;
    protected long timeout = 10000;

    public SocketLoggingBackend( String host, int port, Path location, int bufferSize ) {
        this.host = host;
        this.port = port;
        this.buffers = new Buffers( location, bufferSize );
        this.scheduled = Scheduler.scheduleWithFixedDelay( flushInterval, TimeUnit.MILLISECONDS, this::send );
        Metrics.measureGauge( Metrics.name( "logging_buffers_cache" ), () -> buffers.cache.size() );

    }

    public synchronized void send() {
        if( !closed ) try {
            if( buffers.isEmpty() ) loggingAvailable = true;

            log.debug( "sending data to server..." );
            refreshConnection();

            buffers.forEachReadyData( buffer -> {
                if( !sendBuffer( buffer ) ) {
                    refreshConnection();
                    return sendBuffer( buffer );
                }
                return true;

            } );
            log.debug( "sending done" );
        } catch( Exception e ) {
            loggingAvailable = false;
            log.warn( e.getMessage() );
            log.trace( e.getMessage(), e );
            Closeables.close( connection );
        }

        if( !loggingAvailable ) log.debug( "logging unavailable" );

    }

    private void refreshConnection() {
        if( this.connection == null || !connection.isConnected() ) {
            Closeables.close( connection );
            log.debug( "opening connection..." );
            this.connection = new ChannelConnection( host, port, timeout );
        }
    }

    private Boolean sendBuffer( Buffer buffer ) {
        return Metrics.measureTimer( Metrics.name( "logging_buffer_send_time" ), () -> {
            try {
                log.trace( "sending {}", buffer );<commit_msg>configiurable connection impementation<commit_after>/*
 * The MIT License (MIT)
 *
 * Copyright (c) Open Application Platform Authors
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

package oap.logstream.net;

import lombok.extern.slf4j.Slf4j;
import oap.concurrent.scheduler.Scheduled;
import oap.concurrent.scheduler.Scheduler;
import oap.io.Closeables;
import oap.logstream.LoggingBackend;
import oap.metrics.Metrics;

import java.nio.file.Path;
import java.util.concurrent.TimeUnit;

@Slf4j
public class SocketLoggingBackend implements LoggingBackend {

    private final String host;
    private final int port;
    private final Scheduled scheduled;
    protected int maxBuffers = 5000;
    private Connection connection;
    private Buffers buffers;
    protected long flushInterval = 5000;
    private boolean loggingAvailable = true;
    private boolean closed = false;
    protected long timeout = 10000;
    protected boolean nio = false;

    public SocketLoggingBackend( String host, int port, Path location, int bufferSize ) {
        this.host = host;
        this.port = port;
        this.buffers = new Buffers( location, bufferSize );
        this.scheduled = Scheduler.scheduleWithFixedDelay( flushInterval, TimeUnit.MILLISECONDS, this::send );
        Metrics.measureGauge( Metrics.name( "logging_buffers_cache" ), () -> buffers.cache.size() );

    }

    public synchronized void send() {
        if( !closed ) try {
            if( buffers.isEmpty() ) loggingAvailable = true;

            log.debug( "sending data to server..." );
            refreshConnection();

            buffers.forEachReadyData( buffer -> {
                if( !sendBuffer( buffer ) ) {
                    refreshConnection();
                    return sendBuffer( buffer );
                }
                return true;

            } );
            log.debug( "sending done" );
        } catch( Exception e ) {
            loggingAvailable = false;
            log.warn( e.getMessage() );
            log.trace( e.getMessage(), e );
            Closeables.close( connection );
        }

        if( !loggingAvailable ) log.debug( "logging unavailable" );

    }

    private void refreshConnection() {
        if( this.connection == null || !connection.isConnected() ) {
            Closeables.close( connection );
            log.debug( "opening connection..." );
            this.connection = nio ? new ChannelConnection( host, port, timeout ) : new SocketConnection( host, port );
        }
    }

    private Boolean sendBuffer( Buffer buffer ) {
        return Metrics.measureTimer( Metrics.name( "logging_buffer_send_time" ), () -> {
            try {
                log.trace( "sending {}", buffer );
{'fraction_non_alphanumeric': 0.08625592417061612, 'fraction_numerical': 0.004333107650643196, 'mean_word_length': 2.776073619631902, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '6204093', 'n_tokens_mistral': 1931, 'n_tokens_neox': 1846, 'n_words': 695}