package dk.lessismore.nojpa.masterworker.master;

import dk.lessismore.nojpa.masterworker.JobStatus;
import dk.lessismore.nojpa.masterworker.messages.JobProgressMessage;
import dk.lessismore.nojpa.masterworker.messages.JobResultMessage;
import dk.lessismore.nojpa.masterworker.messages.JobStatusMessage;
import dk.lessismore.nojpa.masterworker.messages.RunMethodRemoteResultMessage;
import dk.lessismore.nojpa.net.link.ServerLink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class MessageSender {

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

    public static void sendStatusToClient(String jobID, JobStatus jobStatus, ServerLink client, FailureHandler failureHandler) {
        log.debug("sendStatusToClient("+ jobID +")");
        JobStatusMessage statusMessage = new JobStatusMessage(jobID, jobStatus);
        send(statusMessage, client, failureHandler);
    }

    public static void sendProgressToClient(String jobID, double progress, ServerLink client, FailureHandler failureHandler) {
        log.debug("sendProgressToClient("+ jobID +")");
        JobProgressMessage progressMessage = new JobProgressMessage(jobID, progress);
        send(progressMessage, client, failureHandler);
    }

    public static void sendResultToClient(JobResultMessage result, ServerLink client, FailureHandler failureHandler) {
        log.debug("sendResultToClient("+ result.getJobID() +") with ServerLink("+ client.getLinkID() +")");
        send(result, client, failureHandler);
    }

    public static void sendRunMethodRemoteResultOfToClient(RunMethodRemoteResultMessage runMethodRemoteResultMessage, ServerLink client, FailureHandler failureHandler) {
        log.debug("sendRunMethodRemoteResultOfToClient: sending " + runMethodRemoteResultMessage);
        send(runMethodRemoteResultMessage, client, failureHandler);
    }
    /**
     * Send message asynchronous.
     * @param message object to send
     * @param client serverLink to recipient
     * @param failureHandler failure callback or null.
     */
    static ThreadPoolExecutor sendExecutor = new ThreadPoolExecutor(40, 5000, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>());

    public static void send(final Object message, final ServerLink client, final FailureHandler failureHandler) {
        sendExecutor.submit(new Runnable() {
            public void run() {
                try {
                    client.write(message);
                } catch (IOException e) {
                    if (failureHandler != null) failureHandler.onFailure(client);
                }
            }});
        synchronized (sendExecutor) {
            if (sendExecutor.getActiveCount() == sendExecutor.getPoolSize()) {
                log.debug("sendExecutor[1.1]: sendExecutor.getActiveCount(" + sendExecutor.getActiveCount() + "), sendExecutor.getPoolSize(" + sendExecutor.getPoolSize() + "), sendExecutor.getQueue().size(" + sendExecutor.getQueue().size() + ")");
                sendExecutor.setCorePoolSize(sendExecutor.getCorePoolSize() + 1);
            }
        }
    }

    public static void sendOrTimeout(final Object message, final ServerLink client, final FailureHandler failureHandler) {
        sendExecutor.submit(new Runnable() {
            public void run() {
                try {
                    client.writeWithTimeout(message, 1000);
                } catch (IOException e) {
                    log.debug(e.getClass().getSimpleName() + " while sending "+message.getClass().getSimpleName()+" to client");
                    if (failureHandler != null) failureHandler.onFailure(client);
                }
            }
        });
        synchronized (sendExecutor) {
            if (sendExecutor.getActiveCount() == sendExecutor.getPoolSize()) {
                log.debug("sendExecutor[1.1]: sendExecutor.getActiveCount(" + sendExecutor.getActiveCount() + "), sendExecutor.getPoolSize(" + sendExecutor.getPoolSize() + "), sendExecutor.getQueue().size(" + sendExecutor.getQueue().size() + ")");
                sendExecutor.setCorePoolSize(sendExecutor.getCorePoolSize() + 1);
            }
        }
    }

    public interface FailureHandler {
        void onFailure(ServerLink client);
    }
}

@@ -1068,32 +1068,110 @@
 , failureHandler
+, %22sendStatusToClient(%22+ jobID +%22) client.getLinkID(%22+ client.getLinkID() +%22)%22
 );%0A    %7D%0A%0A    pu
@@ -1477,32 +1477,112 @@
 , failureHandler
+, %22sendProgressToClient(%22+ jobID +%22) client.getLinkID(%22+ client.getLinkID() +%22)%22
 );%0A    %7D%0A%0A    pu
@@ -1837,32 +1837,122 @@
 , failureHandler
+, %22sendResultToClient(%22+ result.getJobID() +%22) client.getLinkID(%22+ client.getLinkID() +%22)%22
 );%0A    %7D%0A%0A    pu
@@ -2279,16 +2279,51 @@
 eHandler
+, %22RunMethodRemoteResultOfToClient%22
 );%0A    %7D
@@ -2742,32 +2742,49 @@
 r failureHandler
+, String debugLog
 ) %7B%0A        send
@@ -2894,28 +2894,158 @@
     
-client.write(message
+log.debug(%22Writing (%22+ debugLog+%22) - START%22);%0A                    client.write(message);%0A                    log.debug(%22Writing (%22+ debugLog+%22) - END%22
 );%0A 

Debug statements