/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alipay.remoting.rpc;

import java.util.concurrent.RejectedExecutionException;

import com.alipay.remoting.*;
import org.slf4j.Logger;

import com.alipay.remoting.exception.CodecException;
import com.alipay.remoting.exception.ConnectionClosedException;
import com.alipay.remoting.log.BoltLoggerFactory;
import com.alipay.remoting.rpc.exception.InvokeException;
import com.alipay.remoting.rpc.exception.InvokeServerBusyException;
import com.alipay.remoting.rpc.exception.InvokeServerException;
import com.alipay.remoting.rpc.exception.InvokeTimeoutException;
import com.alipay.remoting.rpc.protocol.RpcResponseCommand;

/**
 * Listener which listens the Rpc invoke result, and then invokes the call back.
 *
 * @author jiangping
 * @version $Id: RpcInvokeCallbackListener.java, v 0.1 2015-9-30 AM10:36:34 tao Exp $
 */
public class RpcInvokeCallbackListener implements InvokeCallbackListener {

    private static final Logger logger = BoltLoggerFactory.getLogger("RpcRemoting");

    private String              address;

    public RpcInvokeCallbackListener() {

    }

    public RpcInvokeCallbackListener(String address) {
        this.address = address;
    }

    /**
     * @see com.alipay.remoting.InvokeCallbackListener#onResponse(com.alipay.remoting.InvokeFuture)
     */
    @Override
    public void onResponse(InvokeFuture future) {
        InvokeCallback callback = future.getInvokeCallback();
        if (callback != null) {
            CallbackTask task = new CallbackTask(this.getRemoteAddress(), future);
            if (callback.getExecutor() != null) {
                // There is no need to switch classloader, because executor is provided by user.
                try {
                    callback.getExecutor().execute(task);
                } catch (RejectedExecutionException e) {
                    if (callback instanceof RejectionProcessableInvokeCallback) {
                        switch (((RejectionProcessableInvokeCallback) callback)
                            .rejectedExecutionPolicy()) {
                            case CALLER_RUNS:
                                task.run();
                                break;
                            case CALLER_HANDLE_EXCEPTION:
                                callback.onException(e);
                                break;
                            case DISCARD:
                            default:
                                logger.warn("Callback thread pool busy. discard the callback");
                                break;
                        }
                    } else {
                        logger.warn("Callback thread pool busy.");
                    }
                }
            } else {
                task.run();
            }
        }
    }

    class CallbackTask implements Runnable {

        InvokeFuture future;
        String       remoteAddress;

        /**
         *
         */
        public CallbackTask(String remoteAddress, InvokeFuture future) {
            this.remoteAddress = remoteAddress;
            this.future = future;
        }

        /**
         * @see java.lang.Runnable#run()
         */
        @Override
        public void run() {
            InvokeCallback callback = future.getInvokeCallback();
            // a lot of try-catches to protect thread pool
            ResponseCommand response = null;

            try {
                response = (ResponseCommand) future.waitResponse(0);
            } catch (InterruptedException e) {
                String msg = "Exception caught when getting response from InvokeFuture. The address is "
                             + this.remoteAddress;
                logger.error(msg, e);
            }
            if (response == null || response.getResponseStatus() != ResponseStatus.SUCCESS) {
                try {
                    Exception e;
                    if (response == null) {
                        e = new InvokeException("Exception caught in invocation. The address is "
                                                + this.remoteAddress + " responseStatus:"
                                                + ResponseStatus.UNKNOWN, future.getCause());
                    } else {
                        response.setInvokeContext(future.getInvokeContext());
                        switch (response.getResponseStatus()) {
                            case TIMEOUT:
                                e = new InvokeTimeoutException(
                                    "Invoke timeout when invoke with callback.The address is "
                                            + this.remoteAddress);
                                break;
                            case CONNECTION_CLOSED:
                                e = new ConnectionClosedException(
                                    "Connection closed when invoke with callback.The address is "
                                            + this.remoteAddress);
                                break;
                            case SERVER_THREADPOOL_BUSY:
                                e = new InvokeServerBusyException(
                                    "Server thread pool busy when invoke with callback.The address is "
                                            + this.remoteAddress);
                                break;
                            case SERVER_EXCEPTION:
                                String msg = "Server exception when invoke with callback.Please check the server log! The address is "
                                             + this.remoteAddress;
                                RpcResponseCommand resp = (RpcResponseCommand) response;
                                resp.deserialize();
                                Object ex = resp.getResponseObject();
                                if (ex instanceof Throwable) {
                                    e = new InvokeServerException(msg, (Throwable) ex);
                                } else {
                                    e = new InvokeServerException(msg);
                                }
                                break;
                            default:
                                e = new InvokeException(
                                    "Exception caught in invocation. The address is "
                                            + this.remoteAddress + " responseStatus:"
                                            + response.getResponseStatus(), future.getCause());

                        }
                    }
                    callback.onException(e);
                } catch (Throwable e) {
                    logger
                        .error(
                            "Exception occurred in user defined InvokeCallback#onException() logic, The address is {}",
                            this.remoteAddress, e);
                }
            } else {
                ClassLoader oldClassLoader = null;
                try {
                    if (future.getAppClassLoader() != null) {
                        oldClassLoader = Thread.currentThread().getContextClassLoader();
                        Thread.currentThread().setContextClassLoader(future.getAppClassLoader());
                    }
                    response.setInvokeContext(future.getInvokeContext());
                    RpcResponseCommand rpcResponse = (RpcResponseCommand) response;
                    response.deserialize();
                    try {
                        callback.onResponse(rpcResponse.getResponseObject());
                    } catch (Throwable e) {
                        logger
                            .error(
                                "Exception occurred in user defined InvokeCallback#onResponse() logic.",
                                e);
                    }
                } catch (CodecException e) {
                    logger
                        .error(
                            "CodecException caught on when deserialize response in RpcInvokeCallbackListener. The address is {}.",
                            this.remoteAddress, e);
                } catch (Throwable e) {
                    logger.error(
                        "Exception caught in RpcInvokeCallbackListener. The address is {}",
                        this.remoteAddress, e);
                } finally {
                    if (oldClassLoader != null) {
                        Thread.currentThread().setContextClassLoader(oldClassLoader);
                    }
                }
            } // enf of else
        } // end of run
    }

    /**
     * @see com.alipay.remoting.InvokeCallbackListener#getRemoteAddress()
     */
    @Override
    public String getRemoteAddress() {
        return this.address;
    }
}
