/*
 * This software is distributed under following license based on modified BSD
 * style license.
 * ----------------------------------------------------------------------
 * 
 * Copyright 2009 The Nimbus2 Project. All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE NIMBUS PROJECT ``AS IS'' AND ANY EXPRESS
 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
 * NO EVENT SHALL THE NIMBUS PROJECT OR CONTRIBUTORS BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 * 
 * The views and conclusions contained in the software and documentation are
 * those of the authors and should not be interpreted as representing official
 * policies, either expressed or implied, of the Nimbus2 Project.
 */
package jp.ossc.nimbus.service.beanflow;

import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;
import java.util.HashSet;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;

import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.core.ServiceBase;
import jp.ossc.nimbus.core.ServiceManagerFactory;
import jp.ossc.nimbus.service.context.Context;
import jp.ossc.nimbus.service.distribute.ClusterService;
import jp.ossc.nimbus.service.distribute.ClusterListener;
import jp.ossc.nimbus.service.queue.AsynchContext;
import jp.ossc.nimbus.service.queue.Queue;
import jp.ossc.nimbus.service.queue.DefaultQueueService;
import jp.ossc.nimbus.service.queue.QueueHandler;
import jp.ossc.nimbus.service.queue.QueueHandlerContainer;

/**
 * {@link BeanFlowServer}を呼び出すクライアントとなる{@link BeanFlowFactory}実装サービス。<p>
 *
 * @author M.Takata
 */
public class ClientBeanFlowFactoryService extends ServiceBase
 implements BeanFlowFactory, ClusterListener, ClientBeanFlowFactoryServiceMBean{
    
    private static final long serialVersionUID = 6401533155172726865L;
    private ServiceName clusterServiceName;
    private ClusterService cluster;
    private ServiceName contextServiceName;
    private String[] contextKeys;
    private Map<String, List<BeanFlowServer>> flowMap;
    
    private ServiceName asynchInvokeQueueHandlerContainerServiceName;
    private QueueHandlerContainer<Object[]> asynchInvokeQueueHandlerContainer;
    private String asynchInvokeErrorMessageId = MSG_ID_ASYNCH_INVOKE_ERROR;
    
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public void setClusterServiceName(ServiceName name){
        clusterServiceName = name;
    }
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public ServiceName getClusterServiceName(){
        return clusterServiceName;
    }
    
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public void setContextServiceName(ServiceName name){
        contextServiceName = name;
    }
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public ServiceName getContextServiceName(){
        return contextServiceName;
    }
    
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public void setContextKeys(String[] keys){
        contextKeys = keys;
    }
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public String[] getContextKeys(){
        return contextKeys;
    }
    
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public void setAsynchInvokeQueueHandlerContainerServiceName(ServiceName name){
        asynchInvokeQueueHandlerContainerServiceName = name;
    }
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public ServiceName getAsynchInvokeQueueHandlerContainerServiceName(){
        return asynchInvokeQueueHandlerContainerServiceName;
    }
    
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public void setAsynchInvokeErrorMessageId(String id){
        asynchInvokeErrorMessageId = id;
    }
    // ClientBeanFlowFactoryServiceMBeanのJavaDoc
    public String getAsynchInvokeErrorMessageId(){
        return asynchInvokeErrorMessageId;
    }
    
    public void startService() throws Exception{
        if(clusterServiceName == null){
            throw new IllegalArgumentException(
                "ClusterServiceName must be specified."
            );
        }
        cluster = (ClusterService)ServiceManagerFactory.getServiceObject(clusterServiceName);
        cluster.addClusterListener(this);
        
        if(asynchInvokeQueueHandlerContainerServiceName != null){
            asynchInvokeQueueHandlerContainer = ServiceManagerFactory.getServiceObject(asynchInvokeQueueHandlerContainerServiceName);
            AsynchInvokeQueueHandler queueHandler = new AsynchInvokeQueueHandler();
            asynchInvokeQueueHandlerContainer.setQueueHandler(queueHandler);
            asynchInvokeQueueHandlerContainer.accept();
        }
    }
    
    public void stopService() throws Exception{
        if(asynchInvokeQueueHandlerContainer != null){
            asynchInvokeQueueHandlerContainer.release();
            asynchInvokeQueueHandlerContainer = null;
        }
        if(cluster != null){
            cluster.removeClusterListener(this);
            cluster = null;
        }
    }
    
    private BeanFlowServer selectBeanFlowServer(String flowName){
        Map<String, List<BeanFlowServer>> currentFlowMap = flowMap;
        if(currentFlowMap == null || !currentFlowMap.containsKey(flowName)){
            return null;
        }
        List<BeanFlowServer> serverList = (List<BeanFlowServer>)currentFlowMap.get(flowName);
        BeanFlowServer result = null;
        Comparable<Object> resourceUsage = null;
        for(int i = 0; i < serverList.size(); i++){
            BeanFlowServer server = serverList.get(i);
            try{
                if(!server.isAcceptable()){
                    continue;
                }
                if(result == null){
                    resourceUsage = server.getResourceUsage();
                    result = server;
                }else{
                    Comparable<Object> cmpResourceUsage = server.getResourceUsage();
                    if(cmpResourceUsage == null || cmpResourceUsage.compareTo(resourceUsage) < 0){
                        resourceUsage = cmpResourceUsage;
                        result = server;
                    }
                }
            }catch(RemoteException e){
                continue;
            }
            if(resourceUsage == null){
                break;
            }
        }
        return result;
    }
    
    // BeanFlowFactoryのJavaDoc
    public BeanFlow createFlow(String flowName) throws BeanFlowException{
        return createFlow(flowName, null, true);
    }
    
    // BeanFlowFactoryのJavaDoc
    public BeanFlow createFlow(String flowName, String caller, boolean isOverwride) throws BeanFlowException{
        BeanFlowServer server = selectBeanFlowServer(flowName);
        if(server == null){
            throw new NoSuchBeanFlowException(flowName);
        }
        try{
            return new BeanFlowImpl(server.createFlow(flowName, caller, isOverwride), server, contextServiceName, contextKeys, asynchInvokeQueueHandlerContainer);
        }catch(RemoteException e){
            throw new BeanFlowException(flowName + " remote error.", e);
        }catch(NoSuchBeanFlowIdException e){
            throw new BeanFlowException(flowName + " no mapped FLOW", e);
        }
    }
    
    // BeanFlowFactoryのJavaDoc
    public Set<String> getBeanFlowKeySet(){
        Map<String, List<BeanFlowServer>> currentFlowMap = flowMap;
        return currentFlowMap == null ? new HashSet<String>() : new HashSet<String>(currentFlowMap.keySet());
    }
    
    // BeanFlowFactoryのJavaDoc
    public boolean containsFlow(String key){
        Map<String, List<BeanFlowServer>> currentFlowMap = flowMap;
        return currentFlowMap == null ? false : currentFlowMap.containsKey(key);
    }
    
    private Map<String, List<BeanFlowServer>> createFlowMap(List<ClusterService.GlobalUID> members){
        Map<String, List<BeanFlowServer>> result = new HashMap<String, List<BeanFlowServer>>();
        for(int i = 0; i < members.size(); i++){
            final ClusterService.GlobalUID uid = members.get(i);
            final Object option = uid.getOption();
            if(option instanceof BeanFlowServer){
                final BeanFlowServer server = (BeanFlowServer)option;
                Set<String> flowNames = null;
                try{
                    flowNames = server.getBeanFlowNameSet();
                }catch(RemoteException e){
                    continue;
                }
                if(flowNames != null && flowNames.size() != 0){
                    final Iterator<String> itr = flowNames.iterator();
                    while(itr.hasNext()){
                        final String flowName = itr.next();
                        List<BeanFlowServer> servers = (List<BeanFlowServer>)result.get(flowName);
                        if(servers == null){
                            servers = new ArrayList<BeanFlowServer>();
                            result.put(flowName, servers);
                        }
                        servers.add(server);
                    }
                }
            }
        }
        return result;
    }
    
    // ClusterListenerのJavaDoc
    @SuppressWarnings("unchecked")
    public void memberInit(Object myId, List<? extends Object> members){
        flowMap = createFlowMap((List<ClusterService.GlobalUID>)members);
    }
    
    // ClusterListenerのJavaDoc
    @SuppressWarnings("unchecked")
    public void memberChange(List<? extends Object> oldMembers, List<? extends Object> newMembers){
        flowMap = createFlowMap((List<ClusterService.GlobalUID>)newMembers);
    }
    
    // ClusterListenerのJavaDoc
    public void changeMain() throws Exception{}
    
    // ClusterListenerのJavaDoc
    public void changeSub(){}
    
    private static class BeanFlowImpl implements BeanFlow{
        
        private BeanFlowServer server;
        private Object id;
        private String flowName;
        private long startTime = -1;
        private long endTime = -1;
        private ServiceName contextServiceName;
        private String[] contextKeys;
        private ServerBeanFlowMonitor serverMonitor = new ServerBeanFlowMonitor();
        private QueueHandlerContainer<Object[]> asynchInvokeQueueHandlerContainer;
        
        @SuppressWarnings("unused")
        public BeanFlowImpl(){
        }
        
        public BeanFlowImpl(Object id, BeanFlowServer server, ServiceName contextServiceName, String[] contextKeys, QueueHandlerContainer<Object[]> qhc) throws RemoteException, NoSuchBeanFlowIdException{
            this.flowName = server.getFlowName(id);
            this.id = id;
            this.server = server;
            this.contextServiceName = contextServiceName;
            this.contextKeys = contextKeys;
            asynchInvokeQueueHandlerContainer = qhc;
        }
        
        public Object execute(Object obj) throws Exception{
            return execute(obj, null);
        }
        
        public Object execute(Object obj, BeanFlowMonitor monitor) throws Exception{
            startTime = System.currentTimeMillis();
            Map<Object, Object> ctx = null;
            if(contextServiceName != null){
                Context<Object, Object> context = ServiceManagerFactory.getServiceObject(contextServiceName);
                if(contextKeys != null && contextKeys.length != 0){
                    for(int i = 0; i < contextKeys.length; i++){
                        Object val = context.get(contextKeys[i]);
                        if(val != null){
                            if(ctx == null){
                                ctx = new HashMap<Object, Object>();
                            }
                            ctx.put(contextKeys[i], val);
                        }
                    }
                }else if(context.size() != 0){
                    ctx = new HashMap<Object, Object>();
                    ctx.putAll(context);
                }
            }
            if(monitor != null){
                ((BeanFlowMonitorImpl)monitor).addBeanFlowMonitor(serverMonitor);
            }
            try{
                return server.execute(id, obj, ctx);
            }finally{
                endTime = System.currentTimeMillis();
            }
        }
        
        public Object executeAsynch(Object obj, BeanFlowMonitor monitor, boolean isReply, int maxAsynchWait) throws Exception{
            startTime = System.currentTimeMillis();
            BeanFlowAsynchContext context = null;
            if(monitor != null){
                ((BeanFlowMonitorImpl)monitor).addBeanFlowMonitor(serverMonitor);
            }
            if(isReply){
                final DefaultQueueService<AsynchContext<Object,Object>> replyQueue = new DefaultQueueService<AsynchContext<Object,Object>>();
                replyQueue.create();
                replyQueue.start();
                context = new BeanFlowAsynchContext(this, obj, serverMonitor, replyQueue);
                try{
                    executeAsynch(obj, monitor, new BeanFlowAsynchCallbackImpl(context), maxAsynchWait);
                }finally{
                    endTime = System.currentTimeMillis();
                }
            }else{
                context = new BeanFlowAsynchContext(this, obj, serverMonitor);
                try{
                    executeAsynch(obj, monitor, null, maxAsynchWait);
                }finally{
                    endTime = System.currentTimeMillis();
                }
            }
            return context;
        }
        
        public Object getAsynchReply(Object context, BeanFlowMonitor monitor, long timeout, boolean isCancel) throws BeanFlowAsynchTimeoutException, Exception{
            BeanFlowAsynchContext asynchContext = (BeanFlowAsynchContext)context;
            Queue<AsynchContext<Object,Object>> queue = asynchContext.getResponseQueue();
            if(queue == null){
                return null;
            }
            asynchContext = (BeanFlowAsynchContext)queue.get(timeout);
            if(asynchContext == null){
                if(isCancel){
                    if(monitor != null){
                        monitor.cancel();
                        monitor.stop();
                    }
                    BeanFlow flow = ((BeanFlowAsynchContext)context).getBeanFlow();
                    if(flow != null){
                        flow.end();
                    }
                }
                throw new BeanFlowAsynchTimeoutException(flowName);
            }
            try{
                asynchContext.checkError();
            }catch(Throwable th){
                if(th instanceof Exception){
                    throw (Exception)th;
                }else{
                    throw (Error)th;
                }
            }
            return asynchContext.getOutput();
        }
        
        public Object executeAsynch(Object obj, BeanFlowMonitor monitor, BeanFlowAsynchCallback callback, int maxAsynchWait) throws Exception{
            startTime = System.currentTimeMillis();
            if(monitor != null){
                ((BeanFlowMonitorImpl)monitor).addBeanFlowMonitor(serverMonitor);
            }
            if(!(callback instanceof BeanFlowAsynchCallbackImpl)){
                callback = new BeanFlowAsynchCallbackImpl(callback);
            }
            Map<Object, Object> ctx = null;
            if(contextServiceName != null){
                Context<Object, Object> context = ServiceManagerFactory.getServiceObject(contextServiceName);
                if(contextKeys != null && contextKeys.length != 0){
                    for(int i = 0; i < contextKeys.length; i++){
                        Object val = context.get(contextKeys[i]);
                        if(val != null){
                            if(ctx == null){
                                ctx = new HashMap<Object, Object>();
                            }
                            ctx.put(contextKeys[i], val);
                        }
                    }
                }else if(context.size() != 0){
                    ctx = new HashMap<Object, Object>();
                    ctx.putAll(context);
                }
            }
            if(asynchInvokeQueueHandlerContainer == null){
                try{
                    server.executeAsynch(id, obj, ctx, callback, maxAsynchWait);
                }finally{
                    endTime = System.currentTimeMillis();
                }
            }else{
                asynchInvokeQueueHandlerContainer.push(new Object[]{this, id, obj, ctx, callback, new Integer(maxAsynchWait)});
            }
            return null;
        }
        
        protected BeanFlowServer getBeanFlowServer(){
            return server;
        }
        
        protected void setEndTime(long time){
            endTime = time;
        }
        
        public BeanFlowMonitor createMonitor(){
            return serverMonitor;
        }
        
        public String getFlowName(){
            return flowName;
        }
        
        public String[] getOverrideFlowNames(){
            try{
                return server.getOverwrideFlowNames(id);
            }catch(RemoteException e){
                return new String[0];
            }catch(NoSuchBeanFlowIdException e){
                return new String[0];
            }
        }
        
        public void end(){
            try{
                server.end(id);
            }catch(RemoteException e){}
        }
        
        public class ServerBeanFlowMonitor implements BeanFlowMonitor{
            
            public void suspend(){
                try{
                    server.suspendFlow(id);
                }catch(RemoteException e){
                }catch(NoSuchBeanFlowIdException e){
                }
            }
            
            public boolean isSuspend(){
                try{
                    return server.isSuspendFlow(id);
                }catch(RemoteException e){
                    return false;
                }catch(NoSuchBeanFlowIdException e){
                    return false;
                }
            }
            
            public boolean isSuspended(){
                try{
                    return server.isSuspendedFlow(id);
                }catch(RemoteException e){
                    return false;
                }catch(NoSuchBeanFlowIdException e){
                    return false;
                }
            }
            
            public void resume(){
                try{
                    server.resumeFlow(id);
                }catch(RemoteException e){
                }catch(NoSuchBeanFlowIdException e){
                }
            }
            
            public void cancel(){
                try{
                    server.cancel(id);
                }catch(RemoteException e){
                }
            }
            
            public void stop(){
                try{
                    server.stopFlow(id);
                }catch(RemoteException e){
                }
                end();
            }
            
            public boolean isStop(){
                try{
                    return server.isStopFlow(id);
                }catch(RemoteException e){
                    return false;
                }
            }
            
            public boolean isStopped(){
                try{
                    return server.isStoppedFlow(id);
                }catch(RemoteException e){
                    return false;
                }
            }
            
            public boolean isEnd(){
                try{
                    return server.isExistsFlow(id);
                }catch(RemoteException e){
                    return false;
                }
            }
            
            public String getFlowName(){
                return flowName;
            }
            
            public String getCurrentFlowName(){
                try{
                    return server.getCurrentFlowName(id);
                }catch(RemoteException e){
                    return null;
                }catch(NoSuchBeanFlowIdException e){
                    return null;
                }
            }
            
            public String getCurrentStepName(){
                try{
                    return server.getCurrentStepName(id);
                }catch(RemoteException e){
                    return null;
                }catch(NoSuchBeanFlowIdException e){
                    return null;
                }
            }
            
            public long getStartTime(){
                return startTime == -1 ? 0 : startTime;
            }
            
            public long getCurrentProcessTime(){
                if(startTime == -1){
                    return 0;
                }
                if(endTime != -1){
                    return endTime - startTime;
                }else{
                    return System.currentTimeMillis() - startTime;
                }
            }
            
            public void clear(){
                try{
                    server.clearMonitor(id);
                }catch(RemoteException e){
                }catch(NoSuchBeanFlowIdException e){
                }
            }
        }
    }
    
    public static class BeanFlowAsynchCallbackImpl extends UnicastRemoteObject implements BeanFlowAsynchCallback{
        private static final long serialVersionUID = 9186317666094612196L;
        protected BeanFlowAsynchContext context;
        protected BeanFlowAsynchCallback callback;
        
        public BeanFlowAsynchCallbackImpl(BeanFlowAsynchContext context) throws RemoteException{
            this.context = context;
        }
        public BeanFlowAsynchCallbackImpl(BeanFlowAsynchCallback callback) throws RemoteException{
            this.callback = callback;
        }
        
        public void reply(Object output, Throwable th) throws RemoteException{
            if(context != null){
                if(th == null){
                    context.setOutput(output);
                }else{
                    context.setThrowable(th);
                }
                if(context.getResponseQueue() != null){
                    context.getResponseQueue().push(context);
                }
            }else if(callback != null){
                callback.reply(output, th);
            }
        }
    }
    
    private class AsynchInvokeQueueHandler implements QueueHandler<Object[]>{
        public void handleDequeuedObject(Object[] args) throws Throwable{
            if(args == null){
                return;
            }
            final BeanFlowImpl flow = (BeanFlowImpl)args[0];
            final Object id = args[1];
            final Object input = args[2];
            @SuppressWarnings("unchecked")
            final Map<Object, Object> ctx = (Map<Object, Object>)args[3];
            final BeanFlowAsynchCallback callback = (BeanFlowAsynchCallback)args[4];
            final int maxAsynchWait = ((Integer)args[5]).intValue();
            try{
                flow.getBeanFlowServer().executeAsynch(id, input, ctx, callback, maxAsynchWait);
            }finally{
                flow.setEndTime(System.currentTimeMillis());
            }
        }
        
        public boolean handleError(Object[] args, Throwable th) throws Throwable{
            return true;
        }
        
        public void handleRetryOver(Object[] args, Throwable th) throws Throwable{
            final BeanFlowImpl flow = (BeanFlowImpl)args[0];
            if(asynchInvokeErrorMessageId != null){
                getLogger().write(asynchInvokeErrorMessageId, flow.getFlowName(), th);
            }
        }
    }
}