Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
apache
Shardingsphere
提交
ecd0a9f7
Shardingsphere
项目概览
apache
/
Shardingsphere
通知
56
Star
3
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
Shardingsphere
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
ecd0a9f7
编写于
4月 21, 2018
作者:
ShardingSphere
提交者:
GitHub
4月 21, 2018
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #748 from tuohai666/dev
for #675 COM_STMT_EXECUTE procedure
上级
40c961f7
a9d611d8
变更
13
隐藏空白更改
内联
并排
Showing
13 changed file
with
247 addition
and
57 deletion
+247
-57
sharding-proxy/src/main/java/io/shardingjdbc/proxy/backend/common/StatementExecuteBackendHandler.java
.../proxy/backend/common/StatementExecuteBackendHandler.java
+102
-46
sharding-proxy/src/main/java/io/shardingjdbc/proxy/backend/mysql/MySQLPacketStatementExecuteQueryResult.java
...backend/mysql/MySQLPacketStatementExecuteQueryResult.java
+20
-8
sharding-proxy/src/main/java/io/shardingjdbc/proxy/frontend/mysql/MySQLFrontendHandler.java
...ardingjdbc/proxy/frontend/mysql/MySQLFrontendHandler.java
+4
-2
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/MySQLPacketPayload.java
...jdbc/proxy/transport/mysql/packet/MySQLPacketPayload.java
+1
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/CommandPacket.java
...c/proxy/transport/mysql/packet/command/CommandPacket.java
+15
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/UnsupportedCommandPacket.java
...nsport/mysql/packet/command/UnsupportedCommandPacket.java
+11
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/statement/close/ComStmtClosePacket.java
...ql/packet/command/statement/close/ComStmtClosePacket.java
+11
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/statement/execute/ComStmtExecutePacket.java
...acket/command/statement/execute/ComStmtExecutePacket.java
+28
-1
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/statement/prepare/ComStmtPreparePacket.java
...acket/command/statement/prepare/ComStmtPreparePacket.java
+11
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/fieldlist/ComFieldListPacket.java
...sql/packet/command/text/fieldlist/ComFieldListPacket.java
+11
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/initdb/ComInitDbPacket.java
...ort/mysql/packet/command/text/initdb/ComInitDbPacket.java
+11
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/query/ComQueryPacket.java
...sport/mysql/packet/command/text/query/ComQueryPacket.java
+11
-0
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/quit/ComQuitPacket.java
...ansport/mysql/packet/command/text/quit/ComQuitPacket.java
+11
-0
未找到文件。
sharding-proxy/src/main/java/io/shardingjdbc/proxy/backend/common/StatementExecuteBackendHandler.java
浏览文件 @
ecd0a9f7
...
...
@@ -65,9 +65,27 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
private
final
PreparedStatementRoutingEngine
routingEngine
;
private
List
<
Connection
>
connections
;
private
List
<
ResultSet
>
resultSets
;
private
MergedResult
mergedResult
;
private
int
currentSequenceId
;
private
int
columnCount
;
private
final
List
<
ColumnType
>
columnTypes
;
private
boolean
hasMoreResultValueFlag
;
public
StatementExecuteBackendHandler
(
final
List
<
PreparedStatementParameter
>
preparedStatementParameters
,
final
int
statementId
,
final
DatabaseType
databaseType
,
final
boolean
showSQL
)
{
this
.
preparedStatementParameters
=
preparedStatementParameters
;
routingEngine
=
new
PreparedStatementRoutingEngine
(
PreparedStatementRegistry
.
getInstance
().
getSQL
(
statementId
),
ShardingRuleRegistry
.
getInstance
().
getShardingRule
(),
databaseType
,
showSQL
);
connections
=
new
ArrayList
<>(
1024
);
resultSets
=
new
ArrayList
<>(
1024
);
columnTypes
=
new
ArrayList
<>(
32
);
hasMoreResultValueFlag
=
true
;
}
@Override
...
...
@@ -77,24 +95,23 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
if
(
routeResult
.
getExecutionUnits
().
isEmpty
())
{
return
new
CommandResponsePackets
(
new
OKPacket
(
1
,
0
,
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
(),
0
,
""
));
}
List
<
ColumnType
>
columnTypes
=
new
ArrayList
<>(
32
);
List
<
CommandResponsePackets
>
result
=
new
LinkedList
<>();
for
(
SQLExecutionUnit
each
:
routeResult
.
getExecutionUnits
())
{
// TODO multiple threads
result
.
add
(
execute
(
routeResult
.
getSqlStatement
(),
each
,
columnTypes
));
result
.
add
(
execute
(
routeResult
.
getSqlStatement
(),
each
));
}
return
merge
(
routeResult
.
getSqlStatement
(),
result
,
columnTypes
);
return
merge
(
routeResult
.
getSqlStatement
(),
result
);
}
private
CommandResponsePackets
execute
(
final
SQLStatement
sqlStatement
,
final
SQLExecutionUnit
sqlExecutionUnit
,
final
List
<
ColumnType
>
columnTypes
)
{
private
CommandResponsePackets
execute
(
final
SQLStatement
sqlStatement
,
final
SQLExecutionUnit
sqlExecutionUnit
)
{
switch
(
sqlStatement
.
getType
())
{
case
DQL:
return
executeQuery
(
ShardingRuleRegistry
.
getInstance
().
getDataSourceMap
().
get
(
sqlExecutionUnit
.
getDataSource
()),
sqlExecutionUnit
.
getSql
()
,
columnTypes
);
return
executeQuery
(
ShardingRuleRegistry
.
getInstance
().
getDataSourceMap
().
get
(
sqlExecutionUnit
.
getDataSource
()),
sqlExecutionUnit
.
getSql
());
case
DML:
case
DDL:
return
executeUpdate
(
ShardingRuleRegistry
.
getInstance
().
getDataSourceMap
().
get
(
sqlExecutionUnit
.
getDataSource
()),
sqlExecutionUnit
.
getSql
(),
sqlStatement
);
default
:
return
executeCommon
(
ShardingRuleRegistry
.
getInstance
().
getDataSourceMap
().
get
(
sqlExecutionUnit
.
getDataSource
()),
sqlExecutionUnit
.
getSql
()
,
columnTypes
);
return
executeCommon
(
ShardingRuleRegistry
.
getInstance
().
getDataSourceMap
().
get
(
sqlExecutionUnit
.
getDataSource
()),
sqlExecutionUnit
.
getSql
());
}
}
...
...
@@ -112,13 +129,16 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
}
}
private
CommandResponsePackets
executeQuery
(
final
DataSource
dataSource
,
final
String
sql
,
final
List
<
ColumnType
>
columnTypes
)
{
try
(
Connection
connection
=
dataSource
.
getConnection
();
PreparedStatement
preparedStatement
=
connection
.
prepareStatement
(
sql
))
{
private
CommandResponsePackets
executeQuery
(
final
DataSource
dataSource
,
final
String
sql
)
{
PreparedStatement
preparedStatement
;
try
{
Connection
connection
=
dataSource
.
getConnection
();
connections
.
add
(
connection
);
preparedStatement
=
connection
.
prepareStatement
(
sql
);
preparedStatement
.
setFetchSize
(
Integer
.
MIN_VALUE
);
setJDBCPreparedStatementParameters
(
preparedStatement
);
ResultSet
resultSet
=
preparedStatement
.
executeQuery
(
);
return
getDatabaseProtocolPackets
(
resultSet
,
columnTypes
);
resultSets
.
add
(
preparedStatement
.
executeQuery
()
);
return
getDatabaseProtocolPackets
();
}
catch
(
final
SQLException
ex
)
{
return
new
CommandResponsePackets
(
new
ErrPacket
(
1
,
ex
.
getErrorCode
(),
""
,
ex
.
getSQLState
(),
ex
.
getMessage
()));
}
...
...
@@ -144,24 +164,24 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
}
catch
(
final
SQLException
ex
)
{
return
new
CommandResponsePackets
(
new
ErrPacket
(
1
,
ex
.
getErrorCode
(),
""
,
ex
.
getSQLState
(),
ex
.
getMessage
()));
}
finally
{
if
(
preparedStatement
!=
null
)
{
if
(
null
!=
preparedStatement
)
{
try
{
preparedStatement
.
close
();
}
catch
(
final
SQLException
ignore
)
{
}
}
}
}
private
CommandResponsePackets
executeCommon
(
final
DataSource
dataSource
,
final
String
sql
,
final
List
<
ColumnType
>
columnTypes
)
{
private
CommandResponsePackets
executeCommon
(
final
DataSource
dataSource
,
final
String
sql
)
{
try
(
Connection
connection
=
dataSource
.
getConnection
();
PreparedStatement
preparedStatement
=
connection
.
prepareStatement
(
sql
))
{
setJDBCPreparedStatementParameters
(
preparedStatement
);
boolean
hasResultSet
=
preparedStatement
.
execute
();
if
(
hasResultSet
)
{
return
getDatabaseProtocolPackets
(
preparedStatement
.
getResultSet
(),
columnTypes
);
resultSets
.
add
(
preparedStatement
.
getResultSet
());
return
getDatabaseProtocolPackets
();
}
else
{
return
new
CommandResponsePackets
(
new
OKPacket
(
1
,
preparedStatement
.
getUpdateCount
(),
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
(),
0
,
""
));
}
...
...
@@ -170,11 +190,11 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
}
}
private
CommandResponsePackets
getDatabaseProtocolPackets
(
final
ResultSet
resultSet
,
final
List
<
ColumnType
>
columnTypes
)
throws
SQLException
{
private
CommandResponsePackets
getDatabaseProtocolPackets
()
throws
SQLException
{
CommandResponsePackets
result
=
new
CommandResponsePackets
();
int
currentSequenceId
=
0
;
ResultSetMetaData
resultSetMetaData
=
resultSet
.
getMetaData
();
int
columnCount
=
resultSetMetaData
.
getColumnCount
();
ResultSetMetaData
resultSetMetaData
=
resultSet
s
.
get
(
resultSets
.
size
()
-
1
)
.
getMetaData
();
columnCount
=
resultSetMetaData
.
getColumnCount
();
if
(
0
==
columnCount
)
{
result
.
addPacket
(
new
OKPacket
(++
currentSequenceId
,
0
,
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
(),
0
,
""
));
return
result
;
...
...
@@ -188,14 +208,6 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
columnTypes
.
add
(
columnType
);
}
result
.
addPacket
(
new
EofPacket
(++
currentSequenceId
,
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
()));
while
(
resultSet
.
next
())
{
List
<
Object
>
data
=
new
ArrayList
<>(
columnCount
);
for
(
int
i
=
1
;
i
<=
columnCount
;
i
++)
{
data
.
add
(
resultSet
.
getObject
(
i
));
}
result
.
addPacket
(
new
BinaryResultSetRowPacket
(++
currentSequenceId
,
columnCount
,
data
,
columnTypes
));
}
result
.
addPacket
(
new
EofPacket
(++
currentSequenceId
,
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
()));
return
result
;
}
...
...
@@ -208,7 +220,7 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
return
result
;
}
private
CommandResponsePackets
merge
(
final
SQLStatement
sqlStatement
,
final
List
<
CommandResponsePackets
>
packets
,
final
List
<
ColumnType
>
columnTypes
)
{
private
CommandResponsePackets
merge
(
final
SQLStatement
sqlStatement
,
final
List
<
CommandResponsePackets
>
packets
)
{
if
(
1
==
packets
.
size
())
{
return
packets
.
iterator
().
next
();
}
...
...
@@ -225,7 +237,7 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
return
mergeDML
(
headPackets
);
}
if
(
SQLType
.
DQL
==
sqlStatement
.
getType
()
||
SQLType
.
DAL
==
sqlStatement
.
getType
())
{
return
mergeDQLorDAL
(
sqlStatement
,
packets
,
columnTypes
);
return
mergeDQLorDAL
(
sqlStatement
,
packets
);
}
return
packets
.
get
(
0
);
}
...
...
@@ -241,44 +253,88 @@ public final class StatementExecuteBackendHandler implements BackendHandler {
return
new
CommandResponsePackets
(
new
OKPacket
(
1
,
affectedRows
,
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
(),
0
,
""
));
}
private
CommandResponsePackets
mergeDQLorDAL
(
final
SQLStatement
sqlStatement
,
final
List
<
CommandResponsePackets
>
packets
,
final
List
<
ColumnType
>
columnTypes
)
{
private
CommandResponsePackets
mergeDQLorDAL
(
final
SQLStatement
sqlStatement
,
final
List
<
CommandResponsePackets
>
packets
)
{
List
<
QueryResult
>
queryResults
=
new
ArrayList
<>(
packets
.
size
());
for
(
CommandResponsePackets
each
:
packets
)
{
for
(
int
i
=
0
;
i
<
packets
.
size
();
i
++
)
{
// TODO replace to a common PacketQueryResult
queryResults
.
add
(
new
MySQLPacketStatementExecuteQueryResult
(
each
));
queryResults
.
add
(
new
MySQLPacketStatementExecuteQueryResult
(
packets
.
get
(
i
),
resultSets
.
get
(
i
),
columnTypes
));
}
MergedResult
mergedResult
;
try
{
mergedResult
=
MergeEngineFactory
.
newInstance
(
ShardingRuleRegistry
.
getInstance
().
getShardingRule
(),
queryResults
,
sqlStatement
).
merge
();
}
catch
(
final
SQLException
ex
)
{
return
new
CommandResponsePackets
(
new
ErrPacket
(
1
,
ex
.
getErrorCode
(),
""
,
ex
.
getSQLState
(),
ex
.
getMessage
()));
}
return
buildPackets
(
packets
,
mergedResult
,
columnTypes
);
return
buildPackets
(
packets
);
}
private
CommandResponsePackets
buildPackets
(
final
List
<
CommandResponsePackets
>
packets
,
final
MergedResult
mergedResult
,
final
List
<
ColumnType
>
columnTypes
)
{
private
CommandResponsePackets
buildPackets
(
final
List
<
CommandResponsePackets
>
packets
)
{
CommandResponsePackets
result
=
new
CommandResponsePackets
();
Iterator
<
DatabaseProtocolPacket
>
databaseProtocolPacketsSampling
=
packets
.
iterator
().
next
().
getDatabaseProtocolPackets
().
iterator
();
FieldCountPacket
fieldCountPacketSampling
=
(
FieldCountPacket
)
databaseProtocolPacketsSampling
.
next
();
result
.
addPacket
(
fieldCountPacketSampling
);
int
columnCount
=
fieldCountPacketSampling
.
getColumnCount
()
;
++
currentSequenceId
;
for
(
int
i
=
0
;
i
<
columnCount
;
i
++)
{
result
.
addPacket
(
databaseProtocolPacketsSampling
.
next
());
++
currentSequenceId
;
}
result
.
addPacket
(
databaseProtocolPacketsSampling
.
next
());
int
currentSequenceId
=
result
.
size
();
++
currentSequenceId
;
return
result
;
}
/**
* Has more Result value.
*
* @return has more result value
* @throws SQLException sql exception
*/
public
boolean
hasMoreResultValue
()
throws
SQLException
{
if
(!
hasMoreResultValueFlag
)
{
return
false
;
}
if
(!
mergedResult
.
next
())
{
hasMoreResultValueFlag
=
false
;
cleanJDBCResources
();
}
return
true
;
}
/**
* Get result value.
*
* @return database protocol packet
*/
public
DatabaseProtocolPacket
getResultValue
()
{
if
(!
hasMoreResultValueFlag
)
{
return
new
EofPacket
(++
currentSequenceId
,
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
());
}
try
{
while
(
mergedResult
.
next
())
{
List
<
Object
>
data
=
new
ArrayList
<>(
columnCount
);
for
(
int
i
=
1
;
i
<=
columnCount
;
i
++)
{
data
.
add
(
mergedResult
.
getValue
(
i
,
Object
.
class
));
}
result
.
addPacket
(
new
BinaryResultSetRowPacket
(++
currentSequenceId
,
columnCount
,
data
,
columnTypes
));
List
<
Object
>
data
=
new
ArrayList
<>(
columnCount
);
for
(
int
i
=
1
;
i
<=
columnCount
;
i
++)
{
data
.
add
(
mergedResult
.
getValue
(
i
,
Object
.
class
));
}
return
new
BinaryResultSetRowPacket
(++
currentSequenceId
,
columnCount
,
data
,
columnTypes
);
}
catch
(
final
SQLException
ex
)
{
return
new
CommandResponsePackets
(
new
ErrPacket
(
1
,
ex
.
getErrorCode
(),
""
,
ex
.
getSQLState
(),
ex
.
getMessage
()));
return
new
ErrPacket
(
1
,
ex
.
getErrorCode
(),
""
,
ex
.
getSQLState
(),
ex
.
getMessage
());
}
}
private
void
cleanJDBCResources
()
{
for
(
ResultSet
each
:
resultSets
)
{
if
(
null
!=
each
)
{
try
{
each
.
close
();
}
catch
(
final
SQLException
ignore
)
{
}
}
}
for
(
Connection
each
:
connections
)
{
if
(
null
!=
each
)
{
try
{
each
.
close
();
}
catch
(
final
SQLException
ignore
)
{
}
}
}
result
.
addPacket
(
new
EofPacket
(++
currentSequenceId
,
0
,
StatusFlag
.
SERVER_STATUS_AUTOCOMMIT
.
getValue
()));
return
result
;
}
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/backend/mysql/MySQLPacketStatementExecuteQueryResult.java
浏览文件 @
ecd0a9f7
...
...
@@ -19,6 +19,7 @@ package io.shardingjdbc.proxy.backend.mysql;
import
io.shardingjdbc.core.merger.QueryResult
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.constant.ColumnType
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandResponsePackets
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.statement.execute.BinaryResultSetRowPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.text.query.ColumnDefinition41Packet
;
...
...
@@ -26,9 +27,13 @@ import io.shardingjdbc.proxy.transport.mysql.packet.command.text.query.FieldCoun
import
lombok.RequiredArgsConstructor
;
import
java.io.InputStream
;
import
java.sql.ResultSet
;
import
java.sql.SQLException
;
import
java.util.ArrayList
;
import
java.util.Calendar
;
import
java.util.HashMap
;
import
java.util.Iterator
;
import
java.util.List
;
import
java.util.Map
;
/**
...
...
@@ -41,15 +46,19 @@ public final class MySQLPacketStatementExecuteQueryResult implements QueryResult
private
final
int
columnCount
;
private
final
List
<
ColumnType
>
columnTypes
;
private
final
Map
<
Integer
,
String
>
columnIndexAndLabelMap
;
private
final
Map
<
String
,
Integer
>
columnLabelAndIndexMap
;
private
final
Iterator
<
DatabaseProtocolPacket
>
data
;
private
final
ResultSet
resultSet
;
private
int
currentSequenceId
;
private
BinaryResultSetRowPacket
currentRow
;
public
MySQLPacketStatementExecuteQueryResult
(
final
CommandResponsePackets
packets
)
{
public
MySQLPacketStatementExecuteQueryResult
(
final
CommandResponsePackets
packets
,
final
ResultSet
resultSet
,
final
List
<
ColumnType
>
columnTypes
)
{
Iterator
<
DatabaseProtocolPacket
>
packetIterator
=
packets
.
getDatabaseProtocolPackets
().
iterator
();
columnCount
=
((
FieldCountPacket
)
packetIterator
.
next
()).
getColumnCount
();
columnIndexAndLabelMap
=
new
HashMap
<>(
columnCount
,
1
);
...
...
@@ -59,15 +68,18 @@ public final class MySQLPacketStatementExecuteQueryResult implements QueryResult
columnIndexAndLabelMap
.
put
(
i
,
columnDefinition41Packet
.
getName
());
columnLabelAndIndexMap
.
put
(
columnDefinition41Packet
.
getName
(),
i
);
}
packetIterator
.
next
()
;
data
=
packetIterator
;
this
.
resultSet
=
resultSet
;
this
.
columnTypes
=
columnTypes
;
}
@Override
public
boolean
next
()
{
DatabaseProtocolPacket
databaseProtocolPacket
=
data
.
next
();
if
(
databaseProtocolPacket
instanceof
BinaryResultSetRowPacket
)
{
currentRow
=
(
BinaryResultSetRowPacket
)
databaseProtocolPacket
;
public
boolean
next
()
throws
SQLException
{
if
(
resultSet
.
next
())
{
List
<
Object
>
data
=
new
ArrayList
<>(
columnCount
);
for
(
int
i
=
1
;
i
<=
columnCount
;
i
++)
{
data
.
add
(
resultSet
.
getObject
(
i
));
}
currentRow
=
new
BinaryResultSetRowPacket
(++
currentSequenceId
,
columnCount
,
data
,
columnTypes
);
return
true
;
}
return
false
;
...
...
sharding-proxy/src/main/java/io/shardingjdbc/proxy/frontend/mysql/MySQLFrontendHandler.java
浏览文件 @
ecd0a9f7
...
...
@@ -69,9 +69,11 @@ public final class MySQLFrontendHandler extends FrontendHandler {
int
sequenceId
=
mysqlPacketPayload
.
readInt1
();
CommandPacket
commandPacket
=
CommandPacketFactory
.
getCommandPacket
(
sequenceId
,
mysqlPacketPayload
);
for
(
DatabaseProtocolPacket
each
:
commandPacket
.
execute
().
getDatabaseProtocolPackets
())
{
context
.
write
(
each
);
context
.
writeAndFlush
(
each
);
}
while
(
commandPacket
.
hasMoreResultValue
())
{
context
.
writeAndFlush
(
commandPacket
.
getResultValue
());
}
context
.
flush
();
}
});
}
...
...
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/MySQLPacketPayload.java
浏览文件 @
ecd0a9f7
...
...
@@ -33,6 +33,7 @@ import java.util.Date;
* @see <a href="https://dev.mysql.com/doc/internals/en/binary-protocol-value.html">binary protocol value</a>
*
* @author zhangliang
* @author zhangyonglun
*/
@RequiredArgsConstructor
@Getter
...
...
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/CommandPacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -17,6 +17,7 @@
package
io.shardingjdbc.proxy.transport.mysql.packet.command
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacket
;
/**
...
...
@@ -36,4 +37,18 @@ public abstract class CommandPacket extends MySQLPacket {
* @return result packets to be sent
*/
public
abstract
CommandResponsePackets
execute
();
/**
* Has more result value.
*
* @return has more result value
*/
public
abstract
boolean
hasMoreResultValue
();
/**
* Get result value.
*
* @return result to be sent
*/
public
abstract
DatabaseProtocolPacket
getResultValue
();
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/UnsupportedCommandPacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -17,6 +17,7 @@
package
io.shardingjdbc.proxy.transport.mysql.packet.command
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
import
io.shardingjdbc.proxy.transport.mysql.packet.generic.ErrPacket
;
...
...
@@ -50,4 +51,14 @@ public final class UnsupportedCommandPacket extends CommandPacket {
@Override
public
void
write
(
final
MySQLPacketPayload
mysqlPacketPayload
)
{
}
@Override
public
boolean
hasMoreResultValue
()
{
return
false
;
}
@Override
public
DatabaseProtocolPacket
getResultValue
()
{
return
null
;
}
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/statement/close/ComStmtClosePacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -17,6 +17,7 @@
package
io.shardingjdbc.proxy.transport.mysql.packet.command.statement.close
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandResponsePackets
;
...
...
@@ -47,4 +48,14 @@ public class ComStmtClosePacket extends CommandPacket {
log
.
debug
(
"COM_STMT_CLOSE received for Sharding-Proxy: {}"
,
statementId
);
return
new
CommandResponsePackets
(
new
DummyPacket
());
}
@Override
public
boolean
hasMoreResultValue
()
{
return
false
;
}
@Override
public
DatabaseProtocolPacket
getResultValue
()
{
return
null
;
}
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/statement/execute/ComStmtExecutePacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -23,6 +23,7 @@ import io.shardingjdbc.core.parsing.SQLParsingEngine;
import
io.shardingjdbc.core.parsing.parser.sql.SQLStatement
;
import
io.shardingjdbc.proxy.backend.common.StatementExecuteBackendHandler
;
import
io.shardingjdbc.proxy.config.ShardingRuleRegistry
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.constant.ColumnType
;
import
io.shardingjdbc.proxy.transport.mysql.constant.NewParametersBoundFlag
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
...
...
@@ -32,6 +33,7 @@ import io.shardingjdbc.proxy.transport.mysql.packet.command.statement.PreparedSt
import
lombok.Getter
;
import
lombok.extern.slf4j.Slf4j
;
import
java.sql.SQLException
;
import
java.util.ArrayList
;
import
java.util.List
;
...
...
@@ -63,6 +65,8 @@ public final class ComStmtExecutePacket extends CommandPacket {
private
final
List
<
PreparedStatementParameter
>
preparedStatementParameters
=
new
ArrayList
<>(
32
);
private
final
StatementExecuteBackendHandler
statementExecuteBackendHandler
;
public
ComStmtExecutePacket
(
final
int
sequenceId
,
final
MySQLPacketPayload
mysqlPacketPayload
)
{
super
(
sequenceId
);
statementId
=
mysqlPacketPayload
.
readInt4
();
...
...
@@ -77,6 +81,7 @@ public final class ComStmtExecutePacket extends CommandPacket {
}
newParametersBoundFlag
=
NewParametersBoundFlag
.
valueOf
(
mysqlPacketPayload
.
readInt1
());
setParameterList
(
mysqlPacketPayload
,
numParameters
,
newParametersBoundFlag
);
statementExecuteBackendHandler
=
new
StatementExecuteBackendHandler
(
preparedStatementParameters
,
statementId
,
DatabaseType
.
MySQL
,
true
);
}
private
void
setParameterList
(
final
MySQLPacketPayload
mysqlPacketPayload
,
final
int
numParameters
,
final
NewParametersBoundFlag
newParametersBoundFlag
)
{
...
...
@@ -144,6 +149,28 @@ public final class ComStmtExecutePacket extends CommandPacket {
@Override
public
CommandResponsePackets
execute
()
{
log
.
debug
(
"COM_STMT_EXECUTE received for Sharding-Proxy: {}"
,
statementId
);
return
new
StatementExecuteBackendHandler
(
preparedStatementParameters
,
statementId
,
DatabaseType
.
MySQL
,
true
).
execute
();
return
statementExecuteBackendHandler
.
execute
();
}
/**
* Has more Result value.
*
* @return has more result value
*/
public
boolean
hasMoreResultValue
()
{
try
{
return
statementExecuteBackendHandler
.
hasMoreResultValue
();
}
catch
(
final
SQLException
ex
)
{
return
false
;
}
}
/**
* Get result value.
*
* @return database protocol packet
*/
public
DatabaseProtocolPacket
getResultValue
()
{
return
statementExecuteBackendHandler
.
getResultValue
();
}
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/statement/prepare/ComStmtPreparePacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -24,6 +24,7 @@ import io.shardingjdbc.core.parsing.parser.sql.SQLStatement;
import
io.shardingjdbc.core.parsing.parser.sql.dml.insert.InsertStatement
;
import
io.shardingjdbc.core.parsing.parser.sql.dql.select.SelectStatement
;
import
io.shardingjdbc.proxy.config.ShardingRuleRegistry
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.constant.ColumnType
;
import
io.shardingjdbc.proxy.transport.mysql.constant.StatusFlag
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
...
...
@@ -75,6 +76,16 @@ public final class ComStmtPreparePacket extends CommandPacket {
return
result
;
}
@Override
public
boolean
hasMoreResultValue
()
{
return
false
;
}
@Override
public
DatabaseProtocolPacket
getResultValue
()
{
return
null
;
}
private
int
getNumColumns
(
final
SQLStatement
sqlStatement
)
{
if
(
sqlStatement
instanceof
SelectStatement
)
{
// TODO select * cannot know items num
...
...
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/fieldlist/ComFieldListPacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -20,6 +20,7 @@ package io.shardingjdbc.proxy.transport.mysql.packet.command.text.fieldlist;
import
io.shardingjdbc.core.constant.DatabaseType
;
import
io.shardingjdbc.core.constant.ShardingConstant
;
import
io.shardingjdbc.proxy.backend.common.SQLExecuteBackendHandler
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandPacketType
;
...
...
@@ -60,4 +61,14 @@ public final class ComFieldListPacket extends CommandPacket {
// TODO use common database type
return
new
SQLExecuteBackendHandler
(
sql
,
DatabaseType
.
MySQL
,
true
).
execute
();
}
@Override
public
boolean
hasMoreResultValue
()
{
return
false
;
}
@Override
public
DatabaseProtocolPacket
getResultValue
()
{
return
null
;
}
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/initdb/ComInitDbPacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -18,6 +18,7 @@
package
io.shardingjdbc.proxy.transport.mysql.packet.command.text.initdb
;
import
io.shardingjdbc.core.constant.ShardingConstant
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.constant.StatusFlag
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandPacket
;
...
...
@@ -57,4 +58,14 @@ public final class ComInitDbPacket extends CommandPacket {
}
return
new
CommandResponsePackets
(
new
ErrPacket
(
getSequenceId
()
+
1
,
1049
,
""
,
""
,
String
.
format
(
"Unknown database '%s'"
,
schemaName
)));
}
@Override
public
boolean
hasMoreResultValue
()
{
return
false
;
}
@Override
public
DatabaseProtocolPacket
getResultValue
()
{
return
null
;
}
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/query/ComQueryPacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -19,6 +19,7 @@ package io.shardingjdbc.proxy.transport.mysql.packet.command.text.query;
import
io.shardingjdbc.core.constant.DatabaseType
;
import
io.shardingjdbc.proxy.backend.common.SQLExecuteBackendHandler
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandPacket
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandResponsePackets
;
...
...
@@ -50,4 +51,14 @@ public final class ComQueryPacket extends CommandPacket {
log
.
debug
(
"COM_QUERY received for Sharding-Proxy: {}"
,
sql
);
return
new
SQLExecuteBackendHandler
(
sql
,
DatabaseType
.
MySQL
,
true
).
execute
();
}
@Override
public
boolean
hasMoreResultValue
()
{
return
false
;
}
@Override
public
DatabaseProtocolPacket
getResultValue
()
{
return
null
;
}
}
sharding-proxy/src/main/java/io/shardingjdbc/proxy/transport/mysql/packet/command/text/quit/ComQuitPacket.java
浏览文件 @
ecd0a9f7
...
...
@@ -17,6 +17,7 @@
package
io.shardingjdbc.proxy.transport.mysql.packet.command.text.quit
;
import
io.shardingjdbc.proxy.transport.common.packet.DatabaseProtocolPacket
;
import
io.shardingjdbc.proxy.transport.mysql.constant.StatusFlag
;
import
io.shardingjdbc.proxy.transport.mysql.packet.MySQLPacketPayload
;
import
io.shardingjdbc.proxy.transport.mysql.packet.command.CommandPacket
;
...
...
@@ -45,4 +46,14 @@ public final class ComQuitPacket extends CommandPacket {
public
void
write
(
final
MySQLPacketPayload
mysqlPacketPayload
)
{
mysqlPacketPayload
.
writeInt1
(
CommandPacketType
.
COM_QUIT
.
getValue
());
}
@Override
public
boolean
hasMoreResultValue
()
{
return
false
;
}
@Override
public
DatabaseProtocolPacket
getResultValue
()
{
return
null
;
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录