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

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Set;

import org.eclipse.jetty.websocket.WebSocket;

import jp.ossc.nimbus.core.ServiceBase;
import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.core.ServiceManagerFactory;
import jp.ossc.nimbus.beans.dataset.DataSet;
import jp.ossc.nimbus.beans.dataset.Header;
import jp.ossc.nimbus.beans.dataset.Record;
import jp.ossc.nimbus.beans.dataset.RecordList;
import jp.ossc.nimbus.util.converter.StreamConverter;
import jp.ossc.nimbus.util.converter.BindingStreamConverter;
import jp.ossc.nimbus.util.converter.ConvertException;
import jp.ossc.nimbus.util.converter.StringStreamConverter;
import jp.ossc.nimbus.service.publish.Message;
import jp.ossc.nimbus.service.publish.MessageListener;
import jp.ossc.nimbus.service.publish.MessageReceiver;
import jp.ossc.nimbus.service.websocket.Authenticator;
import jp.ossc.nimbus.service.websocket.AuthenticateException;
import jp.ossc.nimbus.service.websocket.jetty.PingSender;
import jp.ossc.nimbus.service.journal.Journal;
import jp.ossc.nimbus.service.journal.EditorFinder;
import jp.ossc.nimbus.service.context.Context;
import jp.ossc.nimbus.service.sequence.Sequence;

/**
 * org.eclipse.jetty.websocket.WebSocket実装サービス。<p>
 * WebSocketの通信をDataSetで行う。WebSocketへのリクエストで、{@link MessageReceiver}に配信の登録及び解除を行い、MessageReceiverから受信した{@link Message}をクライアントに配信する。<br>
 *
 * @author M.Takata
 */
public class DataSetWebSocketService extends ServiceBase
 implements WebSocket, WebSocket.OnFrame, WebSocket.OnControl,
            WebSocket.OnBinaryMessage, WebSocket.OnTextMessage,
            MessageListener, DataSetWebSocketServiceMBean{

    private static final long serialVersionUID = 2835025401033438035L;

    protected static final String PUBLISH_HEADER_PROPERTY_NAME_SUBJECT = "subject";
    protected static final String PUBLISH_HEADER_PROPERTY_NAME_KEY = "key";

    protected static final String PUBLISH_HEADER_SCHEMA
        = ':' + PUBLISH_HEADER_PROPERTY_NAME_SUBJECT + ",java.lang.String\n"
        + ':' + PUBLISH_HEADER_PROPERTY_NAME_KEY + ",java.lang.String";

    protected ServiceName messageReceiverServiceName;
    protected ServiceName reuqestDataSetConverterServiceName;
    protected ServiceName sendDataSetConverterServiceName;
    protected ServiceName pingSenderServiceName;
    protected ServiceName requestJournalServiceName;
    protected ServiceName requestEditorFinderServiceName;
    protected String requestIdKey = DEFAULT_REQUEST_ID_KEY;
    protected String requestJournalKey = DEFAULT_REQUEST_JOURNAL_KEY;
    protected String requestMessageJournalKey = DEFAULT_REQUEST_MESSAGE_JOURNAL_KEY;
    protected String requestDataSetJournalKey = DEFAULT_REQUEST_DATASET_JOURNAL_KEY;
    protected String idJournalKey = DEFAULT_ID_JOURNAL_KEY;
    protected String exceptionJournalKey = DEFAULT_EXCEPTION_JOURNAL_KEY;
    protected ServiceName sendJournalServiceName;
    protected ServiceName sendEditorFinderServiceName;
    protected String sendJournalKey = DEFAULT_SEND_JOURNAL_KEY;
    protected String sendMessageJournalKey = DEFAULT_SEND_MESSAGE_JOURNAL_KEY;
    protected String sendDataSetJournalKey = DEFAULT_SEND_DATASET_JOURNAL_KEY;
    protected String sendBytesJournalKey = DEFAULT_SEND_BYTES_JOURNAL_KEY;
    protected ServiceName sequenceServiceName;
    protected ServiceName contextServiceName;
    protected ServiceName authenticatorServiceName;
    protected String characterEncoding = "UTF-8";
    protected int sendMode = SEND_MODE_STRING;
    protected boolean isAppendPublishHeader = true;
    protected String publishHeaderName = PUBLISH_HEADER_DEFAULT_NAME;
    protected boolean isSendBySubject = true;

    protected int maxBinaryMessageSize;
    protected int maxTextMessageSize;
    protected int maxIdleTime;

    protected String requestConvertErrorMessageId = MSG_ID_REQUEST_CONVERT_ERROR;
    protected String unknownRequestObjectErrorMessageId = MSG_ID_UNKNOWN_REQUEST_OBJECT_ERROR;
    protected String noAuthenticatedErrorMessageId = MSG_ID_NO_AUTHENTICATED_ERROR;
    protected String requestProcessErrorMessageId = MSG_ID_REQUEST_PROCESS_ERROR;
    protected String sendMessageConvertErrorMessageId = MSG_ID_SEND_MESSAGE_CONVERT_ERROR;
    protected String sendMessageErrorMessageId = MSG_ID_SEND_MESSAGE_ERROR;
    protected String sendPongErrorMessageId = MSG_ID_SEND_PONG_ERROR;
    protected String logoutErrorMessageId = MSG_ID_LOGOUT_ERROR;

    protected int closeCodeForRequestConvertError = 1003;
    protected int closeCodeForUnknownRequestObjectError = 1011;
    protected int closeCodeForNoAuthenticatedError = 1008;
    protected int closeCodeForRequestProcessError = 1011;
    protected int closeCodeForSendMessageError = 1001;

    protected MessageReceiver messageReceiver;
    protected BindingStreamConverter requestDataSetConverter;
    protected StreamConverter sendDataSetConverter;
    protected Journal requestJournal;
    protected EditorFinder requestEditorFinder;
    protected Journal sendJournal;
    protected EditorFinder sendEditorFinder;
    protected Sequence sequence;
    protected Context<String, Object> context;
    protected StringStreamConverter stringStreamConverter;
    protected WebSocket.Connection connection;
    protected WebSocket.FrameConnection frameConnection;
    protected PingSender pingSender;
    protected Authenticator authenticator;
    protected long receiveRequestCount;
    protected long receiveMessageCount;
    protected long sendMessageCount;
    protected String id;

    @Override
    public void setMessageReceiverServiceName(ServiceName name){
        messageReceiverServiceName = name;
    }
    @Override
    public ServiceName getMessageReceiverServiceName(){
        return messageReceiverServiceName;
    }

    @Override
    public void setRequestDataSetConverterServiceName(ServiceName name){
        reuqestDataSetConverterServiceName = name;
    }
    @Override
    public ServiceName getRequestDataSetConverterServiceName(){
        return reuqestDataSetConverterServiceName;
    }

    @Override
    public void setSendDataSetConverterServiceName(ServiceName name){
        sendDataSetConverterServiceName = name;
    }
    @Override
    public ServiceName getSendDataSetConverterServiceName(){
        return sendDataSetConverterServiceName;
    }

    @Override
    public void setPingSenderServiceName(ServiceName name){
        pingSenderServiceName = name;
    }
    @Override
    public ServiceName getPingSenderServiceName(){
        return pingSenderServiceName;
    }

    @Override
    public void setRequestJournalServiceName(ServiceName name){
        requestJournalServiceName = name;
    }
    @Override
    public ServiceName getRequestJournalServiceName(){
        return requestJournalServiceName;
    }

    @Override
    public void setRequestEditorFinderServiceName(ServiceName name){
        requestEditorFinderServiceName = name;
    }
    @Override
    public ServiceName getRequestEditorFinderServiceName(){
        return requestEditorFinderServiceName;
    }

    @Override
    public void setRequestIDKey(String key){
        requestIdKey = key;
    }
    @Override
    public String getRequestIDKey(){
        return requestIdKey;
    }

    @Override
    public void setRequestJournalKey(String key){
        requestJournalKey = key;
    }
    @Override
    public String getRequestJournalKey(){
        return requestJournalKey;
    }

    @Override
    public void setRequestMessageJournalKey(String key){
        requestMessageJournalKey = key;
    }
    @Override
    public String getRequestMessageJournalKey(){
        return requestMessageJournalKey;
    }

    @Override
    public void setRequestDataSetJournalKey(String key){
        requestDataSetJournalKey = key;
    }
    @Override
    public String getRequestDataSetJournalKey(){
        return requestDataSetJournalKey;
    }

    @Override
    public void setIdJournalKey(String key){
        idJournalKey = key;
    }
    @Override
    public String getIdJournalKey(){
        return idJournalKey;
    }

    @Override
    public void setExceptionJournalKey(String key){
        exceptionJournalKey = key;
    }
    @Override
    public String getExceptionJournalKey(){
        return exceptionJournalKey;
    }

    @Override
    public void setSendJournalServiceName(ServiceName name){
        sendJournalServiceName = name;
    }
    @Override
    public ServiceName getSendJournalServiceName(){
        return sendJournalServiceName;
    }

    @Override
    public void setSendEditorFinderServiceName(ServiceName name){
        sendEditorFinderServiceName = name;
    }
    @Override
    public ServiceName getSendEditorFinderServiceName(){
        return sendEditorFinderServiceName;
    }

    @Override
    public void setSendJournalKey(String key){
        sendJournalKey = key;
    }
    @Override
    public String getSendJournalKey(){
        return sendJournalKey;
    }

    @Override
    public void setSendMessageJournalKey(String key){
        sendMessageJournalKey = key;
    }
    @Override
    public String getSendMessageJournalKey(){
        return sendMessageJournalKey;
    }

    @Override
    public void setSendDataSetJournalKey(String key){
        sendDataSetJournalKey = key;
    }
    @Override
    public String getSendDataSetJournalKey(){
        return sendDataSetJournalKey;
    }

    @Override
    public void setSendBytesJournalKey(String key){
        sendBytesJournalKey = key;
    }
    @Override
    public String getSendBytesJournalKey(){
        return sendBytesJournalKey;
    }

    @Override
    public void setSequenceServiceName(ServiceName name){
        sequenceServiceName = name;
    }
    @Override
    public ServiceName getSequenceServiceName(){
        return sequenceServiceName;
    }

    @Override
    public void setContextServiceName(ServiceName name){
        contextServiceName = name;
    }
    @Override
    public ServiceName getContextServiceName(){
        return contextServiceName;
    }

    @Override
    public void setAuthenticatorServiceName(ServiceName name){
        authenticatorServiceName = name;
    }
    @Override
    public ServiceName getAuthenticatorServiceName(){
        return authenticatorServiceName;
    }

    @Override
    public void setCharacterEncoding(String encoding){
        characterEncoding = encoding;
    }
    @Override
    public String getCharacterEncoding(){
        return characterEncoding;
    }

    @Override
    public void setSendMode(int mode) throws IllegalArgumentException{
        switch(mode){
        case SEND_MODE_STRING:
        case SEND_MODE_BINARY:
            break;
        default:
            throw new IllegalArgumentException("Illegal send mode. mode=" + mode);
        }
        sendMode = mode;
    }
    @Override
    public int getSendMode(){
        return sendMode;
    }

    @Override
    public void setAppendPublishHeader(boolean isAppend){
        isAppendPublishHeader = isAppend;
    }
    @Override
    public boolean isAppendPublishHeader(){
        return isAppendPublishHeader;
    }

    @Override
    public void setPublishHeaderName(String name){
        publishHeaderName = name;
    }
    @Override
    public String getPublishHeaderName(){
        return publishHeaderName;
    }

    @Override
    public void setSendBySubject(boolean isSend){
        isSendBySubject = isSend;
    }
    @Override
    public boolean isSendBySubject(){
        return isSendBySubject;
    }
    @Override
    public Set<String> getSubjects(){
        return messageReceiver == null ? null : messageReceiver.getSubjects(this);
    }
    @Override
    public void setMaxBinaryMessageSize(int size){
        maxBinaryMessageSize = size;
    }
    @Override
    public int getMaxBinaryMessageSize(){
        return connection == null ? maxBinaryMessageSize : connection.getMaxBinaryMessageSize();
    }

    @Override
    public void setMaxTextMessageSize(int size){
        maxTextMessageSize = size;
    }
    @Override
    public int getMaxTextMessageSize(){
        return connection == null ? maxTextMessageSize : connection.getMaxTextMessageSize();
    }

    @Override
    public void setMaxIdleTime(int ms){
        maxIdleTime = ms;
    }
    @Override
    public int getMaxIdleTime(){
        return connection == null ? maxIdleTime : connection.getMaxIdleTime();
    }

    @Override
    public void setRequestConvertErrorMessageId(String id){
        requestConvertErrorMessageId = id;
    }
    @Override
    public String getRequestConvertErrorMessageId(){
        return requestConvertErrorMessageId;
    }

    @Override
    public void setUnknownRequestObjectErrorMessageId(String id){
        unknownRequestObjectErrorMessageId = id;
    }
    @Override
    public String getUnknownRequestObjectErrorMessageId(){
        return unknownRequestObjectErrorMessageId;
    }

    @Override
    public void setNoAuthenticatedErrorMessageId(String id){
        noAuthenticatedErrorMessageId = id;
    }
    @Override
    public String getNoAuthenticatedErrorMessageId(){
        return noAuthenticatedErrorMessageId;
    }

    @Override
    public void setRequestProcessErrorMessageId(String id){
        requestProcessErrorMessageId = id;
    }
    @Override
    public String getRequestProcessErrorMessageId(){
        return requestProcessErrorMessageId;
    }

    @Override
    public void setSendMessageConvertErrorMessageId(String id){
        sendMessageConvertErrorMessageId = id;
    }
    @Override
    public String getSendMessageConvertErrorMessageId(){
        return sendMessageConvertErrorMessageId;
    }

    @Override
    public void setSendMessageErrorMessageId(String id){
        sendMessageErrorMessageId = id;
    }
    @Override
    public String getSendMessageErrorMessageId(){
        return sendMessageErrorMessageId;
    }

    @Override
    public void setSendPongErrorMessageId(String id){
        sendPongErrorMessageId = id;
    }
    @Override
    public String getSendPongErrorMessageId(){
        return sendPongErrorMessageId;
    }

    @Override
    public void setLogoutErrorMessageId(String id){
        logoutErrorMessageId = id;
    }
    @Override
    public String getLogoutErrorMessageId(){
        return logoutErrorMessageId;
    }

    @Override
    public void setCloseCodeForRequestConvertError(int code){
        closeCodeForRequestConvertError = code;
    }
    @Override
    public int getCloseCodeForRequestConvertError(){
        return closeCodeForRequestConvertError;
    }

    @Override
    public void setCloseCodeForUnknownRequestObjectError(int code){
        closeCodeForUnknownRequestObjectError = code;
    }
    @Override
    public int getCloseCodeForUnknownRequestObjectError(){
        return closeCodeForUnknownRequestObjectError;
    }

    @Override
    public void setCloseCodeForNoAuthenticatedError(int code){
        closeCodeForNoAuthenticatedError = code;
    }
    @Override
    public int getCloseCodeForNoAuthenticatedError(){
        return closeCodeForNoAuthenticatedError;
    }

    @Override
    public void setCloseCodeForRequestProcessError(int code){
        closeCodeForRequestProcessError = code;
    }
    @Override
    public int getCloseCodeForRequestProcessError(){
        return closeCodeForRequestProcessError;
    }

    @Override
    public void setCloseCodeForSendMessageError(int code){
        closeCodeForSendMessageError = code;
    }
    @Override
    public int getCloseCodeForSendMessageError(){
        return closeCodeForSendMessageError;
    }

    @Override
    public long getReceiveRequestCount(){
        return receiveRequestCount;
    }

    @Override
    public long getReceiveMessageCount(){
        return receiveMessageCount;
    }

    @Override
    public long getSendMessageCount(){
        return sendMessageCount;
    }

    @Override
    public void resetCount(){
        receiveRequestCount = 0;
        receiveMessageCount = 0;
        sendMessageCount = 0;
    }

    @Override
    public String getProtocol(){
        return connection == null ? null : connection.getProtocol();
    }

    @Override
    public String getId(){
        return id;
    }

    public void setMessageReceiver(MessageReceiver receiver){
        messageReceiver = receiver;
    }

    public void setRequestDataSetConverter(BindingStreamConverter converter){
        requestDataSetConverter = converter;
    }

    public void setSendDataSetConverter(StreamConverter converter){
        sendDataSetConverter = converter;
    }

    public void setPingSender(PingSender sender){
        pingSender = sender;
    }

    public void setRequestJournal(Journal journal){
        requestJournal = journal;
    }

    public void setRequestEditorFinder(EditorFinder finder){
        requestEditorFinder = finder;
    }

    public void setSendJournal(Journal journal){
        sendJournal = journal;
    }

    public void setSendEditorFinder(EditorFinder finder){
        sendEditorFinder = finder;
    }

    public void setSequence(Sequence seq){
        sequence = seq;
    }

    public void setContext(Context<String, Object> ctx){
        context = ctx;
    }

    public void setAuthenticator(Authenticator auth){
        authenticator = auth;
    }

    @Override
    public void startService() throws Exception{
        if(messageReceiverServiceName != null){
            messageReceiver = ServiceManagerFactory.getServiceObject(messageReceiverServiceName);
        }
        if(messageReceiver == null){
            throw new IllegalArgumentException("MessageReceiver is null.");
        }
        if(reuqestDataSetConverterServiceName != null){
            requestDataSetConverter = ServiceManagerFactory.getServiceObject(reuqestDataSetConverterServiceName);
        }
        if(requestDataSetConverter == null){
            throw new IllegalArgumentException("RequestDataSetConverter is null.");
        }
        if(sendDataSetConverterServiceName != null){
            sendDataSetConverter = ServiceManagerFactory.getServiceObject(sendDataSetConverterServiceName);
        }
        if(sendDataSetConverter == null){
            throw new IllegalArgumentException("SendDataSetConverter is null.");
        }
        if(pingSenderServiceName != null){
            pingSender = ServiceManagerFactory.getServiceObject(pingSenderServiceName);
        }
        if(requestJournalServiceName != null){
            requestJournal = ServiceManagerFactory.getServiceObject(requestJournalServiceName);
        }
        if(requestEditorFinderServiceName != null){
            requestEditorFinder = ServiceManagerFactory.getServiceObject(requestEditorFinderServiceName);
        }
        if(sendJournalServiceName != null){
            sendJournal = ServiceManagerFactory.getServiceObject(sendJournalServiceName);
        }
        if(sendEditorFinderServiceName != null){
            sendEditorFinder = ServiceManagerFactory.getServiceObject(sendEditorFinderServiceName);
        }
        if(sequenceServiceName != null){
            sequence = ServiceManagerFactory.getServiceObject(sequenceServiceName);
        }
        if(contextServiceName != null){
            context = ServiceManagerFactory.getServiceObject(contextServiceName);
        }
        if(authenticatorServiceName != null){
            authenticator = ServiceManagerFactory.getServiceObject(authenticatorServiceName);
        }
        stringStreamConverter = new StringStreamConverter();
        if(characterEncoding != null){
            stringStreamConverter.setCharacterEncodingToStream(characterEncoding);
        }
    }

    @Override
    public void onOpen(WebSocket.Connection connection){
        synchronized(this){
            this.connection = connection;
            if(maxBinaryMessageSize > 0){
                connection.setMaxBinaryMessageSize(maxBinaryMessageSize);
            }
            if(maxTextMessageSize > 0){
                connection.setMaxTextMessageSize(maxTextMessageSize);
            }
            if(maxIdleTime > 0){
                connection.setMaxIdleTime(maxIdleTime);
            }
        }
        if(pingSender != null && this.connection != null && frameConnection != null){
            pingSender.regist(connection, frameConnection);
        }
    }

    @Override
    public void onHandshake(WebSocket.FrameConnection connection){
        frameConnection = connection;
        if(pingSender != null && this.connection != null && frameConnection != null){
            pingSender.regist(connection, frameConnection);
        }
    }

    @Override
    public boolean onFrame(byte flags, byte opcode, byte[] data, int offset, int length){
        return false;
    }

    public boolean onControl(byte controlCode, byte[] data, int offset, int length){
        if(frameConnection == null){
            return false;
        }
        if(frameConnection.isPing(controlCode)){
            onPing(data, offset, length);
        }else if(frameConnection.isPong(controlCode)){
            onPong(data, offset, length);
        }
        return false;
    }

    protected void onPing(byte[] data, int offset, int length){
        sendPong(null, 0, 0);
    }

    protected void sendPong(byte[] data, int offset, int length){
        if(frameConnection.isPong((byte)0x03)){
            try{
                frameConnection.sendControl((byte)0x03, data, offset, length);
            }catch(IOException e){
                getLogger().write(sendPongErrorMessageId, e);
            }
        }else if(frameConnection.isPong((byte)0x0A)){
            try{
                frameConnection.sendControl((byte)0x0A, data, offset, length);
            }catch(IOException e){
                getLogger().write(sendPongErrorMessageId, e);
            }
        }
    }

    protected void onPong(byte[] data, int offset, int length){
        if(pingSender != null && connection != null){
            pingSender.onPong(connection, data, offset, length);
        }
    }

    @Override
    public void onClose(int closeCode, String message){
        if(pingSender != null && connection != null){
            pingSender.unregist(connection);
        }
        try{
            messageReceiver.removeMessageListener(this);
        }catch(Exception e){
        }
        if(authenticator != null && id != null){
            try{
                authenticator.logout(id, closeCode == 1000 ? false : true);
            }catch(AuthenticateException e){
                getLogger().write(logoutErrorMessageId, e);
            }
        }
        synchronized(this){
            connection = null;
            frameConnection = null;
        }
        stop();
        destroy();
    }

    @Override
    public void onMessage(byte[] data, int offset, int length){
        onMessage(new ByteArrayInputStream(data, offset, length), data);
    }

    @Override
    public void onMessage(String data){
        onMessage(stringStreamConverter.convertToStream(data), data);
    }

    protected void onMessage(InputStream is, Object data){
        receiveRequestCount++;
        try{
            if(requestJournal != null){
                requestJournal.startJournal(requestJournalKey, requestEditorFinder);
                if(sequence != null){
                    String sequenceId = sequence.increment();
                    if(context != null){
                        context.put(requestIdKey, sequenceId);
                    }
                    requestJournal.setRequestId(sequenceId);
                }else if(context != null){
                    requestJournal.setRequestId(
                        (String)context.get(requestIdKey)
                    );
                }
                if(id != null){
                    requestJournal.addInfo(idJournalKey, id);
                }
                requestJournal.addInfo(requestMessageJournalKey, data);
            }
            Object message = null;
            try{
                message = requestDataSetConverter.convertToObject(is);
            }catch(ConvertException e){
                getLogger().write(requestConvertErrorMessageId, e, data);
                if(requestJournal != null){
                    requestJournal.addInfo(exceptionJournalKey, e);
                }
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForRequestConvertError, null);
                        connection = null;
                    }
                }
                return;
            }
            if(!(message instanceof DataSet)){
                getLogger().write(unknownRequestObjectErrorMessageId, message.getClass());
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForUnknownRequestObjectError, null);
                        connection = null;
                    }
                }
                return;
            }
            onMessage((DataSet)message);
        }finally{
            if(requestJournal != null){
                requestJournal.endJournal();
            }
        }
    }

    protected void onMessage(DataSet ds){
        String name = ds.getName();
        if(requestJournal != null){
            requestJournal.addInfo(requestDataSetJournalKey, ds);
        }
        if(AuthenticateDataSet.NAME.equals(name)){
            try{
                id = onReceiveAuthenticate(ds);
                if(id != null){
                    requestJournal.addInfo(idJournalKey, id);
                }
            }catch(Exception e){
                getLogger().write(
                    requestProcessErrorMessageId,
                    e,
                    "authenticate",
                    ds.getHeader(AuthenticateDataSet.HEADER_AUTH)
                );
                if(requestJournal != null){
                    requestJournal.addInfo(exceptionJournalKey, e);
                }
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForRequestProcessError, null);
                        connection = null;
                    }
                }
                return;
            }
        }else if(AddSubjectDataSet.NAME.equals(name)){
            if(!isAuthenticated()){
                getLogger().write(noAuthenticatedErrorMessageId, ds);
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForNoAuthenticatedError, null);
                        connection = null;
                    }
                }
                return;
            }
            try{
                onReceiveAddSubject(ds);
            }catch(Exception e){
                getLogger().write(
                    requestProcessErrorMessageId,
                    e,
                    "addSubject",
                    ds.getHeader(AddSubjectDataSet.HEADER_ADD_SUBJECT)
                );
                if(requestJournal != null){
                    requestJournal.addInfo(exceptionJournalKey, e);
                }
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForRequestProcessError, null);
                        connection = null;
                    }
                }
                return;
            }
        }else if(RemoveSubjectDataSet.NAME.equals(name)){
            if(!isAuthenticated()){
                getLogger().write(noAuthenticatedErrorMessageId, ds);
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForNoAuthenticatedError, null);
                        connection = null;
                    }
                }
                return;
            }
            try{
                onReceiveRemoveSubject(ds);
            }catch(Exception e){
                getLogger().write(
                    requestProcessErrorMessageId,
                    e,
                    "removeSubject",
                    ds.getHeader(RemoveSubjectDataSet.HEADER_REMOVE_SUBJECT)
                );
                if(requestJournal != null){
                    requestJournal.addInfo(exceptionJournalKey, e);
                }
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForRequestProcessError, null);
                        connection = null;
                    }
                }
                return;
            }
        }else if(ByeDataSet.NAME.equals(name)){
            onReceiveBye(ds);
        }else{
            try{
                onReceiveExtentionDataSet(ds);
            }catch(Exception e){
                getLogger().write(
                    requestProcessErrorMessageId,
                    e,
                    "extentionRequest",
                    ds
                );
                if(requestJournal != null){
                    requestJournal.addInfo(exceptionJournalKey, e);
                }
                synchronized(this){
                    if(connection != null){
                        connection.close(closeCodeForRequestProcessError, null);
                        connection = null;
                    }
                }
                return;
            }
        }
    }

    protected String onReceiveAuthenticate(DataSet ds) throws Exception{
        Header header = ds.getHeader(AuthenticateDataSet.HEADER_AUTH);
        if(authenticator != null){
            boolean isSuccess = authenticator.login(
                header.getStringProperty(AuthenticateDataSet.HEADER_AUTH_PROPERTY_ID),
                header.getStringProperty(AuthenticateDataSet.HEADER_AUTH_PROPERTY_TICKET)
            );
            return isSuccess ? header.getStringProperty(AuthenticateDataSet.HEADER_AUTH_PROPERTY_ID) : null;
        }
        return header.getStringProperty(AuthenticateDataSet.HEADER_AUTH_PROPERTY_ID);
    }

    protected boolean isAuthenticated(){
        return authenticator == null ? true : id != null;
    }

    protected void onReceiveAddSubject(DataSet ds) throws Exception{
        Header header = ds.getHeader(AddSubjectDataSet.HEADER_ADD_SUBJECT);
        messageReceiver.addSubject(
            this,
            header.getStringProperty(AddSubjectDataSet.HEADER_ADD_SUBJECT_PROPERTY_SUBJECT),
            (String[])header.getProperty(AddSubjectDataSet.HEADER_ADD_SUBJECT_PROPERTY_KEYS)
        );
    }

    protected void onReceiveRemoveSubject(DataSet ds) throws Exception{
        Header header = ds.getHeader(RemoveSubjectDataSet.HEADER_REMOVE_SUBJECT);
        messageReceiver.removeSubject(
            this,
            header.getStringProperty(RemoveSubjectDataSet.HEADER_REMOVE_SUBJECT_PROPERTY_SUBJECT),
            (String[])header.getProperty(RemoveSubjectDataSet.HEADER_REMOVE_SUBJECT_PROPERTY_KEYS)
        );
    }

    protected void onReceiveBye(DataSet ds){
        synchronized(this){
            if(connection != null){
                connection.close();
                connection = null;
            }
        }
    }

    protected void onReceiveExtentionDataSet(DataSet ds) throws Exception{
        throw new UnsupportedOperationException("Unsupported DataSet. dataSet=" + ds);
    }

    @Override
    public void onMessage(Message message){
        receiveMessageCount++;

        Set<String> subjects = message.getSubjects();
        if(subjects.size() <= 1){
            try{
                if(sendJournal != null){
                    sendJournal.startJournal(sendJournalKey, sendEditorFinder);
                    sendJournal.addInfo(sendMessageJournalKey, message);
                    if(id != null){
                        sendJournal.addInfo(idJournalKey, id);
                    }
                }
                DataSet ds = null;
                try{
                    ds = messageToDataSet(message.getSubject(), message);
                }catch(Exception e){
                    getLogger().write(sendMessageConvertErrorMessageId, e, message);
                    if(sendJournal != null){
                        sendJournal.addInfo(exceptionJournalKey, e);
                    }
                    return;
                }
                if(isAppendPublishHeader){
                    ds = appendPublishHeader(message.getSubject(), message, ds);
                }
                send(ds);
            }finally{
                if(sendJournal != null){
                    sendJournal.endJournal();
                }
            }
        }else{
            if(isSendBySubject){
                for(String subject : subjects){
                    Set<String> keys = messageReceiver.getKeys(this, subject);
                    if(keys == null || !keys.contains(message.getKey(subject))){
                        continue;
                    }
                    try{
                        if(sendJournal != null){
                            sendJournal.startJournal(sendJournalKey, sendEditorFinder);
                            sendJournal.addInfo(sendMessageJournalKey, message);
                            if(id != null){
                                sendJournal.addInfo(idJournalKey, id);
                            }
                        }
                        DataSet ds = null;
                        try{
                            ds = messageToDataSet(subject, message);
                        }catch(Exception e){
                            getLogger().write(sendMessageConvertErrorMessageId, e, message);
                            if(sendJournal != null){
                                sendJournal.addInfo(exceptionJournalKey, e);
                            }
                            continue;
                        }
                        if(isAppendPublishHeader){
                            ds = appendPublishHeader(subject, message, ds);
                        }
                        send(ds);
                    }finally{
                        if(sendJournal != null){
                            sendJournal.endJournal();
                        }
                    }
                }
            }else{
                try{
                    if(sendJournal != null){
                        sendJournal.startJournal(sendJournalKey, sendEditorFinder);
                        sendJournal.addInfo(sendMessageJournalKey, message);
                        if(id != null){
                            sendJournal.addInfo(idJournalKey, id);
                        }
                    }
                    DataSet ds = null;
                    try{
                        ds = messageToDataSet(message.getSubject(), message);
                    }catch(Exception e){
                        getLogger().write(sendMessageConvertErrorMessageId, e, message);
                        if(sendJournal != null){
                            sendJournal.addInfo(exceptionJournalKey, e);
                        }
                        return;
                    }
                    if(isAppendPublishHeader){
                        for(String subject : subjects){
                            if(isAppendPublishHeader){
                                ds = appendPublishHeader(subject, message, ds);
                            }
                        }
                    }
                    send(ds);
                }finally{
                    if(sendJournal != null){
                        sendJournal.endJournal();
                    }
                }
            }
        }
    }

    protected DataSet appendPublishHeader(String subject, Message message, DataSet ds){
        if(isSendBySubject){
            Header header = ds.getHeader(publishHeaderName);
            if(header == null){
                ds.setHeaderSchema(publishHeaderName, PUBLISH_HEADER_SCHEMA);
                header = ds.getHeader(publishHeaderName);
            }
            header.setProperty(PUBLISH_HEADER_PROPERTY_NAME_SUBJECT, subject);
            header.setProperty(PUBLISH_HEADER_PROPERTY_NAME_KEY, message.getKey(subject));
        }else{
            RecordList list = ds.getRecordList(publishHeaderName);
            if(list == null){
                ds.setRecordListSchema(publishHeaderName, PUBLISH_HEADER_SCHEMA);
                list = ds.getRecordList(publishHeaderName);
            }
            Record record = list.createRecord();
            record.setProperty(PUBLISH_HEADER_PROPERTY_NAME_SUBJECT, subject);
            record.setProperty(PUBLISH_HEADER_PROPERTY_NAME_KEY, message.getKey(subject));
            list.addRecord(record);
        }
        return ds;
    }

    protected DataSet messageToDataSet(String subject, Message message) throws Exception{
        return (DataSet)message.getObject();
    }

    protected void send(DataSet ds){
        if(sendJournal != null){
            sendJournal.addInfo(sendDataSetJournalKey, ds);
        }
        byte[] bytes = null;
        try{
            InputStream is = sendDataSetConverter.convertToStream(ds);
            bytes = toBytes(is);
            if(bytes == null || bytes.length == 0){
                return;
            }
        }catch(ConvertException e){
            getLogger().write(sendMessageConvertErrorMessageId, e, ds);
            return;
        }catch(IOException e){
            getLogger().write(sendMessageConvertErrorMessageId, e, ds);
            return;
        }
        try{
            send(bytes);
        }catch(IOException e){
            getLogger().write(sendMessageErrorMessageId, e, ds);
            return;
        }
    }

    protected void send(byte[] bytes) throws IOException{
        if(sendJournal != null){
            sendJournal.addInfo(sendBytesJournalKey, bytes);
        }
        try{
            switch(sendMode){
            case SEND_MODE_BINARY:
                synchronized(this){
                    if(connection != null){
                        connection.sendMessage(bytes, 0, bytes.length);
                    }
                }
                break;
            case SEND_MODE_STRING:
            default:
                synchronized(this){
                    if(connection != null){
                        connection.sendMessage(
                            characterEncoding == null ? new String(bytes) : new String(bytes, characterEncoding)
                        );
                    }
                }
            }
            sendMessageCount++;
        }catch(IOException e){
            if(sendJournal != null){
                sendJournal.addInfo(exceptionJournalKey, e);
            }
            throw e;
        }
    }

    protected byte[] toBytes(InputStream is) throws IOException{
        byte[] bytes = new byte[1024];
        final ByteArrayOutputStream baos = new ByteArrayOutputStream();
        int length = 0;
        while((length = is.read(bytes)) != -1){
            baos.write(bytes, 0, length);
        }
        return baos.toByteArray();
    }

    public static class AuthenticateDataSet extends DataSet{

        private static final long serialVersionUID = -1186408833828282199L;

        public static final String NAME = "Authenticate";

        public static final String HEADER_AUTH = "Authenticate";

        public static final String HEADER_AUTH_PROPERTY_ID = "id";
        public static final String HEADER_AUTH_PROPERTY_TICKET = "ticket";

        public static final String HEADER_SCHEMA_AUTH
            = ':' + HEADER_AUTH_PROPERTY_ID + ",java.lang.String\n"
            + ':' + HEADER_AUTH_PROPERTY_TICKET + ",java.lang.String"
            ;

        public AuthenticateDataSet(){
            setName(NAME);
            setHeaderSchema(HEADER_AUTH, HEADER_SCHEMA_AUTH);
        }
    }

    public static class AddSubjectDataSet extends DataSet{

        private static final long serialVersionUID = -3694906611138615108L;

        public static final String NAME = "AddSubject";

        public static final String HEADER_ADD_SUBJECT = "AddSubject";

        public static final String HEADER_ADD_SUBJECT_PROPERTY_SUBJECT = "subject";
        public static final String HEADER_ADD_SUBJECT_PROPERTY_KEYS = "keys";

        public static final String HEADER_SCHEMA_ADD_SUBJECT
            = ':' + HEADER_ADD_SUBJECT_PROPERTY_SUBJECT + ",java.lang.String\n"
            + ':' + HEADER_ADD_SUBJECT_PROPERTY_KEYS + ",java.lang.String[]"
            ;

        public AddSubjectDataSet(){
            setName(NAME);
            setHeaderSchema(HEADER_ADD_SUBJECT, HEADER_SCHEMA_ADD_SUBJECT);
        }
    }

    public static class RemoveSubjectDataSet extends DataSet{

        private static final long serialVersionUID = -252165941836313576L;

        public static final String NAME = "RemoveSubject";

        public static final String HEADER_REMOVE_SUBJECT = "RemoveSubject";

        public static final String HEADER_REMOVE_SUBJECT_PROPERTY_SUBJECT = "subject";
        public static final String HEADER_REMOVE_SUBJECT_PROPERTY_KEYS = "keys";

        public static final String HEADER_SCHEMA_REMOVE_SUBJECT
            = ':' + HEADER_REMOVE_SUBJECT_PROPERTY_SUBJECT + ",java.lang.String\n"
            + ':' + HEADER_REMOVE_SUBJECT_PROPERTY_KEYS + ",java.lang.String[]"
            ;

        public RemoveSubjectDataSet(){
            setName(NAME);
            setHeaderSchema(HEADER_REMOVE_SUBJECT, HEADER_SCHEMA_REMOVE_SUBJECT);
        }
    }

    public static class ByeDataSet extends DataSet{

        private static final long serialVersionUID = -5788398695541333870L;

        public static final String NAME = "Bye";

        public ByeDataSet(){
            setName(NAME);
        }
    }
}