This is an automated email from the ASF dual-hosted git repository.
zhonghongsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 9b7891f79ef Remove CDCConnectionStatus (#25638)
9b7891f79ef is described below
commit 9b7891f79efcbd2f020d102b8b6f46bd2c4bc294
Author: Xinze Guo <[email protected]>
AuthorDate: Sat May 13 14:23:46 2023 +0800
Remove CDCConnectionStatus (#25638)
---
.../pipeline/cdc/constant/CDCConnectionStatus.java | 26 ----------------------
.../pipeline/cdc/context/CDCConnectionContext.java | 7 +++---
.../frontend/netty/CDCChannelInboundHandler.java | 18 +++++----------
3 files changed, 9 insertions(+), 42 deletions(-)
diff --git
a/kernel/data-pipeline/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/constant/CDCConnectionStatus.java
b/kernel/data-pipeline/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/constant/CDCConnectionStatus.java
deleted file mode 100644
index dfcbc49bc33..00000000000
---
a/kernel/data-pipeline/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/constant/CDCConnectionStatus.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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 org.apache.shardingsphere.data.pipeline.cdc.constant;
-
-/**
- * CDC connection status.
- */
-public enum CDCConnectionStatus {
-
- NOT_LOGGED_IN, LOGGED_IN
-}
diff --git
a/kernel/data-pipeline/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/context/CDCConnectionContext.java
b/kernel/data-pipeline/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/context/CDCConnectionContext.java
index 51b689790bf..49c796ffb78 100644
---
a/kernel/data-pipeline/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/context/CDCConnectionContext.java
+++
b/kernel/data-pipeline/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/context/CDCConnectionContext.java
@@ -18,8 +18,8 @@
package org.apache.shardingsphere.data.pipeline.cdc.context;
import lombok.Getter;
+import lombok.RequiredArgsConstructor;
import lombok.Setter;
-import
org.apache.shardingsphere.data.pipeline.cdc.constant.CDCConnectionStatus;
import org.apache.shardingsphere.infra.metadata.user.ShardingSphereUser;
/**
@@ -27,13 +27,12 @@ import
org.apache.shardingsphere.infra.metadata.user.ShardingSphereUser;
*/
@Getter
@Setter
+@RequiredArgsConstructor
public final class CDCConnectionContext {
- private volatile CDCConnectionStatus status;
+ private final ShardingSphereUser currentUser;
private volatile String database;
private volatile String jobId;
-
- private volatile ShardingSphereUser currentUser;
}
diff --git
a/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandler.java
b/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandler.java
index fa97f2059ac..1cb278db782 100644
---
a/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandler.java
+++
b/proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/CDCChannelInboundHandler.java
@@ -26,7 +26,6 @@ import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.authority.model.ShardingSpherePrivileges;
import org.apache.shardingsphere.authority.rule.AuthorityRule;
-import
org.apache.shardingsphere.data.pipeline.cdc.constant.CDCConnectionStatus;
import
org.apache.shardingsphere.data.pipeline.cdc.context.CDCConnectionContext;
import
org.apache.shardingsphere.data.pipeline.cdc.exception.CDCExceptionWrapper;
import org.apache.shardingsphere.data.pipeline.cdc.exception.CDCLoginException;
@@ -72,9 +71,6 @@ public final class CDCChannelInboundHandler extends
ChannelInboundHandlerAdapter
@Override
public void channelActive(final ChannelHandlerContext ctx) {
- CDCConnectionContext context = new CDCConnectionContext();
- context.setStatus(CDCConnectionStatus.NOT_LOGGED_IN);
- ctx.channel().attr(CONNECTION_CONTEXT_KEY).setIfAbsent(context);
CDCResponse response =
CDCResponse.newBuilder().setServerGreetingResult(ServerGreetingResult.newBuilder().setServerVersion(ShardingSphereVersion.VERSION).setProtocolVersion("1")
.build()).build();
ctx.writeAndFlush(response);
@@ -93,7 +89,6 @@ public final class CDCChannelInboundHandler extends
ChannelInboundHandlerAdapter
public void exceptionCaught(final ChannelHandlerContext ctx, final
Throwable cause) {
log.error("caught CDC resolution error", cause);
// TODO add CDC exception to wrapper this exception, and add the
parameters requestId and whether to close connect
- CDCConnectionContext connectionContext =
ctx.channel().attr(CONNECTION_CONTEXT_KEY).get();
ChannelFuture channelFuture;
if (cause instanceof CDCExceptionWrapper) {
CDCExceptionWrapper wrapper = (CDCExceptionWrapper) cause;
@@ -102,7 +97,8 @@ public final class CDCChannelInboundHandler extends
ChannelInboundHandlerAdapter
} else {
channelFuture = ctx.writeAndFlush(CDCResponseGenerator.failed("",
XOpenSQLState.GENERAL_ERROR.getValue(), String.valueOf(cause.getMessage())));
}
- if (CDCConnectionStatus.NOT_LOGGED_IN ==
connectionContext.getStatus()) {
+ CDCConnectionContext connectionContext =
ctx.channel().attr(CONNECTION_CONTEXT_KEY).get();
+ if (null == connectionContext) {
channelFuture.addListener(ChannelFutureListener.CLOSE);
}
}
@@ -110,10 +106,9 @@ public final class CDCChannelInboundHandler extends
ChannelInboundHandlerAdapter
@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg)
{
CDCConnectionContext connectionContext =
ctx.channel().attr(CONNECTION_CONTEXT_KEY).get();
- CDCConnectionStatus status = connectionContext.getStatus();
CDCRequest request = (CDCRequest) msg;
- if (CDCConnectionStatus.NOT_LOGGED_IN == status) {
- processLogin(ctx, request, connectionContext);
+ if (null == connectionContext) {
+ processLogin(ctx, request);
return;
}
switch (request.getType()) {
@@ -141,7 +136,7 @@ public final class CDCChannelInboundHandler extends
ChannelInboundHandlerAdapter
}
}
- private void processLogin(final ChannelHandlerContext ctx, final
CDCRequest request, final CDCConnectionContext connectionContext) {
+ private void processLogin(final ChannelHandlerContext ctx, final
CDCRequest request) {
if (!request.hasLoginRequestBody() ||
!request.getLoginRequestBody().hasBasicBody()) {
throw new CDCExceptionWrapper(request.getRequestId(), new
PipelineInvalidParameterException("Login request body is empty"));
}
@@ -149,8 +144,7 @@ public final class CDCChannelInboundHandler extends
ChannelInboundHandlerAdapter
AuthorityRule authorityRule =
ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData().getGlobalRuleMetaData().getSingleRule(AuthorityRule.class);
Optional<ShardingSphereUser> user = authorityRule.findUser(new
Grantee(body.getUsername(), getHostAddress(ctx)));
if (user.isPresent() &&
Objects.equals(Hashing.sha256().hashBytes(user.get().getPassword().getBytes()).toString().toUpperCase(),
body.getPassword())) {
- connectionContext.setStatus(CDCConnectionStatus.LOGGED_IN);
- connectionContext.setCurrentUser(user.get());
+ ctx.channel().attr(CONNECTION_CONTEXT_KEY).set(new
CDCConnectionContext(user.get()));
ctx.writeAndFlush(CDCResponseGenerator.succeedBuilder(request.getRequestId()).build());
} else {
throw new CDCExceptionWrapper(request.getRequestId(), new
CDCLoginException("Illegal username or password"));
