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;
    }
}

@@ -498,181 +498,8 @@
 fer;
-%0Aimport java.util.concurrent.ArrayBlockingQueue;%0Aimport java.util.concurrent.BlockingQueue;%0Aimport java.util.concurrent.Future;%0Aimport java.util.concurrent.SynchronousQueue;
 %0A%0A/*
@@ -770,130 +770,23 @@
 ate 
-int queueSize = 0;%0A    private Session session;%0A    private BlockingQueue%3CILoggingEvent%3E queue;%0A    private Future%3C?%3E task
+Session session
 ;%0A%0A 
@@ -919,53 +919,8 @@
 urn;
-%0A        final WebSocketAppender self = this;
 %0A%0A  
@@ -1349,77 +1349,8 @@
 ) %7B%0A
-                    System.out.println(%22close log: %22 + closeReason);%0A
     
@@ -1441,685 +1441,8 @@
 i);%0A
-            queue = newBlockingQueue(queueSize);%0A            task = self.getContext().getExecutorService().submit(new Runnable() %7B%0A                @Override%0A                public void run() %7B%0A                    try %7B%0A                        while(!Thread.currentThread().isInterrupted()) %7B%0A                            dispatchEvents();%0A                        %7D%0A                    %7D catch (InterruptedException ex) %7B%0A                        assert true;%0A                    %7D catch (IOException ex) %7B%0A                        addError(%22I/O error while sending logs.%22, ex);%0A                    %7D%0A                    addInfo(%22shutting down%22);%0A                %7D%0A            %7D);%0A
     
@@ -1655,35 +1655,8 @@
 n);%0A
-        task.cancel(true);%0A
     
@@ -1798,220 +1798,8 @@
 rn;%0A
-        addInfo(%22append: %22 + event);%0A        queue.offer(event);%0A    %7D%0A%0A    private void dispatchEvents() throws InterruptedException, IOException %7B%0A        RemoteEndpoint.Basic remote = session.getBasicRemote();
 %0A   
@@ -1865,25 +1865,24 @@
 (4096);%0A
-%0A
         
 while (t
@@ -1877,75 +1877,64 @@
     
-while (true) %7B%0A            ILoggingEvent event = queue.tak
+RemoteEndpoint.Async remote = session.getAsyncRemot
 e();%0A
-    
     
@@ -1995,25 +1995,13 @@
     
-    baos.reset();
+try %7B
 %0A   
@@ -2209,220 +2209,88 @@
    %7D
-%0A    %7D%0A%0A    BlockingQueue%3CILoggingEvent%3E newBlockingQueue(int queueSize) %7B%0A        return queueSize %3C= 0 ?%0A                new SynchronousQueue%3CILoggingEvent%3E() : new ArrayBlockingQueue%3CILoggingEvent%3E(queueSize);
+ catch (IOException ex) %7B%0A            throw new RuntimeException(ex);%0A        %7D%0A
 %0A   
@@ -2387,96 +2387,7 @@
    %7D
-%0A%0A    public void setQueueSize(int queueSize) %7B%0A        this.queueSize = queueSize;%0A    %7D
 %0A%7D%0A

Change the remote endpoint to use the async endpoint