Created
October 21, 2021 07:12
-
-
Save junwen12221/93e61a856a3e4e22a060b75489c06523 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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(); | |
} | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment