<commit_before>package net.unit8.logback;

import ch.qos.logback.classic.net.LoggingEventPreSerializationTransformer;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;
import ch.qos.logback.core.spi.PreSerializationTransformer;
import ch.qos.logback.core.util.CloseUtil;

import javax.websocket.*;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;

/**
 * @author kawasima
 */
public class WebSocketAppender extends AppenderBase<ILoggingEvent> {
    private static final PreSerializationTransformer<ILoggingEvent> pst =
            new LoggingEventPreSerializationTransformer();

    private URI serverUri;
    private int queueSize = 0;
    private Session session;
    private BlockingQueue<ILoggingEvent> queue;
    private Future<?> task;

    private void started() {
        super.start();
    }

    @Override
    public void start() {
        if (isStarted()) return;
        final WebSocketAppender self = this;

        WebSocketContainer container = ContainerProvider.getWebSocketContainer();
        try {
            session = container.connectToServer(new Endpoint() {
                @Override
                public void onOpen(Session session, EndpointConfig endpointConfig) {
                    started();
                }

                @Override
                public void onClose(Session session, CloseReason closeReason) {
                    System.out.println("close log: " + closeReason);
                }
            }, ClientEndpointConfig.Builder.create().build(), serverUri);
            queue = newBlockingQueue(queueSize);
            task = self.getContext().getExecutorService().submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        while(!Thread.currentThread().isInterrupted()) {
                            dispatchEvents();
                        }
                    } catch (InterruptedException ex) {
                        assert true;
                    } catch (IOException ex) {
                        addError("I/O error while sending logs.", ex);
                    }
                    addInfo("shutting down");
                }
            });
        } catch (Exception ex) {
            addError("Connect server error", ex);
        }
    }

    @Override
    public void stop() {
        if (!isStarted()) return;
        CloseUtil.closeQuietly(session);
        task.cancel(true);
        super.stop();
    }

    @Override
    protected void append(ILoggingEvent event) {
        if (event == null || !isStarted()) return;
        addInfo("append: " + event);
        queue.offer(event);
    }

    private void dispatchEvents() throws InterruptedException, IOException {
        RemoteEndpoint.Basic remote = session.getBasicRemote();
        ByteArrayOutputStream baos = new ByteArrayOutputStream(4096);

        while (true) {
            ILoggingEvent event = queue.take();
            Serializable serEvent = pst.transform(event);
            baos.reset();
            ObjectOutputStream oos = new ObjectOutputStream(baos);
            oos.writeObject(serEvent);
            oos.flush();
            remote.sendBinary(ByteBuffer.wrap(baos.toByteArray()));
        }
    }

    BlockingQueue<ILoggingEvent> newBlockingQueue(int queueSize) {
        return queueSize <= 0 ?
                new SynchronousQueue<ILoggingEvent>() : new ArrayBlockingQueue<ILoggingEvent>(queueSize);
    }

    public void setServerUri(String uri) {
        this.serverUri = URI.create(uri);
    }

    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
    }
}<commit_msg>Change the remote endpoint to use the async endpoint<commit_after>package net.unit8.logback;

import ch.qos.logback.classic.net.LoggingEventPreSerializationTransformer;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;
import ch.qos.logback.core.spi.PreSerializationTransformer;
import ch.qos.logback.core.util.CloseUtil;

import javax.websocket.*;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.URI;
import java.nio.ByteBuffer;

/**
 * @author kawasima
 */
public class WebSocketAppender extends AppenderBase<ILoggingEvent> {
    private static final PreSerializationTransformer<ILoggingEvent> pst =
            new LoggingEventPreSerializationTransformer();

    private URI serverUri;
    private Session session;

    private void started() {
        super.start();
    }

    @Override
    public void start() {
        if (isStarted()) return;

        WebSocketContainer container = ContainerProvider.getWebSocketContainer();
        try {
            session = container.connectToServer(new Endpoint() {
                @Override
                public void onOpen(Session session, EndpointConfig endpointConfig) {
                    started();
                }

                @Override
                public void onClose(Session session, CloseReason closeReason) {
                }
            }, ClientEndpointConfig.Builder.create().build(), serverUri);
        } catch (Exception ex) {
            addError("Connect server error", ex);
        }
    }

    @Override
    public void stop() {
        if (!isStarted()) return;
        CloseUtil.closeQuietly(session);
        super.stop();
    }

    @Override
    protected void append(ILoggingEvent event) {
        if (event == null || !isStarted()) return;

        ByteArrayOutputStream baos = new ByteArrayOutputStream(4096);
        RemoteEndpoint.Async remote = session.getAsyncRemote();
        Serializable serEvent = pst.transform(event);
        try {
            ObjectOutputStream oos = new ObjectOutputStream(baos);
            oos.writeObject(serEvent);
            oos.flush();
            remote.sendBinary(ByteBuffer.wrap(baos.toByteArray()));
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }

    }

    public void setServerUri(String uri) {
        this.serverUri = URI.create(uri);
    }
}
{'fraction_non_alphanumeric': 0.09577114427860696, 'fraction_numerical': 0.0018656716417910447, 'mean_word_length': 2.336618257261411, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 12, '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': '3397832', 'n_tokens_mistral': 1712, 'n_tokens_neox': 1661, 'n_words': 360}