public interface Processor<T> {    
    ProcessorType getType();
    
    int getOrder();
    
    boolean shouldProcess(RequestContext context);
    
    //對外統一封裝為Maybe    
    Maybe<T> process(RequestContext context) throws Exception; 
}

public abstract class AbstractProcessor implements Processor { 
    //同步&無響應，繼承此方法 
    //場景：常規業務處理 
    protected void processSync(RequestContext context) throws Exception {}


    //同步&有響應，繼承此方法，健康檢測
    //場景：健康檢測、未通過校驗時的靜態響應
    protected T processSyncAndGetReponse(RequestContext context) throws Exception {
        process(context);
        return null;
    };


    //異步，繼承此方法
    //場景：認證、鑒權等涉及遠程調用的模塊
    protected Maybe<T> processAsync(RequestContext context) throws Exception 
    {
        T response = processSyncAndGetReponse(context);
        if (response == null) {
            return Maybe.empty();
        } else {
            return Maybe.just(response);
        }
    };


    @Override
    public Maybe<T> process(RequestContext context) throws Exception {
        Maybe<T> maybe = processAsync(context);
        if (maybe instanceof ScalarCallable) {
            //標識同步方法，無需額外封裝
            return maybe;
        } else {
            //統一加超時，默認忽略錯誤
            return maybe.timeout(getAsyncTimeout(context), TimeUnit.MILLISECONDS,
                                 Schedulers.from(context.getEventloop()), timeoutFallback(context));
        }
    }


    protected long getAsyncTimeout(RequestContext context) {
        return 2000;
    }


    protected Maybe<T> timeoutFallback(RequestContext context) {
        return Maybe.empty();
    }
}

public class RxUtil{
    //組合某階段（如Inbound）內的多個filter（即Callable<Maybe<T>>）
    public static <T> Maybe<T> concat(Iterable<? extends Callable<Maybe<T>>> iterable) {
        Iterator<? extends Callable<Maybe<T>>> sources = iterable.iterator();
        while (sources.hasNext()) {
            Maybe<T> maybe;
            try {
                maybe = sources.next().call();
            } catch (Exception e) {
                return Maybe.error(e);
            }
            if (maybe != null) {
                if (maybe instanceof ScalarCallable) {
                    //同步方法
                    T response = ((ScalarCallable<T>)maybe).call();
                    if (response != null) {
                        //有response，中斷
                        return maybe;
                    }
                } else {
                    //異步方法
                    if (sources.hasNext()) {
                        //將sources傳入回調，后續filter重復此邏輯
                        return new ConcattedMaybe(maybe, sources);
                    } else {
                        return maybe;
                    }
                }
            }
        }
        return Maybe.empty();
    }
}

public class ProcessEngine{
    //各個階段，增加默認超時與錯誤處理
    private void process(RequestContext context) {
        List<Callable<Maybe<Response>>> inboundTask = get(ProcessorType.INBOUND, context);
        List<Callable<Maybe<Void>>> outboundTask = get(ProcessorType.OUTBOUND, context);
        List<Callable<Maybe<Response>>> errorTask = get(ProcessorType.ERROR, context);
        List<Callable<Maybe<Void>>> logTask = get(ProcessorType.LOG, context);

        RxUtil.concat(inboundTask)    //inbound階段                    
            .toSingle()        //獲取response                          
            .flatMapMaybe(response -> {
                context.setOriginResponse(response);
                return RxUtil.concat(outboundTask);
            })            //進入outbound
            .onErrorResumeNext(e -> {
                context.setThrowable(e);
                return RxUtil.concat(errorTask).flatMap(response -> {
                    context.resetResponse(response);
                    return RxUtil.concat(outboundTask);
                });
            })            //異常則進入error，并重新進入outbound
            .flatMap(response -> RxUtil.concat(logTask))  //日志階段
            .timeout(asyncTimeout.get(), TimeUnit.MILLISECONDS, Schedulers.from(context.getEventloop()),
                     Maybe.error(new ServerException(500, "Async-Timeout-Processing"))
                    )            //全局兜底超時
            .subscribe(        //釋放資源
            unused -> {
                logger.error("this should not happen, " + context);
                context.release();
            },
            e -> {
                logger.error("this should not happen, " + context, e);
                context.release();
            },
            () -> context.release()
        );
    }   
}

{
    //匹配方式
    "type": "uri",

    //HTTP默認采用uri前綴匹配，內部通過樹結構尋址；私有協議（SOTP）通過服務唯一標識定位。
    "value": "/hotel/order",
    "matcherType": "prefix",

    //標簽與屬性
    //用于portal端權限管理、切面邏輯運行（如按核心/非核心）等
    "tags": [
        "owner_admin",
        "org_framework",
        "appId_123456"
    ],
    "properties": {
        "core": "true"
    },

    //endpoint信息
    "routes": [{
        //condition用于二級路由，如按app版本劃分、按query重分配等
        "condition": "true",
        "conditionParam": {},
        "zone": "PRO",

        //具體服務地址，權重用于灰度場景
        "targets": [{
            "url": "http://test.ctrip.com/hotel",
            "weight": 100
        }
                   ]
    }]
}

{
    //模塊名稱，對應網關內部某個具體模塊
    "name": "addResponseHeader",

    //執行階段
    "stage": "PRE_RESPONSE",

    //執行順序
    "ruleOrder": 0,

    //灰度比例
    "grayRatio": 100,

    //執行條件
    "condition": "true",
    "conditionParam": {},

    //執行參數
    //大量${}形式的內置模板，用于獲取運行時數據
    "actionParam": {
        "connection": "keep-alive",
        "x-service-call": "${request.func.remoteCost}",
        "Access-Control-Expose-Headers": "x-service-call",
        "x-gate-root-id": "${func.catRootMessageId}"
    },

    //異常處理方式，可以拋出或忽略
    "exceptionHandle": "return"
}