SQLCallStatementHandler.java

package io.mycat.sqlhandler.ddl;

import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLName;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.druid.sql.ast.statement.SQLCallStatement;
import io.mycat.*;
import io.mycat.api.collector.RowBaseIterator;
import io.mycat.beans.mycat.MycatRowMetaData;
import io.mycat.beans.mycat.ResultSetBuilder;
import io.mycat.config.NormalProcedureConfig;
import io.mycat.sqlhandler.AbstractSQLHandler;
import io.mycat.sqlhandler.SQLRequest;
import io.mycat.sqlhandler.procedure.NormalProcedureInfo;
import io.mycat.swapbuffer.PacketRequest;
import io.mycat.swapbuffer.PacketResponse;
import io.mycat.util.StringUtil;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Emitter;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.vertx.core.Future;

import java.util.ArrayList;
import java.util.List;

public class SQLCallStatementHandler extends AbstractSQLHandler<SQLCallStatement> {
    @Override
    protected Future<Void> onExecute(SQLRequest<SQLCallStatement> request, MycatDataContext dataContext, Response response) {
        SQLCallStatement ast = request.getAst();
        SQLName procedureExpr = ast.getProcedureName();
        if (procedureExpr instanceof SQLIdentifierExpr && dataContext.getDefaultSchema() != null) {
            String name = ((SQLIdentifierExpr) procedureExpr).getName();
            procedureExpr = new SQLPropertyExpr(new SQLIdentifierExpr(dataContext.getDefaultSchema()), name);
        }
        SQLPropertyExpr sqlPropertyExpr = (SQLPropertyExpr) procedureExpr;
        String schemaName = SQLUtils.normalize(sqlPropertyExpr.getOwnerName(), true);
        String pName = SQLUtils.normalize(sqlPropertyExpr.getName(), true);

        MetadataManager metadataManager = MetaClusterCurrent.wrapper(MetadataManager.class);
        NormalProcedureInfo procedureConfig = getProcedure(schemaName, pName);
        List<RowBaseIterator> rowBaseIterators = executeProcedure(procedureConfig);

        ProcedureWriter procedureWriter = new ProcedureWriter() {

            byte packetId = 1;
            List<Object> resultsetList = new ArrayList<>();
            boolean onComplete = false;
            String errorMessage;

            @Override
            public void writeOK(int affectRows, int lastInsertId) {
                resultsetList.add(new int[]{affectRows, lastInsertId});
            }

            @Override
            public void writeResult(RowBaseIterator rowBaseIterator) {
                resultsetList.add(rowBaseIterator);
            }

            @Override
            public void onComplete() {
                this.onComplete = true;

            }

            @Override
            public void onError(String message) {
                this.onComplete = true;
                this.errorMessage = message;
            }

            @Override
            public List<byte[]> build() {
                for (Object o : resultsetList) {
                    if ( o instanceof int[]){

                    }else if ( o instanceof RowBaseIterator){

                    }
                }

                return null;
            }
        };
        return response.swapBuffer(Observable.create(new ObservableOnSubscribe<PacketRequest>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<PacketRequest> emitter) throws Throwable {
                procedure(procedureConfig, procedureWriter);
            }
        }), new Emitter<PacketResponse>() {
            @Override
            public void onNext(@NonNull PacketResponse value) {
                value.close();
            }

            @Override
            public void onError(@NonNull Throwable error) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void procedure(NormalProcedureInfo procedureConfig, ProcedureWriter procedureWriter) {

    }

    private List<RowBaseIterator> executeProcedure(NormalProcedureInfo procedureConfig) {
        return null;
    }

    public NormalProcedureInfo getProcedure(String schemaName, String pName) {
        return null;
    }

    interface ProcedureWriter {
        void writeOK(int affectRows, int lastInsertId);

        void writeResult(RowBaseIterator rowBaseIterator);

        void onComplete();

        void onError(String message);

        List<byte[]> build();
    }

    public static void main(String[] args) {
        String s = " create procedure in_param(in p_in int)\n" +
                "   begin\n" +
                "   　　select p_in;\n" +
                "   　　set p_in=2;\n" +
                "      select P_in;\n" +
                "    end";
        SQLStatement sqlStatement = SQLUtils.parseSingleMysqlStatement(s);
        System.out.println();
    }
}