/*
 * 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.publish;

import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;

import jp.ossc.nimbus.core.ServiceBase;
import jp.ossc.nimbus.core.ServiceManagerFactory;
import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.util.SynchronizeMonitor;
import jp.ossc.nimbus.util.WaitSynchronizeMonitor;

/**
 * メッセージ送受信用のサーバコネクション生成サービス。<p>
 * メッセージ送受信を行う{@link RequestReplyServerConnection}を生成するファクトリかつ{@link RequestMessageListener}を登録できる{@link MessageReceiver}の機能を持つ。<br>
 * 
 * @author M.Takata
 */
public class RequestConnectionFactoryService extends ServiceBase
 implements ServerConnectionFactory, MessageReceiver, RequestConnectionFactoryServiceMBean{
    
    private static final long serialVersionUID = -3122390503708261498L;
    
    private ServiceName serverConnectionFactoryServiceName;
    private RequestServerConnectionImpl serverConnection;
    
    private ServiceName messageReceiverServiceName;
    private MessageReceiver messageReceiver;
    
    private Map<MessageListener, MessageListenerWrapper> messageListenerMap;
    private int sequence;
    
    private boolean isAsynchResponse;
    private int responseRetryCount = 1;
    private long responseRetryInterval = 50l;
    
    private Timer timeoutTimer;
    
    public void setServerConnectionFactoryServiceName(ServiceName name){
        serverConnectionFactoryServiceName = name;
    }
    public ServiceName getServerConnectionFactoryServiceName(){
        return serverConnectionFactoryServiceName;
    }
    
    public void setMessageReceiverServiceName(ServiceName name){
        messageReceiverServiceName = name;
    }
    public ServiceName getMessageReceiverServiceName(){
        return messageReceiverServiceName;
    }
    
    public void setAsynchResponse(boolean isAsynch){
        isAsynchResponse = isAsynch;
    }
    public boolean isAsynchResponse(){
        return isAsynchResponse;
    }
    
    public void setResponseRetryCount(int count){
        responseRetryCount = count;
    }
    public int getResponseRetryCount(){
        return responseRetryCount;
    }
    
    public void setResponseRetryInterval(long interval){
        responseRetryInterval = interval;
    }
    public long getResponseRetryInterval(){
        return responseRetryInterval;
    }
    
    public void createService() throws Exception{
        messageListenerMap = Collections.synchronizedMap(new HashMap<MessageListener, MessageListenerWrapper>());
    }
    
    public void startService() throws Exception{
        timeoutTimer = new Timer(true);
        if(messageReceiverServiceName == null){
            throw new IllegalArgumentException("MessageReceiverServiceName must be specified.");
        }
        messageReceiver = ServiceManagerFactory.getServiceObject(messageReceiverServiceName);
        
        if(serverConnectionFactoryServiceName == null){
            throw new IllegalArgumentException("ServerConnectionFactoryServiceName must be specified.");
        }
        ServerConnectionFactory serverConnectionFactory = ServiceManagerFactory.getServiceObject(serverConnectionFactoryServiceName);
        serverConnection = new RequestServerConnectionImpl(serverConnectionFactory.getServerConnection());
    }
    
    public void stopService() throws Exception{
        serverConnection.close();
        serverConnection = null;
        sequence = 0;
        timeoutTimer.cancel();
        timeoutTimer = null;
    }
    
    public void destroyService() throws Exception{
        messageListenerMap = null;
    }
    
    public ServerConnection getServerConnection() throws ConnectionCreateException{
        return serverConnection;
    }
    
    public void addSubject(MessageListener listener, String subject) throws MessageSendException{
        addSubject(listener, subject, null);
    }
    
    public void addSubject(MessageListener listener, String subject, String[] keys) throws MessageSendException{
        listener = getMessageListenerWrapper(listener, true);
        messageReceiver.addSubject(listener, subject, keys);
    }
    
    private MessageListener getMessageListenerWrapper(MessageListener listener, boolean isNew){
        if(listener instanceof RequestMessageListener){
            MessageListenerWrapper wrapper = messageListenerMap.get(listener);
            if(wrapper == null && isNew){
                wrapper = new MessageListenerWrapper((RequestMessageListener)listener);
                messageListenerMap.put(listener, wrapper);
            }
            listener = wrapper;
        }
        return listener;
    }
    
    public void removeSubject(MessageListener listener, String subject) throws MessageSendException{
        removeSubject(listener, subject, null);
    }
    
    public void removeSubject(MessageListener listener, String subject, String[] keys) throws MessageSendException{
        MessageListener lst = listener;
        boolean hasWrapper = false;
        if(lst instanceof RequestMessageListener){
            MessageListenerWrapper wrapper = messageListenerMap.get(lst);
            if(wrapper == null){
                return;
            }
            hasWrapper = true;
            lst = wrapper;
        }
        messageReceiver.removeSubject(lst, subject, keys);
        final Set<String> subjects = messageReceiver.getSubjects(lst);
        if(hasWrapper && (subjects == null || subjects.size() == 0)){
            messageListenerMap.remove(listener);
        }
    }
    
    public void removeMessageListener(MessageListener listener) throws MessageSendException{
        if(listener instanceof RequestMessageListener){
            MessageListenerWrapper wrapper = messageListenerMap.remove(listener);
            if(wrapper == null){
                return;
            }
            listener = wrapper;
        }
        messageReceiver.removeMessageListener(listener);
    }
    
    public Set<String> getSubjects(MessageListener listener){
        listener = getMessageListenerWrapper(listener, false);
        return messageReceiver == null ? null : messageReceiver.getSubjects(listener);
    }
    
    public Set<String> getKeys(MessageListener listener, String subject){
        listener = getMessageListenerWrapper(listener, false);
        return messageReceiver == null ? null : messageReceiver.getKeys(listener, subject);
    }
    
    public ClientConnection getClientConnection(){
        return messageReceiver == null ? null : messageReceiver.getClientConnection();
    }
    
    public void connect() throws Exception{
        messageReceiver.connect();
    }
    
    public void close(){
        messageReceiver.close();
    }
    
    public boolean isConnected(){
        return messageReceiver == null ? false : messageReceiver.isConnected();
    }
    
    public void startReceive() throws MessageSendException{
        messageReceiver.startReceive();
    }
    
    public void stopReceive() throws MessageSendException{
        messageReceiver.stopReceive();
    }
    
    public boolean isStartReceive(){
        return messageReceiver == null ? false : messageReceiver.isStartReceive();
    }
    
    public Object getId(){
        return messageReceiver == null ? null : messageReceiver.getId();
    }
    
    private synchronized int getSequence(){
        return ++sequence;
    }
    
    private class RequestServerConnectionImpl implements RequestServerConnection{
        private ServerConnection serverConnection;
        private Map<Integer, ResponseContainer> responseMap = Collections.synchronizedMap(new HashMap<Integer, ResponseContainer>());
        private boolean isClosed;
        
        public RequestServerConnectionImpl(ServerConnection serverConnection){
            this.serverConnection = serverConnection;
        }
        
        public Message createMessage(String subject, String key) throws MessageCreateException{
            return serverConnection.createMessage(subject, key);
        }
        
        public Message castMessage(Message message) throws MessageException{
            return serverConnection.castMessage(message);
        }
        
        public void send(Message message) throws MessageSendException{
            serverConnection.send(message);
        }
        
        public void sendAsynch(Message message) throws MessageSendException{
            serverConnection.send(message);
        }
        
        public void addServerConnectionListener(ServerConnectionListener listener){
            serverConnection.addServerConnectionListener(listener);
        }
        
        public void removeServerConnectionListener(ServerConnectionListener listener){
            serverConnection.removeServerConnectionListener(listener);
        }
        
        public int getClientCount(){
            return serverConnection.getClientCount();
        }
        
        public Set<Object> getClientIds(){
            return serverConnection.getClientIds();
        }
        
        public Set<Object> getReceiveClientIds(Message message){
            return serverConnection.getReceiveClientIds(message);
        }
        
        public Set<String> getSubjects(Object id){
            return serverConnection.getSubjects(id);
        }
        
        public Set<String> getKeys(Object id, String subject){
            return serverConnection.getKeys(id, subject);
        }
        
        public Message[] request(Message message, int replyCount, long timeout) throws MessageSendException{
            return request(message, null, null, replyCount, timeout);
        }
        
        public Message[] request(Message message, String responseSubject, String responseKey, int replyCount, long timeout) throws MessageSendException{
            if(isClosed){
                throw new MessageSendException("Closed.");
            }
            Set<Object> requestClients = serverConnection.getReceiveClientIds(message);
            if(requestClients.size() == 0){
                return new Message[0];
            }
            int sequence = getSequence();
            try{
                message.setObject(new RequestMessage(messageReceiver.getId(), sequence, responseSubject, responseKey, message.getObject()));
            }catch(MessageException e){
                throw new MessageSendException(e);
            }
            ResponseContainer container = new ResponseContainer(requestClients, replyCount);
            Integer sequenceVal = new Integer(sequence);
            responseMap.put(sequenceVal, container);
            container.init();
            try{
                serverConnection.send(message);
                return container.getResponse(timeout);
            }catch(InterruptedException e){
                throw new MessageSendException(e);
            }finally{
                responseMap.remove(sequenceVal);
            }
        }
        
        public void request(Message message, int replyCount, long timeout, RequestServerConnection.ResponseCallBack callback) throws MessageSendException{
            request(message, null, null, replyCount, timeout, callback);
        }
        
        public void request(Message message, String responseSubject, String responseKey, int replyCount, long timeout, RequestServerConnection.ResponseCallBack callback) throws MessageSendException{
            if(isClosed){
                throw new MessageSendException("Closed.");
            }
            Set<Object> requestClients = serverConnection.getReceiveClientIds(message);
            if(requestClients.size() == 0){
                callback.onResponse(null, true);
                return;
            }
            int sequence = getSequence();
            try{
                message.setObject(new RequestMessage(messageReceiver.getId(), sequence, responseSubject, responseKey, message.getObject()));
            }catch(MessageException e){
                throw new MessageSendException(e);
            }
            Integer sequenceVal = new Integer(sequence);
            ResponseContainer container = new ResponseContainer(sequenceVal, requestClients, replyCount, timeout, callback);
            responseMap.put(sequenceVal, container);
            if(timeout > 0){
                container.startTimer();
            }
            try{
                serverConnection.send(message);
            }catch(MessageSendException e){
                if(timeout > 0){
                    container.cancel();
                }
                throw e;
            }
        }
        
        public void response(Object sourceId, int sequence, Message message) throws MessageSendException{
            message.addDestinationId(sourceId);
            try{
                Object responseObj = message.getObject();
                message.setObject(new ResponseMessage(messageReceiver.getId(), sequence, responseObj));
            }catch(MessageException e){
                throw new MessageSendException(e);
            }
            int count = 0;
            do{
                Set<Object> receivers = serverConnection.getReceiveClientIds(message);
                if(receivers != null && receivers.contains(sourceId)){
                    try{
                        if(isAsynchResponse){
                            serverConnection.sendAsynch(message);
                        }else{
                            serverConnection.send(message);
                        }
                        break;
                    }catch(MessageSendException e){
                        if(count > responseRetryCount){
                            throw e;
                        }
                        // TODO
                        e.printStackTrace();
                    }
                }
                count++;
                try{
                    Thread.sleep(responseRetryInterval);
                }catch(InterruptedException e){}
            }while(count <= responseRetryCount);
        }
        
        protected void reply(Message message, ResponseMessage response){
            ResponseContainer container = responseMap.get(new Integer(response.getSequence()));
            if(container == null){
                return;
            }
            container.onResponse(message, response);
        }
        
        private class ResponseContainer extends TimerTask{
            
            private SynchronizeMonitor monitor = new WaitSynchronizeMonitor();
            private List<Message> responseList = new ArrayList<Message>();
            private final Set<Object> requestClients;
            private final int replyCount;
            private RequestServerConnection.ResponseCallBack callback;
            private long timeout;
            private Object key;
            
            public ResponseContainer(Set<Object> requestClients, int replyCount){
                this.requestClients = requestClients;
                this.replyCount = replyCount;
            }
            
            public ResponseContainer(Object key, Set<Object> requestClients, int replyCount, long timeout, RequestServerConnection.ResponseCallBack callback){
                this.key = key;
                this.requestClients = requestClients;
                this.replyCount = replyCount;
                this.timeout = timeout;
                this.callback = callback;
            }
            
            public synchronized void onResponse(Message message, ResponseMessage response){
                requestClients.remove(response.getSourceId());
                synchronized(responseList){
                    responseList.add(message);
                }
                if(callback == null){
                    if(requestClients.size() == 0 || (replyCount > 0 && responseList.size() >= replyCount)){
                        monitor.notifyAllMonitor();
                    }
                }else{
                    final boolean isLast = requestClients.size() == 0 || (replyCount > 0 && responseList.size() >= replyCount);
                    if(isLast){
                        if(timeout > 0){
                            cancel();
                        }
                        responseMap.remove(key);
                    }
                    callback.onResponse(message, isLast);
                }
            }
            
            public void init(){
                monitor.initMonitor();
            }
            
            public Message[] getResponse(long timeout) throws InterruptedException{
                monitor.waitMonitor(timeout);
                synchronized(responseList){
                    return responseList.toArray(new Message[responseList.size()]);
                }
            }
            
            public void interrupt(){
                Thread[] threads = monitor.getWaitThreads();
                for(int i = 0; i < threads.length; i++){
                    threads[i].interrupt();
                }
            }
            
            public void startTimer(){
                timeoutTimer.schedule(this, timeout);
            }
            
            public void run(){
                responseMap.remove(key);
                callback.onResponse(null, true);
            }
        }
        public void close(){
            isClosed = true;
            synchronized(responseMap){
                Iterator<ResponseContainer> itr = responseMap.values().iterator();
                while(itr.hasNext()){
                    ResponseContainer container = itr.next();
                    itr.remove();
                    container.interrupt();
                }
            }
        }
    }
    
    private class MessageListenerWrapper implements MessageListener{
        private RequestMessageListener requestMessageListener;
        public MessageListenerWrapper(RequestMessageListener listener){
            requestMessageListener = listener;
        }
        public void onMessage(Message message){
            Object obj = null;
            try{
                obj = message.getObject();
            }catch(MessageException e){
                    // TODO
            }
            if(obj instanceof RequestMessage){
                final RequestMessage request = (RequestMessage)obj;
                final Object requestObj = request.getObject();
                try{
                    message.setObject(requestObj);
                }catch(MessageException e){
                    // TODO
                }
                final Message responseMessage = requestMessageListener.onRequestMessage(
                    request.getSourceId(),
                    request.getSequence(),
                    message,
                    request.getResponseSubject(message),
                    request.getResponseKey(message)
                );
                if(responseMessage == null){
                    return;
                }
                responseMessage.addDestinationId(request.getSourceId());
                try{
                    Object responseObj = responseMessage.getObject();
                    responseMessage.setObject(new ResponseMessage(messageReceiver.getId(), request.getSequence(), responseObj));
                }catch(MessageException e){
                    e.printStackTrace();
                    // TODO
                }
                int count = 0;
                do{
                    Set receivers = serverConnection.getReceiveClientIds(responseMessage);
                    if(receivers != null && receivers.contains(request.getSourceId())){
                        try{
                            if(isAsynchResponse){
                                serverConnection.sendAsynch(responseMessage);
                            }else{
                                serverConnection.send(responseMessage);
                            }
                            break;
                        }catch(MessageSendException e){
                            // TODO
                            e.printStackTrace();
                        }
                    }
                    count++;
                    try{
                        Thread.sleep(responseRetryInterval);
                    }catch(InterruptedException e){}
                }while(count <= responseRetryCount);
            }else if(obj instanceof ResponseMessage){
                final ResponseMessage response = (ResponseMessage)obj;
                final Object responseObj = response.getObject();
                try{
                    message.setObject(responseObj);
                }catch(MessageException e){
                    // TODO
                }
                serverConnection.reply(message, response);
            }else{
                requestMessageListener.onMessage(message);
            }
        }
    }
    
    private static abstract class AbstractMessage implements Externalizable{
        private Object sourceId;
        private Object object;
        private int sequence;
        
        public AbstractMessage(){}
        
        public AbstractMessage(Object source, int sequence, Object obj){
            sourceId = source;
            this.sequence = sequence;
            object = obj;
        }
        
        public Object getSourceId(){
            return sourceId;
        }
        
        public int getSequence(){
            return sequence;
        }
        
        public Object getObject(){
            return object;
        }
        
        public void writeExternal(ObjectOutput out) throws IOException{
            out.writeObject(sourceId);
            out.writeInt(sequence);
            out.writeObject(object);
        }
        
        public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException{
            sourceId = in.readObject();
            sequence = in.readInt();
            object = in.readObject();
        }
        
        public String toString(){
            StringBuffer buf = new StringBuffer(super.toString());
            buf.append('{');
            buf.append("sourceId=").append(sourceId);
            buf.append(", sequence=").append(sequence);
            buf.append(", object=").append(object);
            buf.append('}');
            return buf.toString();
        }
    }
    
    private static class RequestMessage extends AbstractMessage{
        
        private String responseSubject;
        private String responseKey;
        
        @SuppressWarnings("unused")
        public RequestMessage(){}
        
        public RequestMessage(Object source, int sequence, String responseSubject, String responseKey, Object obj){
            super(source, sequence, obj);
            this.responseSubject = responseSubject;
            this.responseKey = responseKey;
        }
        
        public String getResponseSubject(Message request){
            if(responseSubject != null){
                return responseSubject;
            }else{
                return request.getSubject();
            }
        }
        
        public String getResponseKey(Message request){
            if(responseSubject != null || responseKey != null){
                return responseKey;
            }else{
                return request.getKey();
            }
        }
        
        public void writeExternal(ObjectOutput out) throws IOException{
            super.writeExternal(out);
            out.writeObject(responseSubject);
            out.writeObject(responseKey);
        }
        
        public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException{
            super.readExternal(in);
            responseSubject = (String)in.readObject();
            responseKey = (String)in.readObject();
        }
    }
    
    private static class ResponseMessage extends AbstractMessage{
        
        @SuppressWarnings("unused")
        public ResponseMessage(){}
        
        public ResponseMessage(Object source, int sequence, Object obj){
            super(source, sequence, obj);
        }
    }
}