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

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.service.log.Logger;
import jp.ossc.nimbus.service.publish.Client;
import jp.ossc.nimbus.service.publish.Message;
import jp.ossc.nimbus.service.publish.MessageCreateException;
import jp.ossc.nimbus.service.publish.MessageSendException;
import jp.ossc.nimbus.service.publish.MessageException;
import jp.ossc.nimbus.service.publish.ServerConnection;
import jp.ossc.nimbus.service.publish.ServerConnectionListener;
import jp.ossc.nimbus.service.publish.ConnectException;
import jp.ossc.nimbus.service.queue.AsynchContext;
import jp.ossc.nimbus.service.queue.DefaultQueueService;
import jp.ossc.nimbus.service.queue.QueueHandler;
import jp.ossc.nimbus.service.queue.QueueHandlerContainerService;
import jp.ossc.nimbus.service.queue.AbstractDistributedQueueSelectorService;
import jp.ossc.nimbus.service.queue.DistributedQueueHandlerContainerService;

/**
 * ローカル用の{@link ServerConnection}インタフェース実装クラス。<p>
 *
 * @author M.Takata
 */
public class ServerConnectionImpl implements ServerConnection{
    
    private ServiceName serverConnectionFactroyServiceName;
    private Map<Object, ClientImpl> clients = Collections.synchronizedMap(new LinkedHashMap<Object, ClientImpl>());
    private Logger logger;
    private String sendErrorMessageId;
    private String sendErrorRetryOverMessageId;
    private QueueHandlerContainerService<AsynchContext<SendRequest,?>> sendQueueHandlerContainer;
    private ClientDistributedQueueSelector queueSelector;
    private DistributedQueueHandlerContainerService<AsynchContext<SendRequest,?>> asynchSendQueueHandlerContainer;
    private long sendCount;
    private List<ServerConnectionListener> serverConnectionListeners;
    private List<MessageImpl> sendBufferList = Collections.synchronizedList(new ArrayList<MessageImpl>());
    private long sendBufferTime;
    
    public ServerConnectionImpl(
        ServiceName serverConnectionFactroyServiceName,
        int sendThreadSize,
        ServiceName sendQueueServiceName,
        int asynchSendThreadSize,
        ServiceName asynchSendQueueServiceName
    ) throws Exception{
        this.serverConnectionFactroyServiceName = serverConnectionFactroyServiceName;
        initSend(sendQueueServiceName, sendThreadSize);
        initAsynchSend(asynchSendQueueServiceName, asynchSendThreadSize);
    }
    
    private void initSend(ServiceName sendQueueServiceName, int sendThreadSize) throws Exception{
        if(sendThreadSize >= 2){
            sendQueueHandlerContainer = new QueueHandlerContainerService<AsynchContext<SendRequest,?>>();
            sendQueueHandlerContainer.create();
            if(sendQueueServiceName == null){
                DefaultQueueService<AsynchContext<SendRequest,?>> sendQueue = new DefaultQueueService<AsynchContext<SendRequest,?>>();
                sendQueue.create();
                sendQueue.start();
                sendQueueHandlerContainer.setQueueService(sendQueue);
            }else{
                sendQueueHandlerContainer.setQueueServiceName(sendQueueServiceName);
            }
            sendQueueHandlerContainer.setQueueHandlerSize(sendThreadSize);
            sendQueueHandlerContainer.setQueueHandler(new SendQueueHandler());
            sendQueueHandlerContainer.start();
        }
    }
    
    private void initAsynchSend(ServiceName queueFactoryServiceName, int clientQueueDistributedSize) throws Exception{
        if(clientQueueDistributedSize > 0){
            queueSelector = new ClientDistributedQueueSelector();
            queueSelector.create();
            queueSelector.setDistributedSize(clientQueueDistributedSize);
            if(queueFactoryServiceName != null){
                queueSelector.setQueueFactoryServiceName(queueFactoryServiceName);
            }
            queueSelector.start();
            
            asynchSendQueueHandlerContainer = new DistributedQueueHandlerContainerService<AsynchContext<SendRequest,?>>();
            asynchSendQueueHandlerContainer.create();
            asynchSendQueueHandlerContainer.setDistributedQueueSelector(queueSelector);
            asynchSendQueueHandlerContainer.setQueueHandler(new SendQueueHandler());
            asynchSendQueueHandlerContainer.start();
        }
    }
    
    public void setLogger(Logger logger){
        this.logger = logger;
    }
    
    public void setSendErrorMessageId(String id){
        sendErrorMessageId = id;
    }
    
    public void setSendErrorRetryOverMessageId(String id){
        sendErrorRetryOverMessageId = id;
    }
    
    public void setSendBufferTime(long time){
        sendBufferTime = time;
    }
    
    public synchronized void connect(Object id, ClientConnectionImpl cc) throws ConnectException{
        ClientImpl old = clients.get(id);
        if(old != null){
            if(old.getClientConnection() == cc){
                return;
            }
            throw new ConnectException("Already exists. id=" + id + ", client=" + old);
        }
        ClientImpl client = new ClientImpl( cc);
        client.connect(id);
    }
    
    public void addSubject(Object id, String subject, String[] keys) throws MessageSendException{
        ClientImpl client = clients.get(id);
        if(client == null){
            throw new MessageSendException("No connected. id=" + id);
        }
        client.addSubject(subject, keys);
    }
    
    public void removeSubject(Object id, String subject, String[] keys) throws MessageSendException{
        ClientImpl client = clients.get(id);
        if(client == null){
            throw new MessageSendException("No connected. id=" + id);
        }
        client.removeSubject(subject, keys);
    }
    
    public void startReceive(Object id, long from) throws MessageSendException{
        ClientImpl client = clients.get(id);
        if(client == null){
            throw new MessageSendException("No connected. id=" + id);
        }
        client.startReceive(from);
    }
    
    public void stopReceive(Object id){
        ClientImpl client = clients.get(id);
        if(client != null){
            client.stopReceive();
        }
    }
    
    public void close(Object id){
        ClientImpl client = clients.get(id);
        if(client != null){
            client.close();
        }
    }
    
    public Message createMessage(String subject, String key) throws MessageCreateException{
        final MessageImpl message = new MessageImpl();
        message.setSubject(subject, key);
        return message;
    }
    
    public Message castMessage(Message message) throws MessageException{
        if(message instanceof MessageImpl){
            return message;
        }
        Message msg = createMessage(message.getSubject(), message.getKey());
        msg.setObject(message.getObject());
        return msg;
    }
    
    public void send(Message message) throws MessageSendException{
        addSendBuffer((MessageImpl)message);
        if(clients.size() == 0){
            return;
        }
        ClientImpl[] clientArray = clients.values().toArray(new ClientImpl[clients.size()]);
        if(sendQueueHandlerContainer == null){
            for(int i = 0; i < clientArray.length; i++){
                if(!clientArray[i].isStartReceive()
                    || !clientArray[i].isTargetMessage(message)){
                    continue;
                }
                clientArray[i].send((MessageImpl)message);
            }
        }else{
            DefaultQueueService<AsynchContext<SendRequest,Object>> responseQueue = new DefaultQueueService<AsynchContext<SendRequest,Object>>();
            try{
                responseQueue.create();
                responseQueue.start();
            }catch(Exception e){
                throw new MessageSendException(e);
            }
            responseQueue.accept();
            for(int i = 0; i < clientArray.length; i++){
                if(!clientArray[i].isStartReceive()
                    || !clientArray[i].isTargetMessage(message)){
                    clientArray[i] = null;
                    continue;
                }
                sendQueueHandlerContainer.push(new AsynchContext<SendRequest,Object>(new SendRequest(clientArray[i], (MessageImpl)message), responseQueue));
            }
            List<ClientImpl> errorClients = new ArrayList<ClientImpl>();
            for(int i = 0; i < clientArray.length; i++){
                if(clientArray[i] == null){
                    continue;
                }
                AsynchContext<SendRequest,Object> asynchContext = responseQueue.get();
                if(asynchContext.getThrowable() != null){
                    errorClients.add(asynchContext.getInput().client);
                }
            }
            if(errorClients.size() != 0){
                throw new MessageSendException("Send error : clients=" + errorClients + ", message=" + message);
            }
        }
    }
    
    public void sendAsynch(Message message){
        if(asynchSendQueueHandlerContainer == null){
            throw new UnsupportedOperationException();
        }
        addSendBuffer((MessageImpl)message);
        if(clients.size() == 0){
            return;
        }
        ClientImpl[] clientArray = (ClientImpl[])clients.values().toArray(new ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            if(!clientArray[i].isStartReceive()
                || !clientArray[i].isTargetMessage(message)){
                continue;
            }
            asynchSendQueueHandlerContainer.push(new AsynchContext<SendRequest,Object>(new SendRequest(clientArray[i], (MessageImpl)message)));
        }
    }
    
    public long getSendCount(){
        return sendCount;
    }
    
    public void resetSendCount(){
        sendCount = 0;
    }
    
    public Map<Object, ClientImpl> getClients(){
        return clients;
    }
    
    public int getClientCount(){
        return clients.size();
    }
    
    public Set<Object> getClientIds(){
        ClientImpl[] clientArray = clients.values().toArray(new ClientImpl[clients.size()]);
        Set<Object> result = new HashSet<Object>();
        for(int i = 0; i < clientArray.length; i++){
            result.add(clientArray[i].getId());
        }
        return result;
    }
    
    public Set<Object> getReceiveClientIds(Message message){
        ClientImpl[] clientArray = clients.values().toArray(new ClientImpl[clients.size()]);
        Set<Object> result = new HashSet<Object>();
        for(int i = 0; i < clientArray.length; i++){
            if(clientArray[i].isTargetMessage(message)){
                result.add(clientArray[i].getId());
            }
        }
        return result;
    }
    
    public Set<String> getSubjects(Object id){
        ClientImpl client = clients.get(id);
        if(client == null){
            return null;
        }
        return client.getSubjects();
    }
    
    public Set<String> getKeys(Object id, String subject){
        ClientImpl client = clients.get(id);
        if(client == null){
            return null;
        }
        return client.getKeys(subject);
    }
    
    private void addSendBuffer(MessageImpl message){
        final long currentTime = System.currentTimeMillis();
        message.setSendTime(currentTime);
        synchronized(sendBufferList){
            sendBufferList.add(message);
            for(int i = 0, imax = sendBufferList.size(); i < imax; i++){
                MessageImpl msg = sendBufferList.get(0);
                if((currentTime - msg.getSendTime()) > sendBufferTime){
                    sendBufferList.remove(0);
                }else{
                    break;
                }
            }
            sendCount++;
        }
    }
    
    private List<MessageImpl> getSendMessages(long from){
        List<MessageImpl> result = new ArrayList<MessageImpl>();
        synchronized(sendBufferList){
            for(int i = sendBufferList.size(); --i >= 0; ){
                MessageImpl msg = sendBufferList.get(i);
                if(msg.getSendTime() >= from){
                    result.add(0, msg);
                }else{
                    break;
                }
            }
        }
        return result;
    }
    
    public int getSendBufferSize(){
        return sendBufferList.size();
    }
    
    public void close(){
        try{
            send(new MessageImpl(true));
        }catch(MessageSendException e){}
        
        if(sendQueueHandlerContainer != null){
            sendQueueHandlerContainer.stop();
            sendQueueHandlerContainer.destroy();
            sendQueueHandlerContainer = null;
        }
        if(asynchSendQueueHandlerContainer != null){
            asynchSendQueueHandlerContainer.stop();
            asynchSendQueueHandlerContainer.destroy();
            asynchSendQueueHandlerContainer = null;
        }
        if(queueSelector != null){
            queueSelector.stop();
            queueSelector.destroy();
            queueSelector = null;
        }
        
        ClientImpl[] clientArray = clients.values().toArray(new ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            clientArray[i].close();
        }
    }
    
    public void addServerConnectionListener(ServerConnectionListener listener){
        if(serverConnectionListeners == null){
            serverConnectionListeners = new ArrayList<ServerConnectionListener>();
        }
        if(!serverConnectionListeners.contains(listener)){
            serverConnectionListeners.add(listener);
        }
    }
    
    public void removeServerConnectionListener(ServerConnectionListener listener){
        if(serverConnectionListeners == null){
            return;
        }
        serverConnectionListeners.remove(listener);
        if(serverConnectionListeners.size() == 0){
            serverConnectionListeners = null;
        }
    }
    
    public String toString(){
        final StringBuilder buf = new StringBuilder();
        buf.append(super.toString());
        buf.append('{');
        buf.append("server=").append(serverConnectionFactroyServiceName);
        buf.append('}');
        return buf.toString();
    }
    
    public class ClientImpl implements Client{
        private Object id;
        private ClientConnectionImpl clientConnection;
        private Map<String,Set<String>> subjects;
        private long sendCount;
        private boolean isEnabled = true;
        private long fromTime = -1;
        private boolean isStartReceive = false;
        
        public ClientImpl(ClientConnectionImpl cc){
            clientConnection = cc;
            subjects = Collections.synchronizedMap(new HashMap<String,Set<String>>());
        }
        
        public ClientConnectionImpl getClientConnection(){
            return clientConnection;
        }
        
        public boolean isEnabled(){
            return isEnabled;
        }
        public void setEnabled(boolean isEnabled){
            this.isEnabled = isEnabled;
        }
        
        public boolean isStartReceive(){
            return isStartReceive;
        }
        
        public boolean isTargetMessage(Message message){
            if(!message.containsDestinationId(getId())){
                return false;
            }
            if(message.getSubject() != null){
                Set<String> sbjs = message.getSubjects();
                for(String subject : sbjs){
                    Set<String> keySet = subjects.get(subject);
                    String key = message.getKey(subject);
                    if(keySet == null){
                        continue;
                    }else if(keySet.contains(null) || keySet.contains(key)){
                        return true;
                    }
                }
            }
            return false;
        }
        
        public synchronized void send(MessageImpl message){
            if(!isEnabled){
                return;
            }
            sendCount++;
            clientConnection.onMessage(message);
        }
        
        public long getSendCount(){
            return sendCount;
        }
        
        public void resetSendCount(){
            sendCount = 0;
        }
        
        public void connect(Object id){
            this.id = id;
            clients.put(id, ClientImpl.this);
            if(serverConnectionListeners != null){
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onConnect(ClientImpl.this);
                }
            }
        }
        
        public void addSubject(String subject, String[] keys){
            List<String> addKeysList = Collections.synchronizedList(new ArrayList<String>());
            Set<String> keySet = subjects.get(subject);
            if(keySet == null){
                keySet = Collections.synchronizedSet(new HashSet<String>());
                subjects.put(subject, keySet);
            }
            if(keys == null){
                if(keySet.add(null)){
                    addKeysList.add(null);
                }
            }else{
                for(int i = 0; i < keys.length; i++){
                    if(keySet.add(keys[i])){
                        addKeysList.add(keys[i]);
                    }
                }
            }
            if(serverConnectionListeners != null && !addKeysList.isEmpty()){
                String[] addkeys = addKeysList.toArray(new String[0]);
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onAddSubject(ClientImpl.this, subject, addkeys);
                }
            }
        }
        
        public void removeSubject(String subject, String[] keys){
            List<String> removeKeysList = Collections.synchronizedList(new ArrayList<String>());
            Set<String> keySet = subjects.get(subject);
            if(keySet == null){
                return;
            }
            if(keys == null){
                if(keySet.remove(null)){
                    removeKeysList.add(null);
                }
                if(keySet.size() == 0){
                    subjects.remove(subject);
                }
            }else{
                for(int i = 0; i < keys.length; i++){
                    if(keySet.remove(keys[i])){
                        removeKeysList.add(keys[i]);
                    }
                }
                if(keySet.size() == 0){
                    subjects.remove(subject);
                }
            }
            if(serverConnectionListeners != null && !removeKeysList.isEmpty()){
                String[] removeKeys = removeKeysList.toArray(new String[0]);
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onRemoveSubject(ClientImpl.this, subject, removeKeys);
                }
            }
        }
        
        public void startReceive(long from){
            if(fromTime >= 0){
                List<MessageImpl> messages = getSendMessages(fromTime);
                for(int i = 0; i < messages.size(); i++){
                    MessageImpl msg = messages.get(i);
                    send(msg);
                }
            }
            isStartReceive = true;
            if(serverConnectionListeners != null){
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onStartReceive(ClientImpl.this, fromTime);
                }
            }
        }
        
        public void stopReceive(){
            isStartReceive = false;
            if(serverConnectionListeners != null){
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onStopReceive(ClientImpl.this);
                }
            }
        }
        
        public synchronized void close(){
            clients.remove(id);
            if(serverConnectionListeners != null){
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onClose(ClientImpl.this);
                }
            }
            isStartReceive = false;
        }
        
        public String toString(){
            final StringBuilder buf = new StringBuilder();
            buf.append(super.toString());
            buf.append('{');
            buf.append("client=").append(id);
            buf.append(", subject=").append(subjects);
            buf.append(", isEnabled=").append(isEnabled);
            buf.append('}');
            return buf.toString();
        }
        
        public Set<String> getSubjects(){
            if(subjects == null){
                return null;
            }
            return subjects.keySet();
        }
        
        public Set<String> getKeys(String subject){
            if(subjects == null){
                return null;
            }
            return subjects.get(subject);
        }
        
        public Object getId(){
            return id;
        }
    }
    
    private class SendRequest{
        public ClientImpl client;
        public MessageImpl message;
        public SendRequest(ClientImpl client, MessageImpl message){
            this.client = client;
            this.message = message;
        }
    }
    
    private class SendQueueHandler implements QueueHandler<AsynchContext<SendRequest,?>>{
        
        public void handleDequeuedObject(AsynchContext<SendRequest,?> asynchContext) throws Throwable{
            if(asynchContext == null){
                return;
            }
            SendRequest request = asynchContext.getInput();
            if(request.client.isStartReceive()){
                request.client.send(request.message);
            }
            asynchContext.response();
        }
        
        public boolean handleError(AsynchContext<SendRequest,?> asynchContext, Throwable th) throws Throwable{
            if(logger != null && sendErrorMessageId != null){
                SendRequest request = asynchContext.getInput();
                logger.write(
                    sendErrorMessageId,
                    th,
                    request.client,
                    request.message
                );
            }
            return true;
        }
        
        public void handleRetryOver(AsynchContext<SendRequest,?> asynchContext, Throwable th) throws Throwable{
            if(logger != null && sendErrorRetryOverMessageId != null){
                SendRequest request = asynchContext.getInput();
                logger.write(
                    sendErrorRetryOverMessageId,
                    th,
                    request.client,
                    request.message
                );
            }
            asynchContext.setThrowable(th);
            asynchContext.response();
        }
    }
    
    private class ClientDistributedQueueSelector extends AbstractDistributedQueueSelectorService<AsynchContext<SendRequest,?>,ClientImpl>{
        
        private static final long serialVersionUID = 8988745179636312783L;
        
        protected ClientImpl getKey(AsynchContext<SendRequest,?> asynchContext){
            SendRequest request = asynchContext.getInput();
            return request.client;
        }
    }
}
