package org.aktin.broker.client.live.sysproc;


import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.aktin.broker.client.live.AbstractExecutionService;
import org.aktin.broker.client2.BrokerClient2;
import org.aktin.broker.xml.RequestStatus;

import lombok.extern.java.Log;


/**
 * Broker client process execution service.
 * @author R.W.Majeed
 *
 */
@Log
public class ProcessExecutionService extends AbstractExecutionService<ProcessExecution> implements Runnable{
	private ProcessExecutionConfig config;

	public ProcessExecutionService(BrokerClient2 client, ProcessExecutionConfig config) {
		super(client, Executors.newSingleThreadExecutor());
		this.config = config;
	}

	public ProcessExecutionService(BrokerClient2 client, ProcessExecutionConfig config, ExecutorService executor) {
		super(client, executor);
		this.config = config;
	}

	@Override
	protected ProcessExecution initializeExecution(Integer requestId) {
		return new ProcessExecution(requestId, config);
	}

	@Override
	protected void onShutdown(List<ProcessExecution> unprocessedExecutions) {
		// ignore unprocessed executions
	}

	@Override
	protected void onStatusUpdate(ProcessExecution execution, RequestStatus status) {
		if( status == RequestStatus.failed && execution.getCause() != null ) {
			if( execution.isAborted() ) {
				log.warning("aborted "+execution.getRequestId());
			}else {
				// print exception
				execution.getCause().printStackTrace();
				System.err.println();
			}
		}
		log.info("status "+execution.getRequestId()+" -> "+status.toString());
	}

	@Override
	public void loadQueue() {
		// no resume of previous requests
	}

	@Override
	protected void onWebsocketClosed(int status) {
		if( !isAborted() ) {
			// websocket closed by server
			log.warning("websocket closed "+status);
			synchronized( this ) {
				this.notifyAll();
			}
		}
	}

	@Override
	public void run() {
		try {
			startupWebsocketListener();
		} catch (IOException e) {
			log.severe("websocket connection failed: "+e.getMessage());
			// exit if the initial websocket connection fails?
			// for now, just continue trying to establish the connection
		}
		log.info("websocket connection established");

		// nothing to do, wait for exit
		while( !isAborted() ) {

			if( isWebsocketClosed() ) {
				// websocket closed by server
				if( config.getWebsocketReconnectSeconds() == -1 ) {
					log.info("websocket retry disabled. exiting.");
					shutdown();
					break;
				}
				log.info("websocket retry after "+config.getWebsocketReconnectSeconds()+"s");
				try {
					Thread.sleep(1000*config.getWebsocketReconnectSeconds());
					startupWebsocketListener();
					log.info("websocket connection re-established");
					// websocket established. poll for missed requests
					pollRequests();
				} catch (IOException e) {
					log.warning("websocket connection failed: "+e.getMessage());
				} catch (InterruptedException e) {
					// interrupted during sleep, try again..
				}
			}else {
				// websocket established, wait for something to happen
				// e.g. shutdown or websocket closing
				synchronized( this ) {
					try {
						this.wait();
					} catch (InterruptedException e) {
						// we expected this interruption
					}
				}
			}
		}
	}
}

@@ -2264,16 +2264,72 @@
 ished%22);
+%0A%09%09long previousConnection = System.currentTimeMillis();
 %0A%0A%09%09// n
@@ -2448,16 +2448,142 @@
  server%0A
+%09%09%09%09// first try to re-establish the connection immediately. if this fails %3C= reconnectSeconds, then wait for requested delay%0A
 %09%09%09%09if( 
@@ -2714,24 +2714,234 @@
 reak;%0A%09%09%09%09%7D%0A
+%09%09%09%09long currentAttempt = System.currentTimeMillis();%0A%09%09%09%09if( currentAttempt - previousConnection %3C config.getWebsocketReconnectSeconds()*1000 ) %7B%0A%09%09%09%09%09// we need to wait the specified amout before next retry%0A%09
 %09%09%09%09log.info
@@ -3010,30 +3010,32 @@
 )+%22s%22);%0A%09%09%09%09
+%09
 try %7B%0A
+%09
 %09%09%09%09%09Thread.
@@ -3081,24 +3081,240 @@
 Seconds());%0A
+%09%09%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09%09%09// interrupted during sleep, try again..%0A%09%09%09%09%09%09continue;%0A%09%09%09%09%09%7D%0A%09%09%09%09%7Delse %7B%0A%09%09%09%09%09// immediate retry%0A%09%09%09%09%7D%0A%09%09%09%09try %7B%0A%09%09%09%09%09previousConnection = System.currentTimeMillis();%0A
 %09%09%09%09%09startup
@@ -3565,93 +3565,8 @@
 ));%0A
-%09%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09%09// interrupted during sleep, try again..%0A
 %09%09%09%09

enable immediate retry of websocket connection if previous connection attempt old enough