Skip to content

O
oblogclient

oceanbase / oblogclient
大约 2 年 前同步成功

通知 5
Star 15
Fork 11
O
oblogclient
未验证 提交 1686e1dc 编写于 作者: W whhe 提交者： GitHub
浏览文件
操作

add comments and cleanup redundant class (#6) 

* add comment and cleanup redundant class

* update java doc

* add javadoc to listeners

* cleanup exception class
上级 8cae74a2
隐藏空白更改
内联 并排
Showing with 1305 addition and 1628 deletion
common/src/main/java/com/oceanbase/clogproxy/common/config/ShareConf.java → common/src/main/java/com/oceanbase/clogproxy/common/config/SharedConf.java
浏览文件 @ 1686e1dc
...@@ -10,6 +10,13 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,6 +10,13 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.config; package com.oceanbase.clogproxy.common.config;
public class ShareConf { /**
* The class that defines the shared constants.
*/
public class SharedConf {
/**
* Flag of whether to use the hash function to process password.
*/
public static boolean AUTH_PASSWORD_HASH = true; public static boolean AUTH_PASSWORD_HASH = true;
} }
common/src/main/java/com/oceanbase/clogproxy/common/packet/CompressType.java
浏览文件 @ 1686e1dc
...@@ -10,23 +10,40 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,23 +10,40 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.packet; package com.oceanbase.clogproxy.common.packet;
/**
* Compress type enumeration. Primarily used for {@link com.oceanbase.clogproxy.common.packet.protocol.LogProxyProto.RecordData}.
*/
public enum CompressType { public enum CompressType {
/** /**
* no compress * No compress.
*/ */
NONE(0), NONE(0),
/** /**
* lz4 compress * LZ4 compress.
*/ */
LZ4(1); LZ4(1);
private int code; /**
* The ordinal of this enumeration constant.
*/
private final int code;
/**
* Constructor.
*
* @param code The ordinal of this enumeration constant.
*/
CompressType(int code) { CompressType(int code) {
this.code = code; this.code = code;
} }
/**
* Returns the enum constant of CompressType with the specified code.
*
* @param code The ordinal of this enumeration constant.
* @return The enum constant.
*/
public static CompressType codeOf(int code) { public static CompressType codeOf(int code) {
for (CompressType v : values()) { for (CompressType v : values()) {
if (v.code == code) { if (v.code == code) {
...@@ -36,6 +53,11 @@ public enum CompressType { ...@@ -36,6 +53,11 @@ public enum CompressType {
return null; return null;
} }
/**
* Get the ordinal of this enumeration constant.
*
* @return The ordinal of this enumeration constant.
*/
public int code() { public int code() {
return code; return code;
} }
......
common/src/main/java/com/oceanbase/clogproxy/common/packet/HeaderType.java
浏览文件 @ 1686e1dc
...@@ -10,59 +10,75 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,59 +10,75 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.packet; package com.oceanbase.clogproxy.common.packet;
/**
* Header type enumeration. Used to identify the type of request or response in protobuf.
*/
public enum HeaderType { public enum HeaderType {
/** /**
* error response * Error response.
*/ */
ERROR_RESPONSE(-1), ERROR_RESPONSE(-1),
/** /**
* client request handshake * Client handshake request.
*/ */
HANDSHAKE_REQUEST_CLIENT(1), HANDSHAKE_REQUEST_CLIENT(1),
/** /**
* response to client handshake * Client handshake response.
*/ */
HANDSHAKE_RESPONSE_CLIENT(2), HANDSHAKE_RESPONSE_CLIENT(2),
/** /**
* logreader request handshake * LogReader handshake request.
*/ */
HANDSHAKE_REQUEST_LOGREADER(3), HANDSHAKE_REQUEST_LOGREADER(3),
/** /**
* logreader response handshake * LogReader handshake response.
*/ */
HANDSHAKE_RESPONSE_LOGREADER(4), HANDSHAKE_RESPONSE_LOGREADER(4),
/** /**
* logreader data stream * LogReader data stream.
*/ */
DATA_LOGREADER(5), DATA_LOGREADER(5),
/** /**
* client data stream * Client data stream.
*/ */
DATA_CLIENT(6), DATA_CLIENT(6),
/** /**
* status info of server runtime * Status info of server runtime.
*/ */
STATUS(7), STATUS(7),
/** /**
* status info of LogReader * Status info of LogReader.
*/ */
STATUS_LOGREADER(8), STATUS_LOGREADER(8);
;
/**
* The ordinal of this enumeration constant.
*/
private final int code; private final int code;
/**
* Constructor.
*
* @param code The ordinal of this enumeration constant.
*/
HeaderType(int code) { HeaderType(int code) {
this.code = code; this.code = code;
} }
/**
* Returns the enum constant of HeaderType with the specified code.
*
* @param code The ordinal of this enumeration constant.
* @return The enum constant.
*/
public static HeaderType codeOf(int code) { public static HeaderType codeOf(int code) {
for (HeaderType t : values()) { for (HeaderType t : values()) {
if (t.code == code) { if (t.code == code) {
...@@ -72,6 +88,11 @@ public enum HeaderType { ...@@ -72,6 +88,11 @@ public enum HeaderType {
return null; return null;
} }
/**
* Get the ordinal of this enumeration constant.
*
* @return The ordinal of this enumeration constant.
*/
public int code() { public int code() {
return code; return code;
} }
......
common/src/main/java/com/oceanbase/clogproxy/common/packet/LogType.java
浏览文件 @ 1686e1dc
...@@ -10,45 +10,51 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,45 +10,51 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.packet; package com.oceanbase.clogproxy.common.packet;
import java.util.HashMap; /**
import java.util.Map; * Log type enumeration.
*/
public enum LogType { public enum LogType {
/** /**
* LogProxy OceanBase LogReader * LogProxy OceanBase LogReader.
*/ */
OCEANBASE(0); OCEANBASE(0);
/**
* The ordinal of this enumeration constant.
*/
private final int code; private final int code;
private static final Map<Integer, LogType> CODE_TYPES = new HashMap<>(values().length); /**
* Constructor.
static { *
for (LogType logCaptureType : values()) { * @param code The ordinal of this enumeration constant.
CODE_TYPES.put(logCaptureType.code, logCaptureType); */
}
}
LogType(int code) { LogType(int code) {
this.code = code; this.code = code;
} }
public int getCode() { /**
return this.code; * Returns the enum constant of LogType with the specified code.
} *
* @param code The ordinal of this enumeration constant.
public static LogType fromString(String string) { * @return The enum constant.
if (string == null) { */
throw new NullPointerException("logTypeString is null"); public static LogType codeOf(int code) {
for (LogType t : values()) {
if (t.code == code) {
return t;
}
} }
return valueOf(string.toUpperCase()); return null;
} }
public static LogType fromCode(int code) { /**
if (CODE_TYPES.containsKey(code)) { * Get the ordinal of this enumeration constant.
return CODE_TYPES.get(code); *
} * @return The ordinal of this enumeration constant.
return null; */
public int code() {
return this.code;
} }
} }
common/src/main/java/com/oceanbase/clogproxy/common/packet/ProtocolVersion.java
浏览文件 @ 1686e1dc
...@@ -10,20 +10,46 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,20 +10,46 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.packet; package com.oceanbase.clogproxy.common.packet;
/**
* Protocol version enumeration.
*/
public enum ProtocolVersion { public enum ProtocolVersion {
/** /**
* v0 version * Protocol version 0.
*/ */
V0(0), V0(0),
/**
* Protocol version 1.
*/
V1(1), V1(1),
/**
* Protocol version 2.
*/
V2(2); V2(2);
/**
* The ordinal of this enumeration constant.
*/
private final int code; private final int code;
/**
* Constructor.
*
* @param code The ordinal of this enumeration constant.
*/
ProtocolVersion(int code) { ProtocolVersion(int code) {
this.code = code; this.code = code;
} }
/**
* Returns the enum constant of ProtocolVersion with the specified code.
*
* @param code The ordinal of this enumeration constant.
* @return The enum constant.
*/
public static ProtocolVersion codeOf(int code) { public static ProtocolVersion codeOf(int code) {
for (ProtocolVersion v : values()) { for (ProtocolVersion v : values()) {
if (v.code == code) { if (v.code == code) {
...@@ -33,6 +59,11 @@ public enum ProtocolVersion { ...@@ -33,6 +59,11 @@ public enum ProtocolVersion {
return null; return null;
} }
/**
* Get the ordinal of this enumeration constant.
*
* @return The ordinal of this enumeration constant.
*/
public int code() { public int code() {
return code; return code;
} }
......
common/src/main/java/com/oceanbase/clogproxy/common/util/CryptoUtil.java
浏览文件 @ 1686e1dc
...@@ -22,26 +22,73 @@ import java.security.InvalidAlgorithmParameterException; ...@@ -22,26 +22,73 @@ import java.security.InvalidAlgorithmParameterException;
import java.security.InvalidKeyException; import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException; import java.security.NoSuchAlgorithmException;
/**
* Utils class for crypto.
*/
public class CryptoUtil { public class CryptoUtil {
/**
* Default cipher key.
*/
private static final String KEY = "LogProxy123*"; private static final String KEY = "LogProxy123*";
/**
* AES key length.
*/
private static final int AES_KEY_SIZE = 256; private static final int AES_KEY_SIZE = 256;
/**
* GCM tag length.
*/
private static final int GCM_TAG_LENGTH = 16; private static final int GCM_TAG_LENGTH = 16;
/**
* Create an {@link Encryptor} instance using given cipher key.
*
* @param key Cipher key.
* @return An {@link Encryptor} instance.
*/
public static Encryptor newEncryptor(String key) { public static Encryptor newEncryptor(String key) {
return new Encryptor(key); return new Encryptor(key);
} }
/**
* Create an Encryptor instance using default cipher key.
*
* @return An {@link Encryptor} instance.
*/
public static Encryptor newEncryptor() { public static Encryptor newEncryptor() {
return new Encryptor(KEY); return new Encryptor(KEY);
} }
/**
* This class provides the functionality of encryption and decryption with a specific cipher key.
*/
public static class Encryptor { public static class Encryptor {
/**
* The cipher instance.
*/
private Cipher cipher = null; // not thread-safe private Cipher cipher = null; // not thread-safe
private byte[] key = new byte[AES_KEY_SIZE / 16];
private byte[] iv = new byte[12];
/**
* The key material of the secret key.
*
* @see SecretKeySpec#SecretKeySpec(byte[], String)
*/
private final byte[] key = new byte[AES_KEY_SIZE / 16];
/**
* The IV source buffer.
*
* @see GCMParameterSpec#GCMParameterSpec(int, byte[])
*/
private final byte[] iv = new byte[12];
/**
* Constructor.
*
* @param cipherKey The cipher key used to generate {@link Encryptor#key} and {@link Encryptor#iv}.
*/
private Encryptor(String cipherKey) { private Encryptor(String cipherKey) {
try { try {
cipher = Cipher.getInstance("AES/GCM/NoPadding"); cipher = Cipher.getInstance("AES/GCM/NoPadding");
...@@ -55,6 +102,12 @@ public class CryptoUtil { ...@@ -55,6 +102,12 @@ public class CryptoUtil {
} }
} }
/**
* Encrypt given text.
*
* @param text The original text.
* @return Encrypted data.
*/
public byte[] encrypt(String text) { public byte[] encrypt(String text) {
SecretKeySpec keySpec = new SecretKeySpec(key, "AES"); SecretKeySpec keySpec = new SecretKeySpec(key, "AES");
GCMParameterSpec gcmParameterSpec = new GCMParameterSpec(GCM_TAG_LENGTH * 8, iv); GCMParameterSpec gcmParameterSpec = new GCMParameterSpec(GCM_TAG_LENGTH * 8, iv);
...@@ -68,6 +121,12 @@ public class CryptoUtil { ...@@ -68,6 +121,12 @@ public class CryptoUtil {
} }
} }
/**
* Decrypt given data.
*
* @param cipherText Encrypted data.
* @return The original string.
*/
public String decrypt(byte[] cipherText) { public String decrypt(byte[] cipherText) {
SecretKeySpec keySpec = new SecretKeySpec(key, "AES"); SecretKeySpec keySpec = new SecretKeySpec(key, "AES");
GCMParameterSpec gcmParameterSpec = new GCMParameterSpec(GCM_TAG_LENGTH * 8, iv); GCMParameterSpec gcmParameterSpec = new GCMParameterSpec(GCM_TAG_LENGTH * 8, iv);
...@@ -82,10 +141,22 @@ public class CryptoUtil { ...@@ -82,10 +141,22 @@ public class CryptoUtil {
} }
} }
/**
* Compute hash value of given array of bytes.
*
* @param bytes The origin array of bytes.
* @return The array of bytes for the resulting hash value.
*/
public static byte[] sha1(byte[] bytes) { public static byte[] sha1(byte[] bytes) {
return DigestUtils.sha1(bytes); return DigestUtils.sha1(bytes);
} }
/**
* Compute hash value of given string.
*
* @param text The origin string.
* @return The array of bytes for the resulting hash value.
*/
public static byte[] sha1(String text) { public static byte[] sha1(String text) {
return DigestUtils.sha1(text); return DigestUtils.sha1(text);
} }
......
common/src/main/java/com/oceanbase/clogproxy/common/util/Decoder.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.util;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
public class Decoder {
public static String decodeStringInt(ByteBuf buffer) {
if (buffer.readableBytes() < Integer.BYTES) {
return null;
}
buffer.markReaderIndex();
int length = buffer.readInt();
if (buffer.readableBytes() < length) {
buffer.resetReaderIndex();
return null;
}
byte[] bytes = new byte[length];
buffer.readBytes(bytes);
String str = new String(bytes);
if (str.isEmpty()) {
throw new RuntimeException("decode string is null or empty");
}
return str;
}
public static String decodeStringByte(ByteBuf buffer) {
if (buffer.readableBytes() < Byte.BYTES) {
return null;
}
buffer.markReaderIndex();
short length = buffer.readByte();
if (buffer.readableBytes() < length) {
buffer.resetReaderIndex();
return null;
}
byte[] bytes = new byte[length];
buffer.readBytes(bytes);
String str = new String(bytes);
if (str.isEmpty()) {
throw new RuntimeException("decode string is null or empty");
}
return str;
}
public static ByteBuf encodeStringInt(String string) {
if (string == null || string.length() == 0) {
throw new RuntimeException("encode string is null or empty");
}
ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.buffer(4 + string.length());
byteBuf.writeInt(string.length());
byteBuf.writeBytes(string.getBytes());
return byteBuf;
}
}
common/src/main/java/com/oceanbase/clogproxy/common/util/Hex.java
浏览文件 @ 1686e1dc
...@@ -14,24 +14,54 @@ import io.netty.buffer.ByteBufUtil; ...@@ -14,24 +14,54 @@ import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled; import io.netty.buffer.Unpooled;
import org.apache.commons.codec.DecoderException; import org.apache.commons.codec.DecoderException;
/**
* This class is used to convert hexadecimal strings.
*/
public final class Hex { public final class Hex {
/**
* Returns a multi-line hexadecimal dump of the array that is easy to read by humans.
*
* @param array An array of bytes
* @return A multi-line hexadecimal dump string
*/
public static String dump(byte[] array) { public static String dump(byte[] array) {
return dump(array, 0, array.length); return dump(array, 0, array.length);
} }
/**
* Returns a multi-line hexadecimal dump of the specified sub-region of bytes that is easy to read by humans.
*
* @param bytes An array of bytes
* @param offset The offset of the sub-region start position
* @param length The length of the sub-region
* @return A multi-line hexadecimal dump string
*/
public static String dump(byte[] bytes, int offset, int length) { public static String dump(byte[] bytes, int offset, int length) {
return ByteBufUtil.prettyHexDump(Unpooled.wrappedBuffer(bytes, offset, length)); return ByteBufUtil.prettyHexDump(Unpooled.wrappedBuffer(bytes, offset, length));
} }
/**
* Converts an array of bytes into a string representing the hexadecimal values of each byte in order.
*
* @param bytes An array of bytes
* @return A String containing uppercase hexadecimal characters
*/
public static String str(byte[] bytes) { public static String str(byte[] bytes) {
return org.apache.commons.codec.binary.Hex.encodeHexString(bytes, false); return org.apache.commons.codec.binary.Hex.encodeHexString(bytes, false);
} }
public static byte[] toBytes(String hexstr) { /**
* Converts a String representing hexadecimal values into an array of bytes of those same values.
*
* @param hexStr A String representing hexadecimal values
* @return An array of bytes
*/
public static byte[] toBytes(String hexStr) {
try { try {
return org.apache.commons.codec.binary.Hex.decodeHex(hexstr); return org.apache.commons.codec.binary.Hex.decodeHex(hexStr);
} catch (DecoderException e) { } catch (DecoderException e) {
return null; return null;
} }
} }
} }
\ No newline at end of file
common/src/main/java/com/oceanbase/clogproxy/common/util/NetworkUtil.java
浏览文件 @ 1686e1dc
...@@ -13,17 +13,17 @@ package com.oceanbase.clogproxy.common.util; ...@@ -13,17 +13,17 @@ package com.oceanbase.clogproxy.common.util;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import java.net.Inet4Address; import java.net.*;
import java.net.InetAddress;
import java.net.InterfaceAddress;
import java.net.NetworkInterface;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Enumeration; import java.util.Enumeration;
/**
* Utils class for network.
*/
public class NetworkUtil { public class NetworkUtil {
/**
* Local ip.
*/
private static String IP; private static String IP;
static { static {
...@@ -50,9 +50,9 @@ public class NetworkUtil { ...@@ -50,9 +50,9 @@ public class NetworkUtil {
} }
/** /**
* get local ip * Get local ip.
* *
* @return local ip * @return Local ip.
*/ */
public static String getLocalIp() { public static String getLocalIp() {
return IP; return IP;
...@@ -60,6 +60,9 @@ public class NetworkUtil { ...@@ -60,6 +60,9 @@ public class NetworkUtil {
/** /**
* Parse the remote address of the channel. * Parse the remote address of the channel.
*
* @param channel A channel
* @return The address string.
*/ */
public static String parseRemoteAddress(final Channel channel) { public static String parseRemoteAddress(final Channel channel) {
if (null == channel) { if (null == channel) {
...@@ -70,9 +73,10 @@ public class NetworkUtil { ...@@ -70,9 +73,10 @@ public class NetworkUtil {
} }
/** /**
* Parse the address with rules:
* <ol> * <ol>
* <li>if an address starts with a '/', skip it. * <li>If an address starts with a '/', skip it.
* <li>if an address contains a '/', substring it. * <li>If an address contains a '/', substring it.
* </ol> * </ol>
*/ */
private static String doParse(String addr) { private static String doParse(String addr) {
......
common/src/main/java/com/oceanbase/clogproxy/common/util/Password.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.util;
public class Password {
private String value;
public void set(String value) {
this.value = value;
}
public String get() {
return value;
}
@Override
public String toString() {
return "******";
}
}
common/src/main/java/com/oceanbase/clogproxy/common/util/TaskExecutor.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.common.util;
import com.google.common.collect.Maps;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class TaskExecutor {
private static class Singleton {
private static final TaskExecutor INSTANCE = new TaskExecutor();
}
public static TaskExecutor instance() {
return Singleton.INSTANCE;
}
private TaskExecutor() { }
public static class Task<T> {
protected Future<T> future;
protected Failure failure;
public T get() {
try {
return future.get();
} catch (InterruptedException | ExecutionException e) {
if (failure != null) {
failure.onError(e);
}
return null;
}
}
public void join() {
get();
}
}
public static class BackgroundTask extends Task<Void> {
@Override
public void join() {
try {
future.get();
} catch (InterruptedException | ExecutionException e) {
if (failure != null) {
failure.onError(e);
}
}
}
}
public interface Failure {
void onError(Exception e);
}
private ExecutorService asyncTasks = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS, new SynchronousQueue<>(), (ThreadFactory) Thread::new);
private ExecutorService bgTasks = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS, new SynchronousQueue<>(), (ThreadFactory) Thread::new);
private Map<String, ConcurrentTask> concurrentTasks = Maps.newConcurrentMap();
public <T> Task<T> async(Callable<T> callable) {
return async(callable, null);
}
public <T> Task<T> async(Callable<T> callable, Failure failure) {
Task<T> task = new Task<>();
task.future = asyncTasks.submit(callable);
task.failure = failure;
return task;
}
public BackgroundTask background(Callable<Void> callable) {
return background(callable, null);
}
public BackgroundTask background(Callable<Void> callable, Failure failure) {
BackgroundTask task = new BackgroundTask();
task.future = bgTasks.submit(callable);
task.failure = failure;
return task;
}
public static class ConcurrentTask {
private ExecutorService concurrentTasks;
public ConcurrentTask(int parallelism) {
// Never exceed actual CPU core count for init count, or got an Exception
concurrentTasks = new ForkJoinPool(Math.min(parallelism,
Runtime.getRuntime().availableProcessors()),
ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true);
}
public <T> Future<T> concurrent(Callable<T> callable) {
return concurrentTasks.submit(callable);
}
}
public ConcurrentTask refConcurrent(String name, int parallelism) {
ConcurrentTask task = concurrentTasks.get(name);
if (task != null) {
return task;
}
task = new ConcurrentTask(parallelism);
concurrentTasks.put(name, task);
return task;
}
public int getAsyncTaskCount() {
return ((ThreadPoolExecutor) asyncTasks).getActiveCount();
}
public int getBgTaskCount() {
return ((ThreadPoolExecutor) bgTasks).getActiveCount();
}
public int getConcurrentTaskCount() {
int count = 0;
for (ConcurrentTask t : concurrentTasks.values()) {
count += ((ForkJoinPool) t.concurrentTasks).getActiveThreadCount();
}
return count;
}
}
common/src/main/java/com/oceanbase/clogproxy/common/util/TypeTrait.java
浏览文件 @ 1686e1dc
...@@ -14,56 +14,123 @@ import org.apache.commons.lang3.StringUtils; ...@@ -14,56 +14,123 @@ import org.apache.commons.lang3.StringUtils;
import java.lang.reflect.Field; import java.lang.reflect.Field;
/**
* Utils class to check and convert data types.
*/
public class TypeTrait { public class TypeTrait {
/**
* Checks if it is a number type object.
*
* @param obj An object to check.
* @return True if it is a number type object, false otherwise.
*/
public static boolean isNumber(Object obj) { public static boolean isNumber(Object obj) {
return (obj instanceof Byte) || (obj instanceof Short) || return (obj instanceof Byte) || (obj instanceof Short) ||
(obj instanceof Integer) || (obj instanceof Long); (obj instanceof Integer) || (obj instanceof Long);
} }
/**
* Checks if it is a number type {@link Field}.
*
* @param field A field to check.
* @return True if it is a number type field, false otherwise.
*/
public static boolean isNumber(Field field) { public static boolean isNumber(Field field) {
String typeName = field.getGenericType().getTypeName(); String typeName = field.getGenericType().getTypeName();
return "byte".equals(typeName) || "java.lang.Byte".equals(typeName) || return "byte".equals(typeName) || "java.lang.Byte".equals(typeName) ||
"short".equals(typeName) || "java.lang.Short".equals(typeName) || "short".equals(typeName) || "java.lang.Short".equals(typeName) ||
"int".equals(typeName) || "java.lang.Integer".equals(typeName) || "int".equals(typeName) || "java.lang.Integer".equals(typeName) ||
"long".equals(typeName) || "java.lang.Long".equals(typeName); "long".equals(typeName) || "java.lang.Long".equals(typeName);
} }
/**
* Checks if it is a real number type object.
*
* @param obj An object to check.
* @return True if it is a real number type object, false otherwise.
*/
public static boolean isReal(Object obj) { public static boolean isReal(Object obj) {
return (obj instanceof Float) || (obj instanceof Double); return (obj instanceof Float) || (obj instanceof Double);
} }
/**
* Checks if it is a real number type {@link Field}.
*
* @param field A field to check.
* @return True if it is a real number type field, false otherwise.
*/
public static boolean isReal(Field field) { public static boolean isReal(Field field) {
String typeName = field.getGenericType().getTypeName(); String typeName = field.getGenericType().getTypeName();
return "float".equals(typeName) || "java.lang.Float".equals(typeName) || return "float".equals(typeName) || "java.lang.Float".equals(typeName) ||
"double".equals(typeName) || "java.lang.Double".equals(typeName); "double".equals(typeName) || "java.lang.Double".equals(typeName);
} }
/**
* Checks if it is a boolean type object.
*
* @param obj An object to check.
* @return True if it is a boolean type object, false otherwise.
*/
public static boolean isBool(Object obj) { public static boolean isBool(Object obj) {
return obj instanceof Boolean; return obj instanceof Boolean;
} }
/**
* Checks if it is a boolean type {@link Field}.
*
* @param field A field to check.
* @return True if it is a boolean type field, false otherwise.
*/
public static boolean isBool(Field field) { public static boolean isBool(Field field) {
String typeName = field.getGenericType().getTypeName(); String typeName = field.getGenericType().getTypeName();
return "boolean".equals(typeName) || "java.lang.Boolean".equals(typeName); return "boolean".equals(typeName) || "java.lang.Boolean".equals(typeName);
} }
/**
* Checks if it is a string type object.
*
* @param obj An object to check.
* @return True if it is a string type object, false otherwise.
*/
public static boolean isString(Object obj) { public static boolean isString(Object obj) {
return (obj instanceof Character) || (obj instanceof String); return (obj instanceof Character) || (obj instanceof String);
} }
/**
* Checks if it is a string type {@link Field}.
*
* @param field A field to check.
* @return True if it is a string type field, false otherwise.
*/
public static boolean isString(Field field) { public static boolean isString(Field field) {
String typeName = field.getGenericType().getTypeName(); String typeName = field.getGenericType().getTypeName();
return "char".equals(typeName) || "java.lang.Character".equals(typeName) || return "char".equals(typeName) || "java.lang.Character".equals(typeName) ||
"java.lang.String".equals(typeName); "java.lang.String".equals(typeName);
} }
/**
* Checks if the object and field are the same loose type.
*
* @param object An object to check.
* @param field A field to check.
* @return True if the object and field are the same loose type, false otherwise.
*/
public static boolean isSameLooseType(Object object, Field field) { public static boolean isSameLooseType(Object object, Field field) {
return (isNumber(object) && isNumber(field)) || return (isNumber(object) && isNumber(field)) ||
(isReal(object) && isReal(field)) || (isReal(object) && isReal(field)) ||
(isBool(object) && isBool(field)) || (isBool(object) && isBool(field)) ||
(isString(object) && isString(field)); (isString(object) && isString(field));
} }
/**
* Convert a value from string type.
*
* @param value The source string.
* @param clazz Value class.
* @param <T> Expected value type.
* @return The value converted from string.
*/
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public static <T> T fromString(String value, Class<?> clazz) { public static <T> T fromString(String value, Class<?> clazz) {
if (clazz == Byte.class || clazz == byte.class) { if (clazz == Byte.class || clazz == byte.class) {
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/LogProxyClient.java
浏览文件 @ 1686e1dc
...@@ -21,46 +21,87 @@ import com.oceanbase.clogproxy.client.util.Validator; ...@@ -21,46 +21,87 @@ import com.oceanbase.clogproxy.client.util.Validator;
import com.oceanbase.clogproxy.common.packet.ProtocolVersion; import com.oceanbase.clogproxy.common.packet.ProtocolVersion;
import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContext;
/**
* A client that makes it easy to connect to log proxy and start a {@link ClientStream}.
*/
public class LogProxyClient { public class LogProxyClient {
/**
* A {@link ClientStream} instance.
*/
private final ClientStream stream; private final ClientStream stream;
/** /**
* @param host server hostname name or ip * Constructor with {@link SslContext}.
* @param port server port *
* @param config real config object according to what-you-expected * @param host Log proxy hostname name or ip.
* @param sslContext ssl context to create netty handler * @param port Log proxy port.
* @param config {@link AbstractConnectionConfig} used to create the {@link ClientStream}.
* @param sslContext {@link SslContext} to create netty handler.
*/ */
public LogProxyClient(String host, int port, AbstractConnectionConfig config, SslContext sslContext) { public LogProxyClient(String host, int port, AbstractConnectionConfig config, SslContext sslContext) {
Validator.notNull(config.getLogType(), "log type cannot be null"); try {
Validator.notNull(host, "server cannot be null"); Validator.notNull(config.getLogType(), "log type cannot be null");
Validator.validatePort(port, "port is not valid"); Validator.notEmpty(host, "server cannot be null");
Validator.validatePort(port, "port is not valid");
} catch (Exception e) {
throw new IllegalArgumentException("Illegal argument for LogProxyClient");
}
if (!config.valid()) {
throw new IllegalArgumentException("Illegal argument for LogProxyClient");
}
String clientId = ClientConf.USER_DEFINED_CLIENTID.isEmpty() ? ClientIdGenerator.generate() : ClientConf.USER_DEFINED_CLIENTID; String clientId = ClientConf.USER_DEFINED_CLIENTID.isEmpty() ? ClientIdGenerator.generate() : ClientConf.USER_DEFINED_CLIENTID;
ConnectionParams connectionParams = new ConnectionParams(config.getLogType(), clientId, host, port, config); ConnectionParams connectionParams = new ConnectionParams(config.getLogType(), clientId, host, port, config);
connectionParams.setProtocolVersion(ProtocolVersion.V2); connectionParams.setProtocolVersion(ProtocolVersion.V2);
this.stream = new ClientStream(connectionParams, sslContext); this.stream = new ClientStream(connectionParams, sslContext);
} }
/**
* Constructor without {@link SslContext}.
*
* @param host Log proxy hostname name or ip.
* @param port Log proxy port.
* @param config {@link AbstractConnectionConfig} used to create the {@link ClientStream}.
*/
public LogProxyClient(String host, int port, AbstractConnectionConfig config) { public LogProxyClient(String host, int port, AbstractConnectionConfig config) {
this(host, port, config, null); this(host, port, config, null);
} }
/**
* Start the client.
*/
public void start() { public void start() {
stream.start(); stream.start();
} }
/**
* Stop the client.
*/
public void stop() { public void stop() {
stream.stop(); stream.stop();
} }
/**
* Join and wait the client.
*/
public void join() { public void join() {
stream.join(); stream.join();
} }
/**
* Add a {@link RecordListener} to {@link #stream}.
*
* @param recordListener A {@link RecordListener}.
*/
public synchronized void addListener(RecordListener recordListener) { public synchronized void addListener(RecordListener recordListener) {
stream.addListener(recordListener); stream.addListener(recordListener);
} }
/**
* Add a {@link StatusListener} to {@link #stream}.
*
* @param statusListener A {@link StatusListener}.
*/
public synchronized void addStatusListener(StatusListener statusListener) { public synchronized void addStatusListener(StatusListener statusListener) {
stream.addStatusListener(statusListener); stream.addStatusListener(statusListener);
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/config/AbstractConnectionConfig.java
浏览文件 @ 1686e1dc
...@@ -17,33 +17,57 @@ import java.util.HashMap; ...@@ -17,33 +17,57 @@ import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.Map.Entry; import java.util.Map.Entry;
/**
* This is an abstract implementation class of the interface {@link ConnectionConfig}.
*/
public abstract class AbstractConnectionConfig implements ConnectionConfig { public abstract class AbstractConnectionConfig implements ConnectionConfig {
/** /**
* defined structure configurations * Defined configurations map.
*/ */
protected static Map<String, ConfigItem<Object>> configs = new HashMap<>(); protected static Map<String, ConfigItem<Object>> configs = new HashMap<>();
/** /**
* extra configurations * Extra configurations map.
*/ */
protected final Map<String, String> extraConfigs = new HashMap<>(); protected final Map<String, String> extraConfigs = new HashMap<>();
/**
* This class is used to define configuration with a default value.
*
* @param <T> The type of stored value.
*/
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
protected static class ConfigItem<T> { protected static class ConfigItem<T> {
protected String key; protected String key;
protected T val; protected T val;
/**
* Sole constructor.
*
* @param key Config key.
* @param val Config value.
*/
public ConfigItem(String key, T val) { public ConfigItem(String key, T val) {
this.key = key; this.key = key;
this.val = val; this.val = val;
configs.put(key, (ConfigItem<Object>) this); configs.put(key, (ConfigItem<Object>) this);
} }
/**
* Set value to config item.
*
* @param val Value of specific type.
*/
public void set(T val) { public void set(T val) {
this.val = val; this.val = val;
} }
/**
* Set value of specific type from string.
*
* @param val Value of string type.
*/
public void fromString(String val) { public void fromString(String val) {
this.val = TypeTrait.fromString(val, this.val.getClass()); this.val = TypeTrait.fromString(val, this.val.getClass());
} }
...@@ -54,6 +78,11 @@ public abstract class AbstractConnectionConfig implements ConnectionConfig { ...@@ -54,6 +78,11 @@ public abstract class AbstractConnectionConfig implements ConnectionConfig {
} }
} }
/**
* Sole constructor.
*
* @param allConfigs The map of configurations.
*/
public AbstractConnectionConfig(Map<String, String> allConfigs) { public AbstractConnectionConfig(Map<String, String> allConfigs) {
if (allConfigs != null) { if (allConfigs != null) {
for (Entry<String, String> entry : allConfigs.entrySet()) { for (Entry<String, String> entry : allConfigs.entrySet()) {
...@@ -66,12 +95,28 @@ public abstract class AbstractConnectionConfig implements ConnectionConfig { ...@@ -66,12 +95,28 @@ public abstract class AbstractConnectionConfig implements ConnectionConfig {
} }
} }
/**
* Get log type set in configurations.
*
* @return The enum constant of {@link LogType}.
*/
public abstract LogType getLogType(); public abstract LogType getLogType();
/**
* Add configurations to {@link #extraConfigs}
*
* @param extraConfigs A map of configurations.
*/
public void setExtraConfigs(Map<String, String> extraConfigs) { public void setExtraConfigs(Map<String, String> extraConfigs) {
this.extraConfigs.putAll(extraConfigs); this.extraConfigs.putAll(extraConfigs);
} }
/**
* Update value into define configurations map.
*
* @param key Config key.
* @param val New config value.
*/
void set(String key, String val) { void set(String key, String val) {
ConfigItem<Object> cs = configs.get(key); ConfigItem<Object> cs = configs.get(key);
if (cs != null) { if (cs != null) {
...@@ -80,9 +125,9 @@ public abstract class AbstractConnectionConfig implements ConnectionConfig { ...@@ -80,9 +125,9 @@ public abstract class AbstractConnectionConfig implements ConnectionConfig {
} }
/** /**
* validate if defined configurations * Validate defined configurations.
* *
* @return True or False * @return Flag of whether all the defined configurations are valid.
*/ */
public abstract boolean valid(); public abstract boolean valid();
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/config/ClientConf.java
浏览文件 @ 1686e1dc
...@@ -10,29 +10,58 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,29 +10,58 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.config; package com.oceanbase.clogproxy.client.config;
import com.oceanbase.clogproxy.common.config.ShareConf; import com.oceanbase.clogproxy.common.config.SharedConf;
public class ClientConf extends ShareConf { /**
public static final String VERSION = "1.1.0"; * The class that defines the constants that are used to generate the connection.
*/
public class ClientConf extends SharedConf {
/**
* Client version.
*/
public static final String VERSION = "1.0.1";
/**
* Queue size for storing records received from log proxy.
*/
public static int TRANSFER_QUEUE_SIZE = 20000; public static int TRANSFER_QUEUE_SIZE = 20000;
/**
* Connection timeout in milliseconds.
*/
public static int CONNECT_TIMEOUT_MS = 5000; public static int CONNECT_TIMEOUT_MS = 5000;
/**
* Reading queue timeout in milliseconds.
*/
public static int READ_WAIT_TIME_MS = 2000; public static int READ_WAIT_TIME_MS = 2000;
/**
* Time to sleep in seconds when retrying.
*/
public static int RETRY_INTERVAL_S = 2; public static int RETRY_INTERVAL_S = 2;
/** /**
* max retry time after disconnect, if not data income lasting IDLE_TIMEOUT_S, a reconnect we be trigger * Maximum number of retries after disconnect, if not data income lasting {@link #IDLE_TIMEOUT_S}, a reconnection will be triggered.
*/ */
public static int MAX_RECONNECT_TIMES = -1; public static int MAX_RECONNECT_TIMES = -1;
/**
* Idle timeout in seconds for netty handler.
*/
public static int IDLE_TIMEOUT_S = 15; public static int IDLE_TIMEOUT_S = 15;
/**
* Maximum number of reads, after which data will be discarded.
*/
public static int NETTY_DISCARD_AFTER_READS = 16; public static int NETTY_DISCARD_AFTER_READS = 16;
/** /**
* set user defined userid, * User defined client id.
* for inner use only
*/ */
public static String USER_DEFINED_CLIENTID = ""; public static String USER_DEFINED_CLIENTID = "";
/** /**
* ignore unknown or unsupported record type with a warning log instead throwing an exception * Ignore unknown or unsupported record type with a warning log instead of throwing an exception.
*/ */
public static boolean IGNORE_UNKNOWN_RECORD_TYPE = false; public static boolean IGNORE_UNKNOWN_RECORD_TYPE = false;
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/config/ConnectionConfig.java
浏览文件 @ 1686e1dc
...@@ -10,10 +10,30 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,10 +10,30 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.config; package com.oceanbase.clogproxy.client.config;
/**
* This is the interface of connection config.
*/
public interface ConnectionConfig { public interface ConnectionConfig {
/**
* Generate a configuration string from connection parameters.
*
* @return The configuration string.
*/
String generateConfigurationString(); String generateConfigurationString();
/**
* Update the checkpoint.
*
* @param checkpoint A checkpoint string.
*/
void updateCheckpoint(String checkpoint); void updateCheckpoint(String checkpoint);
/**
* Overrides {@link Object#toString()} to structure a string.
*
* @return The structured string.
*/
@Override
String toString(); String toString();
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/config/DRCConfig.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.config;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import com.oceanbase.clogproxy.client.fliter.DataFilterBase;
import com.oceanbase.clogproxy.client.message.Checkpoint;
import com.oceanbase.clogproxy.client.util.StringUtils;
public class DRCConfig {
public enum DataAcquireOption {
ASK_SELF_UNIT, ASK_OTHER_UNIT, ASK_ALL_DATA,
};
/* Configures used by DRCClient. */
private final Map<String, String> configures;
/* All configures used to send to the server is stored */
private final Map<String, String> userDefinedParams;
/* All persistent-required attributes. */
private final Set<String> persists;
private Checkpoint checkpoint;
private DataFilterBase filter;
private String blackList;
private int recordsPerBatch = 0;
private int maxRetryTimes = 100;
private int socketTimeout = 10;
private int connectionTimeout = 10;
private boolean useBinaryFormat = false;
private boolean txnMark = true;
private boolean requireCompleteTxn = false;
private int maxRecordsCached = 10240;
private int maxRecordsBatched = 1024;
private int maxTxnsBatched = 10240;
private long maxTimeoutBatched = 500; // ms
private boolean useDrcNet = false;
private boolean drcMarkWorking = false;
private boolean requiredOtherUnitData = false;
private boolean usePublicIp = false;
private boolean useCaseSensitive = false;
private boolean trimLongType = false;
private boolean useCheckCRC = false;
private boolean useIndexIterator = false;
private boolean needUKRecord = false;
/* Properties defined for DRCClient, always defined in drc.properties. */
private final static String DRC_VERSION = "version";
private final static String DRC_MANAGERHOST = "manager.host";
private final static String HTTPS_USE = "client.https";
private final static String DRC_BINLOGLOGNAME = "DRCClient.Binlog";
private final static String DRC_CHECKPOINT_POLLPERIOD = "checkpoint.period";
private static final String SERVER_MAX_RETRIES = "server.maxRetriedTimes";
private static final String SERVER_MESSAGE_TYPE = "server.messageType";
private static final String CLIENT_SO_TIMEOUT = "client.socketTimeout";
private static final String CLIENT_CONN_TIMEOUT = "client.connectionTimeout";
private static final String CLIENT_REQ_COMP_TXN = "client.requireCompleteTxn";
private static final String CLIENT_MAX_RECS_CACHE = "client.maxNumOfRecordsCached";
private static final String CLIENT_MAX_RECS_BATCH = "client.maxNumOfRecordsPerMessage";
private static final String CLIENT_MAX_TXNS_BATCH = "client.maxNumOfTxnsPerMessage";
private static final String CLIENT_MAX_TIMEOUT_BATCH = "client.maxTimeoutPerMessage";
private static final String CLIENT_USE_INDEX_ITERATOR = "enableIndexIter";
/* Parameters could be sent to remote servers, usually provided by users. */
private final static String USER_FILTERCONDITIONS = "condition";
private final static String USER_DBNAME = "dbname";
private final static String USER_GROUPNAME = "groupname";
private final static String USER_IDENTIFICATION = "password";
private final static String USER_GROUP = "group";
private final static String USER_SUBGROUP = "subgroup";
private final static String USER_MYSQL = "instance";
private final static String DRC_MARK = "drcMark";
private final static String NEED_UK_RECORD = "needUKRecord";
private final static String BLACK_REGION_NO = "black_region_no";
private final static String DEFAULT_DRC_MARK = "drc.t*x_begin4unit_mark_[0-9]*|*.drc_txn";
private static final String USE_DRC_NET = "useDrcNet";
private static final String IPMAPS = "ipmaps";
//from 40，if store receive this parameter，timestamp return is unix timestamp，otherwise yyyy-mm-dd hh:mm:ss
public static final String CLIENT_VERSION = "client.version";
public static final String CLIENT_VERSION_ID = "58_SP";
public static final String REQUIRE_OTHER_UNIT_DATA = "askOtherUnit";
public static final String DATA_REQUIRED_OPTION = "client.data_acquire_option";
public static final String SELF_UNIT = "self_unit";
public static final String OTHER_UNIT = "other_unit";
@Deprecated
private final static String USER_FILTERSTRICT = "strict";
@Deprecated
private final static String USER_FILTERWHERE = "where";
/*
* Mark the persistent location information in the local file.
*/
public final static String POSITION_INFO = "Global_position_info:";
private static Map<String, String> ipportMaps;
private String regionId;
private String blackRegionNo;
/**
* Private constructor providing primary initialization.
*/
private DRCConfig() {
configures = new HashMap<String, String>();
userDefinedParams = new HashMap<String, String>();
userDefinedParams.put(DRC_MARK, DEFAULT_DRC_MARK);
persists = new HashSet<String>();
configures.put(DRC_VERSION, "2.0.0");
configures.put(DRC_CHECKPOINT_POLLPERIOD, "500");
configures.put(CLIENT_SO_TIMEOUT, "120");
configures.put(CLIENT_CONN_TIMEOUT, "120");
checkpoint = new Checkpoint();
ipportMaps = new HashMap<String, String>();
useDrcNet = false;
}
/**
* Read configures from a reader.
* @param reader java.io.Reader
* @throws IOException when properties load failed
*/
public DRCConfig(final Reader reader) throws IOException {
this();
Properties properties = new Properties();
properties.load(reader);
loadProperties(properties);
}
/**
* Read configures from a properties file.
* @param propertiesFile is the file under the classpath.
* @throws IOException when properties load failed
*/
public DRCConfig(final String propertiesFile) throws IOException {
this();
InputStream drcProperties = this.getClass().getClassLoader()
.getResourceAsStream(propertiesFile);
Properties properties = new Properties();
properties.load(drcProperties);
loadProperties(properties);
}
public DRCConfig(final Properties properties) {
this();
loadProperties(properties);
}
/**
* Use privately to load all properties to inner map.
* @param properties
*/
private void loadProperties(Properties properties) {
for (Entry<Object, Object> entry : properties.entrySet()) {
if (entry.getKey().equals(SERVER_MAX_RETRIES)) {
maxRetryTimes = Integer.parseInt((String) entry.getValue());
} else if (entry.getKey().equals(CLIENT_SO_TIMEOUT)) {
socketTimeout = Integer.parseInt((String) entry.getValue());
} else if (entry.getKey().equals(CLIENT_CONN_TIMEOUT)) {
connectionTimeout = Integer.parseInt((String) entry.getValue());
} else if (entry.getKey().equals(SERVER_MESSAGE_TYPE)) {
useBinaryFormat = "binary".equals(entry.getValue());
} else if (entry.getKey().equals(CLIENT_REQ_COMP_TXN)) {
requireCompleteTxn = "true".equals(entry.getValue());
} else if (entry.getKey().equals(CLIENT_MAX_RECS_BATCH)) {
// requireCompleteTxn = true;
maxRecordsBatched = Integer.parseInt((String) entry.getValue());
} else if (entry.getKey().equals(CLIENT_MAX_TIMEOUT_BATCH)) {
// requireCompleteTxn = true;
maxTimeoutBatched = Long.parseLong((String) entry.getValue());
} else if (entry.getKey().equals(CLIENT_MAX_TXNS_BATCH)) {
maxTxnsBatched = Integer.parseInt((String) entry.getValue());
} else if (entry.getKey().equals(CLIENT_MAX_RECS_CACHE)) {
maxRecordsCached = Integer.parseInt((String) entry.getValue());
} else if (entry.getKey().equals(IPMAPS)) {
String[] ipmaps = StringUtils.split((String) entry.getValue(), '|');
for (String ippair : ipmaps) {
String[] ips = StringUtils.split(ippair, '-');
if (ips.length == 2) {
ipportMaps.put(ips[0], ips[1]);
}
}
} else if (entry.getKey().equals(USE_DRC_NET)) {
useDrcNet = "true".equals(entry.getValue());
} else if (entry.getKey().equals(CLIENT_USE_INDEX_ITERATOR)) {
useIndexIterator = "true".equals(entry.getValue());
} else if (entry.getKey().equals(DATA_REQUIRED_OPTION)) {
String value = ((String) entry.getValue());
DataAcquireOption dataAcquireOption = DataAcquireOption.ASK_ALL_DATA;
if (org.apache.commons.lang3.StringUtils.equals(value, SELF_UNIT)) {
dataAcquireOption = DataAcquireOption.ASK_SELF_UNIT;
} else if (org.apache.commons.lang3.StringUtils.equals(value, OTHER_UNIT)) {
dataAcquireOption = DataAcquireOption.ASK_OTHER_UNIT;
}
setRequireOtherUnitData(dataAcquireOption);
} else if (entry.getKey().equals(NEED_UK_RECORD)) {
needUKRecord = "true".equals(entry.getValue());
} else if (entry.getKey().equals(BLACK_REGION_NO)) {
blackRegionNo = (String) entry.getValue();
}
configures.put((String) entry.getKey(), (String) entry.getValue());
}
}
public boolean needUKRecord() {
return needUKRecord;
}
public String getBlackRegionNo() {
return blackRegionNo;
}
public boolean getUseDrcNet() {
return useDrcNet;
}
public void setUseDrcNet(boolean useDrcNet) {
this.useDrcNet = useDrcNet;
}
public void setForceUseIndexIter(boolean forceUseIndexIter) {
this.useIndexIterator = forceUseIndexIter;
}
public boolean getForceUseIndexIter() {
return this.useIndexIterator;
}
public void setDataFilter(final DataFilterBase filter) {
this.filter = filter;
}
public DataFilterBase getDataFilter() {
return filter;
}
public Checkpoint getCheckpoint() {
return checkpoint;
}
public int getMaxRetriedTimes() {
return maxRetryTimes;
}
public int getSocketTimeout() {
return socketTimeout;
}
public int getConnectionTimeout() {
return connectionTimeout;
}
/**
* Get the group name.
* @return the group name.
*/
final String getGroupName() {
return userDefinedParams.get(USER_GROUPNAME);
}
/**
* Set the group name.
* @param groupName is the group name.
*/
final void setGroupName(final String groupName) {
userDefinedParams.put(USER_GROUPNAME, groupName);
}
/**
* Get the group title, if empty, use group name (user) instead.
* @return
*/
final String getGroup() {
final String group = userDefinedParams.get(USER_GROUP);
if (group == null || group.isEmpty()) {
return getGroupName();
}
return group;
}
/**
* Set the group title, note that the group could be different
* from groupname(user name).
* @param group
*/
final void setGroup(final String group) {
userDefinedParams.put(USER_GROUP, group);
}
/**
* Set the target physical database name.
* @param dbname database name such as icdb0, uic_main_000 and so on.
*/
final void setDbname(final String dbname) {
userDefinedParams.put(USER_DBNAME, dbname);
}
/**
* Set the subgroup name.
* @param subgroup
*/
final void setSubGroup(final String subgroup) {
userDefinedParams.put(USER_SUBGROUP, subgroup);
}
/**
* Get the subgroup name, if empty, use db name instead
* @return subgroup
*/
final String getSubGroup() {
final String subgroup = userDefinedParams.get(USER_SUBGROUP);
if (subgroup == null || subgroup.isEmpty()) {
return getDbname();
}
return subgroup;
}
/**
* Get the target physical database name.
* @return the database name.
*/
final String getDbname() {
return userDefinedParams.get(USER_DBNAME);
}
/**
* Set the user's identification, e.t., password.
* @param id is the identification.
*/
final void setIdentification(final String id) {
userDefinedParams.put(USER_IDENTIFICATION, id);
}
/**
* Set a filename to store checkpoint.
* @param filename
*/
final void setBinlogFilename(final String filename) {
configures.put(DRC_BINLOGLOGNAME, filename);
}
final String getIdentification() {
return userDefinedParams.get(USER_IDENTIFICATION);
}
/**
* Get the binlog filename.
* @return the filename.
*/
final String getBinlogFilename() {
return configures.get(DRC_BINLOGLOGNAME);
}
/**
* Get the period to record one checkpoint.
* @return the period.
*/
final int getCheckpointPeriod() {
return Integer.parseInt(configures.get(DRC_CHECKPOINT_POLLPERIOD));
}
/**
* Set user-defined checkpoint.
* @param checkpoint
*/
final void setCheckpoint(final Checkpoint checkpoint) {
this.checkpoint = checkpoint;
}
final void setCheckpoint(final String checkpoint) {
this.checkpoint.setPosition(checkpoint);
}
/**
* Set user-defined starting time stamp.
* @param timestamp
*/
final void setGmtModified(final String timestamp) {
checkpoint.setTimestamp(timestamp);
}
/**
* Set checkpoint or gmtModified as the starting point.
* @param startingPoint
*/
final void setStartingPoint(final String startingPoint) {
if (startingPoint.contains("@")) {
setCheckpoint(startingPoint);
} else {
if (startingPoint.length() == 13) {
throw new IllegalArgumentException(
"Error the unit of the starting time is second, but " + startingPoint
+ " is in ms");
}
setGmtModified(startingPoint);
}
}
/**
* Get meta version.
* @return meta version.
*/
final String getVersion() {
return configures.get(DRC_VERSION);
}
/**
* Get address of the cluster manager.
* @return a URL.
*/
final String getClusterManagerAddresses() {
return configures.get(DRC_MANAGERHOST);
}
/**
* Set the connected mysql address.
* @param mysql is the mysql address.
*/
final void setInstance(final String mysql) {
userDefinedParams.put(USER_MYSQL, mysql);
checkpoint.setServerId(mysql);
}
/**
* Get the last connected mysql address.
* @return the mysql address.
*/
final String getInstance() {
return userDefinedParams.get(USER_MYSQL);
}
/**
* Usually use internally to add user-defined parameters.
* @param key is the name of the parameter.
* @param value is the value of the parameter.
*/
final void addParam(final String key, final String value) {
userDefinedParams.put(key, value);
}
/**
* Get user-defined parameter by name.
* @param key is the name of the parameter.
* @return
*/
final String getParam(final String key) {
return userDefinedParams.get(key);
}
/**
* Get all user-defined parameters.
* @return parameters used by the server.
*/
final Map<String, String> getParams() {
return userDefinedParams;
}
/**
* Usually use internally to add drc-related configures.
* @param key is the name of the configure.
* @param value is the value of the configure.
*/
final void addConfigure(final String key, final String value) {
configures.put(key, value);
}
/**
* Get the value of one specific parameter.
* @param key the name of the parameter.
* @return
*/
final String getConfigure(final String key) {
return configures.get(key);
}
/**
* Get all configures.
* @return all the configures.
*/
final Map<String, String> getConfigures() {
return configures;
}
/**
* Get the persistent-required attributes' names.
* @return the list of names.
*/
final Set<String> getPersists() {
return persists;
}
/**
* Add more persistent-required attributes.
* @param p is the new attributes.
*/
final void addPersists(List<String> p) {
persists.addAll(p);
}
/**
* Set user-wanted tables and columns.
* @param conditions well-formatted tables and columns.
*/
final void setRequiredTablesAndColumns(final String conditions) {
userDefinedParams.put(USER_FILTERCONDITIONS, conditions);
}
/**
* Set user-wanted where conditions.
* @param where well-formated where conditions.
*/
final void setWhereFilters(final String where) {
userDefinedParams.put(USER_FILTERWHERE, where);
}
/**
* Define that the user only want a record which at least has one
* field required changed.
*/
final void setFilterUnchangedRecords() {
userDefinedParams.put(USER_FILTERSTRICT, "true");
}
public void setDRCMark(String mark) {
userDefinedParams.put(DRC_MARK, mark);
}
public String getDRCMark() {
return userDefinedParams.get(DRC_MARK);
}
final void requireTxnMark(boolean need) {
txnMark = need;
}
final boolean isTxnMarkRequired() {
return txnMark;
}
final boolean isBinaryFormat() {
return useBinaryFormat;
}
public void setNumOfRecordsPerBatch(int threshold) {
recordsPerBatch = threshold;
}
public int getNumOfRecordsPerBatch() {
return recordsPerBatch;
}
public boolean isTxnRequiredCompleted() {
return requireCompleteTxn;
}
public int getMaxRecordsPerTxn() {
return maxRecordsCached;
}
public int getMaxRecordsBatched() {
return maxRecordsBatched;
}
public long getMaxTimeoutBatched() {
return maxTimeoutBatched;
}
public int getMaxTxnsBatched() {
return maxTxnsBatched;
}
public final String getMappedIpPort(final String ip) {
return ipportMaps.get(ip);
}
public void useDrcMark() {
drcMarkWorking = true;
}
public boolean isDrcMarkWorking() {
return drcMarkWorking;
}
public String getBlackList() {
return blackList;
}
public void setBlackList(String blackList) {
this.blackList = blackList;
}
public boolean getUseHTTPS() {
String use = configures.get(HTTPS_USE);
return use != null && !"false".equals(use);
}
public void usePublicIp() {
usePublicIp = true;
}
public boolean isUsePublicIp() {
return usePublicIp;
}
public void useCaseSensitive() {
useCaseSensitive = true;
}
public boolean isUseCaseSensitive() {
return useCaseSensitive;
}
public void trimLongType() {
trimLongType = true;
}
public boolean isTrimLongType() {
return trimLongType;
}
public boolean isUseCheckCRC() {
return useCheckCRC;
}
public void setUseCheckCRC(boolean useCheckCRC) {
this.useCheckCRC = useCheckCRC;
}
public String getRegionId() {
return regionId;
}
public void setRegionId(String regionId) {
this.regionId = regionId;
}
public void setRequireOtherUnitData(DataAcquireOption dataAcquireOption) {
switch (dataAcquireOption) {
case ASK_SELF_UNIT: {
this.requiredOtherUnitData = false;
this.drcMarkWorking = true;
break;
}
case ASK_OTHER_UNIT: {
this.requiredOtherUnitData = true;
this.drcMarkWorking = false;
break;
}
case ASK_ALL_DATA: {
this.requiredOtherUnitData = false;
this.drcMarkWorking = false;
break;
}
}
}
public boolean getRequireOtherUnitData() {
return this.requiredOtherUnitData;
}
}
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/config/ObReaderConfig.java
浏览文件 @ 1686e1dc
...@@ -12,7 +12,7 @@ package com.oceanbase.clogproxy.client.config; ...@@ -12,7 +12,7 @@ package com.oceanbase.clogproxy.client.config;
import com.google.common.collect.Maps; import com.google.common.collect.Maps;
import com.oceanbase.clogproxy.client.util.Validator; import com.oceanbase.clogproxy.client.util.Validator;
import com.oceanbase.clogproxy.common.config.ShareConf; import com.oceanbase.clogproxy.common.config.SharedConf;
import com.oceanbase.clogproxy.common.packet.LogType; import com.oceanbase.clogproxy.common.packet.LogType;
import com.oceanbase.clogproxy.common.util.CryptoUtil; import com.oceanbase.clogproxy.common.util.CryptoUtil;
import com.oceanbase.clogproxy.common.util.Hex; import com.oceanbase.clogproxy.common.util.Hex;
...@@ -21,19 +21,49 @@ import org.slf4j.LoggerFactory; ...@@ -21,19 +21,49 @@ import org.slf4j.LoggerFactory;
import java.util.Map; import java.util.Map;
/**
* This is a configuration class for connection to log proxy.
*/
public class ObReaderConfig extends AbstractConnectionConfig { public class ObReaderConfig extends AbstractConnectionConfig {
private static final Logger logger = LoggerFactory.getLogger(ObReaderConfig.class); private static final Logger logger = LoggerFactory.getLogger(ObReaderConfig.class);
/**
* Root server list.
*/
private static final ConfigItem<String> RS_LIST = new ConfigItem<>("rootserver_list", ""); private static final ConfigItem<String> RS_LIST = new ConfigItem<>("rootserver_list", "");
/**
* Cluster username.
*/
private static final ConfigItem<String> CLUSTER_USER = new ConfigItem<>("cluster_user", ""); private static final ConfigItem<String> CLUSTER_USER = new ConfigItem<>("cluster_user", "");
/**
* Cluster password.
*/
private static final ConfigItem<String> CLUSTER_PASSWORD = new ConfigItem<>("cluster_password", ""); private static final ConfigItem<String> CLUSTER_PASSWORD = new ConfigItem<>("cluster_password", "");
/**
* Table whitelist.
*/
private static final ConfigItem<String> TABLE_WHITE_LIST = new ConfigItem<>("tb_white_list", ""); private static final ConfigItem<String> TABLE_WHITE_LIST = new ConfigItem<>("tb_white_list", "");
/**
* Start timestamp.
*/
private static final ConfigItem<Long> START_TIMESTAMP = new ConfigItem<>("first_start_timestamp", 0L); private static final ConfigItem<Long> START_TIMESTAMP = new ConfigItem<>("first_start_timestamp", 0L);
/**
* Constructor with empty arguments.
*/
public ObReaderConfig() { public ObReaderConfig() {
super(Maps.newHashMap()); super(Maps.newHashMap());
} }
/**
* Constructor with a config map.
*
* @param allConfigs Config map.
*/
public ObReaderConfig(Map<String, String> allConfigs) { public ObReaderConfig(Map<String, String> allConfigs) {
super(allConfigs); super(allConfigs);
} }
...@@ -64,7 +94,7 @@ public class ObReaderConfig extends AbstractConnectionConfig { ...@@ -64,7 +94,7 @@ public class ObReaderConfig extends AbstractConnectionConfig {
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
for (Map.Entry<String, ConfigItem<Object>> entry : configs.entrySet()) { for (Map.Entry<String, ConfigItem<Object>> entry : configs.entrySet()) {
String value = entry.getValue().val.toString(); String value = entry.getValue().val.toString();
if (CLUSTER_PASSWORD.key.equals(entry.getKey()) && ShareConf.AUTH_PASSWORD_HASH) { if (CLUSTER_PASSWORD.key.equals(entry.getKey()) && SharedConf.AUTH_PASSWORD_HASH) {
value = Hex.str(CryptoUtil.sha1(value)); value = Hex.str(CryptoUtil.sha1(value));
} }
sb.append(entry.getKey()).append("=").append(value).append(" "); sb.append(entry.getKey()).append("=").append(value).append(" ");
...@@ -88,49 +118,50 @@ public class ObReaderConfig extends AbstractConnectionConfig { ...@@ -88,49 +118,50 @@ public class ObReaderConfig extends AbstractConnectionConfig {
@Override @Override
public String toString() { public String toString() {
return "rootserver_list=" + RS_LIST + ", cluster_user=" + CLUSTER_USER + ", cluster_password=******, " + return "rootserver_list=" + RS_LIST + ", cluster_user=" + CLUSTER_USER + ", cluster_password=******, " +
"tb_white_list=" + TABLE_WHITE_LIST + ", start_timestamp=" + START_TIMESTAMP; "tb_white_list=" + TABLE_WHITE_LIST + ", start_timestamp=" + START_TIMESTAMP;
} }
/** /**
* 设置管控服务列表 * Set root server list.
* *
* @param rsList 管控服务列表 * @param rsList Root server list.
*/ */
public void setRsList(String rsList) { public void setRsList(String rsList) {
RS_LIST.set(rsList); RS_LIST.set(rsList);
} }
/** /**
* 设置连接OB用户名 * Set cluster username
* *
* @param clusterUser 用户名 * @param clusterUser Cluster username.
*/ */
public void setUsername(String clusterUser) { public void setUsername(String clusterUser) {
CLUSTER_USER.set(clusterUser); CLUSTER_USER.set(clusterUser);
} }
/** /**
* 设置连接OB密码 * Set cluster password
* *
* @param clusterPassword 密码 * @param clusterPassword Cluster password.
*/ */
public void setPassword(String clusterPassword) { public void setPassword(String clusterPassword) {
CLUSTER_PASSWORD.set(clusterPassword); CLUSTER_PASSWORD.set(clusterPassword);
} }
/** /**
* 配置过滤规则，由租户.库.表3个维度组成，每一段 * 表示任意，如：A.foo.bar，B.foo.*，C.*.*，*.*.* * Set table whitelist. It is composed of three dimensions: tenant, library, and table.
* Asterisk means any, such as: "A.foo.bar", "B.foo.*", "C.*.*", "*.*.*".
* *
* @param tableWhiteList 监听表的过滤规则 * @param tableWhiteList Table whitelist.
*/ */
public void setTableWhiteList(String tableWhiteList) { public void setTableWhiteList(String tableWhiteList) {
TABLE_WHITE_LIST.set(tableWhiteList); TABLE_WHITE_LIST.set(tableWhiteList);
} }
/** /**
* 设置起始订阅的 UNIX时间戳，0表示从当前，通常不要早于1小时 * Set start timestamp, zero means from now on.
* *
* @param startTimestamp 起始时间戳 * @param startTimestamp Start timestamp.
*/ */
public void setStartTimestamp(Long startTimestamp) { public void setStartTimestamp(Long startTimestamp) {
START_TIMESTAMP.set(startTimestamp); START_TIMESTAMP.set(startTimestamp);
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/ClientHandler.java
浏览文件 @ 1686e1dc
...@@ -11,9 +11,9 @@ See the Mulan PSL v2 for more details. */ ...@@ -11,9 +11,9 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.connection; package com.oceanbase.clogproxy.client.connection;
import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.InvalidProtocolBufferException;
import com.oceanbase.clogproxy.client.config.ClientConf;
import com.oceanbase.clogproxy.client.enums.ErrorCode; import com.oceanbase.clogproxy.client.enums.ErrorCode;
import com.oceanbase.clogproxy.client.exception.LogProxyClientException; import com.oceanbase.clogproxy.client.exception.LogProxyClientException;
import com.oceanbase.clogproxy.client.config.ClientConf;
import com.oceanbase.clogproxy.client.message.LogMessage; import com.oceanbase.clogproxy.client.message.LogMessage;
import com.oceanbase.clogproxy.common.packet.CompressType; import com.oceanbase.clogproxy.common.packet.CompressType;
import com.oceanbase.clogproxy.common.packet.HeaderType; import com.oceanbase.clogproxy.common.packet.HeaderType;
...@@ -35,42 +35,129 @@ import org.slf4j.LoggerFactory; ...@@ -35,42 +35,129 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
/**
* This is an implementation class of {@link ChannelInboundHandlerAdapter}.
*/
public class ClientHandler extends ChannelInboundHandlerAdapter { public class ClientHandler extends ChannelInboundHandlerAdapter {
private static final Logger logger = LoggerFactory.getLogger(ClientHandler.class); private static final Logger logger = LoggerFactory.getLogger(ClientHandler.class);
/**
* Magic string used to request log proxy.
*/
private static final byte[] MAGIC_STRING = new byte[]{'x', 'i', '5', '3', 'g', ']', 'q'}; private static final byte[] MAGIC_STRING = new byte[]{'x', 'i', '5', '3', 'g', ']', 'q'};
/**
* Client ip address.
*/
private static final String CLIENT_IP = NetworkUtil.getLocalIp(); private static final String CLIENT_IP = NetworkUtil.getLocalIp();
private static final int HEAD_LENGTH = 7;
/**
* Length of packet header.
*/
private static final int HEAD_LENGTH = 7;
/**
* A client stream.
*/
private ClientStream stream; private ClientStream stream;
/**
* Connection params.
*/
private ConnectionParams params; private ConnectionParams params;
/**
* Record queue, it's a {@link BlockingQueue} for storing {@link StreamContext.TransferPacket}.
*/
private BlockingQueue<StreamContext.TransferPacket> recordQueue; private BlockingQueue<StreamContext.TransferPacket> recordQueue;
/**
* Handshake type enumeration.
*/
enum HandshakeStateV1 { enum HandshakeStateV1 {
/**
* State of parsing the packet header.
*/
PB_HEAD, PB_HEAD,
/**
* State of handling handshake response.
*/
CLIENT_HANDSHAKE_RESPONSE, CLIENT_HANDSHAKE_RESPONSE,
/**
* State of handling record.
*/
RECORD, RECORD,
/**
* State of handling error response.
*/
ERROR_RESPONSE, ERROR_RESPONSE,
/**
* State of handling runtime status response.
*/
STATUS STATUS
} }
/**
* Handshake state.
*/
private HandshakeStateV1 state = HandshakeStateV1.PB_HEAD; private HandshakeStateV1 state = HandshakeStateV1.PB_HEAD;
/**
* A {@link Cumulator} instance.
*/
private final Cumulator cumulator = ByteToMessageDecoder.MERGE_CUMULATOR; private final Cumulator cumulator = ByteToMessageDecoder.MERGE_CUMULATOR;
/**
* A {@link ByteBuf} used for channel reading.
*/
ByteBuf buffer; ByteBuf buffer;
/**
* A flag of whether channel is active.
*/
private boolean poolFlag = true; private boolean poolFlag = true;
/**
* A flag of whether it is the first part of {@link ByteBuf}.
*/
private boolean first; private boolean first;
/**
* Number of read attempts.
*/
private int numReads = 0; private int numReads = 0;
/**
* A flag of whether the message is not readable.
*/
private boolean dataNotEnough = false; private boolean dataNotEnough = false;
/**
* The length of message body.
*/
private int dataLength = 0; private int dataLength = 0;
/**
* A {@link LZ4Factory} instance.
*/
LZ4Factory factory = LZ4Factory.fastestInstance(); LZ4Factory factory = LZ4Factory.fastestInstance();
/**
* A {@link LZ4FastDecompressor} instance.
*/
LZ4FastDecompressor fastDecompressor = factory.fastDecompressor(); LZ4FastDecompressor fastDecompressor = factory.fastDecompressor();
public ClientHandler() { } /**
* Constructor with empty arguments.
*/
public ClientHandler() {
}
/**
* Reset {@link #state} to {@link HandshakeStateV1#PB_HEAD}.
*/
protected void resetState() { protected void resetState() {
state = HandshakeStateV1.PB_HEAD; state = HandshakeStateV1.PB_HEAD;
} }
...@@ -125,6 +212,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -125,6 +212,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Handle header response.
*/
private void handleHeader() { private void handleHeader() {
if (buffer.readableBytes() >= HEAD_LENGTH) { if (buffer.readableBytes() >= HEAD_LENGTH) {
int version = buffer.readShort(); int version = buffer.readShort();
...@@ -133,13 +223,13 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -133,13 +223,13 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
checkHeader(version, type, dataLength); checkHeader(version, type, dataLength);
HeaderType headerType = HeaderType.codeOf(type); HeaderType headerType = HeaderType.codeOf(type);
if(headerType == HeaderType.HANDSHAKE_RESPONSE_CLIENT) { if (headerType == HeaderType.HANDSHAKE_RESPONSE_CLIENT) {
state = HandshakeStateV1.CLIENT_HANDSHAKE_RESPONSE; state = HandshakeStateV1.CLIENT_HANDSHAKE_RESPONSE;
} else if(headerType == HeaderType.ERROR_RESPONSE) { } else if (headerType == HeaderType.ERROR_RESPONSE) {
state = HandshakeStateV1.ERROR_RESPONSE; state = HandshakeStateV1.ERROR_RESPONSE;
} else if(headerType == HeaderType.DATA_CLIENT) { } else if (headerType == HeaderType.DATA_CLIENT) {
state = HandshakeStateV1.RECORD; state = HandshakeStateV1.RECORD;
} else if(headerType == HeaderType.STATUS) { } else if (headerType == HeaderType.STATUS) {
state = HandshakeStateV1.STATUS; state = HandshakeStateV1.STATUS;
} }
} else { } else {
...@@ -147,8 +237,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -147,8 +237,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Handle handshake response.
*/
private void handleHandshakeResponse() throws InvalidProtocolBufferException { private void handleHandshakeResponse() throws InvalidProtocolBufferException {
if(buffer.readableBytes() >= dataLength) { if (buffer.readableBytes() >= dataLength) {
byte[] bytes = new byte[dataLength]; byte[] bytes = new byte[dataLength];
buffer.readBytes(bytes); buffer.readBytes(bytes);
LogProxyProto.ClientHandshakeResponse response = LogProxyProto.ClientHandshakeResponse.parseFrom(bytes); LogProxyProto.ClientHandshakeResponse response = LogProxyProto.ClientHandshakeResponse.parseFrom(bytes);
...@@ -159,8 +252,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -159,8 +252,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Handle error response.
*/
private void handleErrorResponse() throws InvalidProtocolBufferException { private void handleErrorResponse() throws InvalidProtocolBufferException {
if(buffer.readableBytes() >= dataLength) { if (buffer.readableBytes() >= dataLength) {
byte[] bytes = new byte[dataLength]; byte[] bytes = new byte[dataLength];
buffer.readBytes(bytes); buffer.readBytes(bytes);
LogProxyProto.ErrorResponse response = LogProxyProto.ErrorResponse.parseFrom(bytes); LogProxyProto.ErrorResponse response = LogProxyProto.ErrorResponse.parseFrom(bytes);
...@@ -171,8 +267,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -171,8 +267,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Handle server status response.
*/
private void handleServerStatus() throws InvalidProtocolBufferException { private void handleServerStatus() throws InvalidProtocolBufferException {
if(buffer.readableBytes() >= dataLength) { if (buffer.readableBytes() >= dataLength) {
byte[] bytes = new byte[dataLength]; byte[] bytes = new byte[dataLength];
buffer.readBytes(bytes); buffer.readBytes(bytes);
LogProxyProto.RuntimeStatus response = LogProxyProto.RuntimeStatus.parseFrom(bytes); LogProxyProto.RuntimeStatus response = LogProxyProto.RuntimeStatus.parseFrom(bytes);
...@@ -183,8 +282,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -183,8 +282,11 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Handle record data response.
*/
private void handleRecord() { private void handleRecord() {
if(buffer.readableBytes() >= dataLength) { if (buffer.readableBytes() >= dataLength) {
parseDataNew(); parseDataNew();
state = HandshakeStateV1.PB_HEAD; state = HandshakeStateV1.PB_HEAD;
} else { } else {
...@@ -192,6 +294,13 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -192,6 +294,13 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Check if the header is valid.
*
* @param version Protocol version.
* @param type Header type.
* @param length Data length.
*/
private void checkHeader(int version, int type, int length) { private void checkHeader(int version, int type, int length) {
if (ProtocolVersion.codeOf(version) == null) { if (ProtocolVersion.codeOf(version) == null) {
logger.error("unsupported protocol version: {}", version); logger.error("unsupported protocol version: {}", version);
...@@ -207,6 +316,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -207,6 +316,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Do parse record data from buffer. It will firstly decompress the raw data if necessary.
*/
private void parseDataNew() { private void parseDataNew() {
try { try {
byte[] buff = new byte[dataLength]; byte[] buff = new byte[dataLength];
...@@ -221,7 +333,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -221,7 +333,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
int decompress = fastDecompressor.decompress(rawData, 0, bytes, 0, compressedLen); int decompress = fastDecompressor.decompress(rawData, 0, bytes, 0, compressedLen);
if (decompress != rawLen) { if (decompress != rawLen) {
throw new LogProxyClientException(ErrorCode.E_LEN, "decompressed length [" + decompress throw new LogProxyClientException(ErrorCode.E_LEN, "decompressed length [" + decompress
+ "] is not expected [" + rawLen + "]"); + "] is not expected [" + rawLen + "]");
} }
parseRecord(bytes); parseRecord(bytes);
} else { } else {
...@@ -232,26 +344,30 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -232,26 +344,30 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Do parse record data from an array of bytes to a {@link LogMessage} and add it into {@link #recordQueue}.
*
* @param bytes An array of bytes of record data.
* @throws LogProxyClientException If exception occurs.
*/
private void parseRecord(byte[] bytes) throws LogProxyClientException { private void parseRecord(byte[] bytes) throws LogProxyClientException {
int offset = 0; int offset = 0;
while (offset < bytes.length) { while (offset < bytes.length) {
int dataLength = Conversion.byteArrayToInt(bytes, offset + 4, 0, 0, 4); int dataLength = Conversion.byteArrayToInt(bytes, offset + 4, 0, 0, 4);
LogMessage drcRecord; LogMessage logMessage;
try { try {
/* /*
* We must copy a byte array and call parse after then, * We must copy a byte array and call parse after then,
* or got a !!!RIDICULOUS EXCEPTION!!!, * or got a !!!RIDICULOUS EXCEPTION!!!,
* if we wrap a upooled buffer with offset and call setByteBuf just as same as `parse` function do. * if we wrap an unpooled buffer with offset and call setByteBuf just as same as `parse` function do.
*/ */
drcRecord = new LogMessage(false); logMessage = new LogMessage(false);
byte[] data = new byte[dataLength + 8]; byte[] data = new byte[dataLength + 8];
System.arraycopy(bytes, offset, data, 0, data.length); System.arraycopy(bytes, offset, data, 0, data.length);
drcRecord.parse(data); logMessage.parse(data);
if (ClientConf.IGNORE_UNKNOWN_RECORD_TYPE) { if (ClientConf.IGNORE_UNKNOWN_RECORD_TYPE) {
// unsupported type, ignore // unsupported type, ignore
logger.debug("Unsupported record type: {}", drcRecord); logger.debug("Unsupported record type: {}", logMessage);
offset += (8 + dataLength); offset += (8 + dataLength);
continue; continue;
} }
...@@ -262,7 +378,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -262,7 +378,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
while (true) { while (true) {
try { try {
recordQueue.put(new StreamContext.TransferPacket(drcRecord)); recordQueue.put(new StreamContext.TransferPacket(logMessage));
break; break;
} catch (InterruptedException e) { } catch (InterruptedException e) {
// do nothing // do nothing
...@@ -273,6 +389,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -273,6 +389,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
} }
} }
/**
* Discard the bytes in buffer.
*/
protected final void discardSomeReadBytes() { protected final void discardSomeReadBytes() {
if (buffer != null && !first && buffer.refCnt() == 1) { if (buffer != null && !first && buffer.refCnt() == 1) {
// discard some bytes if possible to make more room in the // discard some bytes if possible to make more room in the
...@@ -299,9 +418,14 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -299,9 +418,14 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
ctx.channel().writeAndFlush(generateConnectRequest(params.getProtocolVersion())); ctx.channel().writeAndFlush(generateConnectRequest(params.getProtocolVersion()));
} }
/**
* Generate the request body for protocol v2.
*
* @return Request body.
*/
public ByteBuf generateConnectRequestV2() { public ByteBuf generateConnectRequestV2() {
LogProxyProto.ClientHandshakeRequest handShake = LogProxyProto.ClientHandshakeRequest.newBuilder(). LogProxyProto.ClientHandshakeRequest handShake = LogProxyProto.ClientHandshakeRequest.newBuilder().
setLogType(params.getLogType().getCode()). setLogType(params.getLogType().code()).
setIp(CLIENT_IP). setIp(CLIENT_IP).
setId(params.getClientId()). setId(params.getClientId()).
setVersion(ClientConf.VERSION). setVersion(ClientConf.VERSION).
...@@ -319,6 +443,12 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -319,6 +443,12 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
return byteBuf; return byteBuf;
} }
/**
* Generate the request body.
*
* @param version Protocol version.
* @return Request body.
*/
public ByteBuf generateConnectRequest(ProtocolVersion version) { public ByteBuf generateConnectRequest(ProtocolVersion version) {
if (version == ProtocolVersion.V2) { if (version == ProtocolVersion.V2) {
return generateConnectRequestV2(); return generateConnectRequestV2();
...@@ -331,7 +461,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter { ...@@ -331,7 +461,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
byteBuf.capacity(byteBuf.capacity() + 2 + 4 + 1); byteBuf.capacity(byteBuf.capacity() + 2 + 4 + 1);
byteBuf.writeShort(ProtocolVersion.V0.code()); byteBuf.writeShort(ProtocolVersion.V0.code());
byteBuf.writeInt(HeaderType.HANDSHAKE_REQUEST_CLIENT.code()); byteBuf.writeInt(HeaderType.HANDSHAKE_REQUEST_CLIENT.code());
byteBuf.writeByte(params.getLogType().getCode()); byteBuf.writeByte(params.getLogType().code());
// body // body
int length = CLIENT_IP.length(); int length = CLIENT_IP.length();
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/ClientStream.java
浏览文件 @ 1686e1dc
...@@ -26,27 +26,64 @@ import java.util.List; ...@@ -26,27 +26,64 @@ import java.util.List;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
/**
* This class represents a stream of log client. Stream means a channel of log transmission here.
*/
public class ClientStream { public class ClientStream {
private static final Logger logger = LoggerFactory.getLogger(ClientStream.class); private static final Logger logger = LoggerFactory.getLogger(ClientStream.class);
// routine /**
* Flag of whether the stream is started.
*/
private final AtomicBoolean started = new AtomicBoolean(false); private final AtomicBoolean started = new AtomicBoolean(false);
/**
* The process thread.
*/
private Thread thread = null; private Thread thread = null;
// status /**
* Context of stream.
*/
private StreamContext context = null; private StreamContext context = null;
/**
* Checkpoint string used to resume writing into the queue.
*/
private String checkpointString; private String checkpointString;
// reconnection /**
* Number of reconnections
*/
private int retryTimes = 0; private int retryTimes = 0;
/**
* Connection to log proxy with netty channel.
*/
private Connection connection = null; private Connection connection = null;
/**
* Flag of whether the stream is reconnecting now.
*/
private final AtomicBoolean reconnecting = new AtomicBoolean(true); private final AtomicBoolean reconnecting = new AtomicBoolean(true);
/**
* Flag of whether the stream need reconnect.
*/
private final AtomicBoolean reconnect = new AtomicBoolean(true); private final AtomicBoolean reconnect = new AtomicBoolean(true);
// user callbacks /**
* The list of {@link RecordListener}.
*/
private final List<RecordListener> listeners = new ArrayList<>(); private final List<RecordListener> listeners = new ArrayList<>();
/**
* The list of {@link StatusListener}
*/
private final List<StatusListener> statusListeners = new ArrayList<>(); private final List<StatusListener> statusListeners = new ArrayList<>();
/**
* Reconnect state type enumeration.
*/
private enum ReconnectState { private enum ReconnectState {
/** /**
* success * success
...@@ -62,10 +99,19 @@ public class ClientStream { ...@@ -62,10 +99,19 @@ public class ClientStream {
EXIT; EXIT;
} }
/**
* Sole constructor.
*
* @param connectionParams Connection params.
* @param sslContext A {@link SslContext} for encrypted communication.
*/
public ClientStream(ConnectionParams connectionParams, SslContext sslContext) { public ClientStream(ConnectionParams connectionParams, SslContext sslContext) {
context = new StreamContext(this, connectionParams, sslContext); context = new StreamContext(this, connectionParams, sslContext);
} }
/**
* Close and wait the connection.
*/
public void stop() { public void stop() {
if (!started.compareAndSet(true, false)) { if (!started.compareAndSet(true, false)) {
logger.info("stopping LogProxy Client...."); logger.info("stopping LogProxy Client....");
...@@ -81,6 +127,9 @@ public class ClientStream { ...@@ -81,6 +127,9 @@ public class ClientStream {
logger.info("stopped LogProxy Client"); logger.info("stopped LogProxy Client");
} }
/**
* Call {@link Thread#join()} method of process thread.
*/
public void join() { public void join() {
if (thread != null) { if (thread != null) {
try { try {
...@@ -91,10 +140,18 @@ public class ClientStream { ...@@ -91,10 +140,18 @@ public class ClientStream {
} }
} }
/**
* Call {@link #stop()} asynchronously.
*/
public void triggerStop() { public void triggerStop() {
new Thread(this::stop).start(); new Thread(this::stop).start();
} }
/**
* Call {@link RecordListener#onException(LogProxyClientException)} asynchronously.
*
* @param e An exception.
*/
public void triggerException(LogProxyClientException e) { public void triggerException(LogProxyClientException e) {
// use thread make sure non-blocking // use thread make sure non-blocking
new Thread(() -> { new Thread(() -> {
...@@ -104,6 +161,9 @@ public class ClientStream { ...@@ -104,6 +161,9 @@ public class ClientStream {
}).start(); }).start();
} }
/**
* Start the process thread.
*/
public void start() { public void start() {
// if status listener exist, enable monitor // if status listener exist, enable monitor
context.params.setEnableMonitor(CollectionUtils.isNotEmpty(statusListeners)); context.params.setEnableMonitor(CollectionUtils.isNotEmpty(statusListeners));
...@@ -182,10 +242,20 @@ public class ClientStream { ...@@ -182,10 +242,20 @@ public class ClientStream {
} }
} }
/**
* Get the flag of whether the stream is started.
*
* @return The flag of whether the stream is started.
*/
public boolean isRunning() { public boolean isRunning() {
return started.get(); return started.get();
} }
/**
* Reconnect to log proxy. It is also used for first time connecting.
*
* @return A {@link ReconnectState}.
*/
private ReconnectState reconnect() { private ReconnectState reconnect() {
// reconnect flag mark, tiny load for checking // reconnect flag mark, tiny load for checking
if (reconnect.compareAndSet(true, false)) { if (reconnect.compareAndSet(true, false)) {
...@@ -233,6 +303,9 @@ public class ClientStream { ...@@ -233,6 +303,9 @@ public class ClientStream {
return ReconnectState.SUCCESS; return ReconnectState.SUCCESS;
} }
/**
* Reset the flags for reconnection.
*/
public void triggerReconnect() { public void triggerReconnect() {
// reconnection action guard, avoid concurrent or multiple invoke // reconnection action guard, avoid concurrent or multiple invoke
if (reconnecting.compareAndSet(false, true)) { if (reconnecting.compareAndSet(false, true)) {
...@@ -240,10 +313,20 @@ public class ClientStream { ...@@ -240,10 +313,20 @@ public class ClientStream {
} }
} }
/**
* Add a {@link RecordListener} to {@link #listeners}.
*
* @param recordListener A {@link RecordListener}.
*/
public synchronized void addListener(RecordListener recordListener) { public synchronized void addListener(RecordListener recordListener) {
listeners.add(recordListener); listeners.add(recordListener);
} }
/**
* Add a {@link StatusListener} to {@link #statusListeners}.
*
* @param statusListener A {@link StatusListener}.
*/
public synchronized void addStatusListener(StatusListener statusListener) { public synchronized void addStatusListener(StatusListener statusListener) {
statusListeners.add(statusListener); statusListeners.add(statusListener);
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/Connection.java
浏览文件 @ 1686e1dc
...@@ -18,18 +18,35 @@ import org.slf4j.LoggerFactory; ...@@ -18,18 +18,35 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
/**
* This class represents a connection which contains a netty channel.
*/
public class Connection { public class Connection {
private static final Logger logger = LoggerFactory.getLogger(Connection.class); private static final Logger logger = LoggerFactory.getLogger(Connection.class);
/**
* A netty channel.
*/
private Channel channel; private Channel channel;
/**
* A flag of whether the channel is closed.
*/
private final AtomicBoolean closed = new AtomicBoolean(false); private final AtomicBoolean closed = new AtomicBoolean(false);
/**
* Sole constructor.
*
* @param channel A netty channel.
*/
public Connection(Channel channel) { public Connection(Channel channel) {
this.channel = channel; this.channel = channel;
} }
/**
* Close this connection.
*/
public void close() { public void close() {
if (!closed.compareAndSet(false, true)) { if (!closed.compareAndSet(false, true)) {
logger.warn("connection already closed"); logger.warn("connection already closed");
...@@ -40,13 +57,19 @@ public class Connection { ...@@ -40,13 +57,19 @@ public class Connection {
channel.close().addListener(this::logCloseResult).syncUninterruptibly(); channel.close().addListener(this::logCloseResult).syncUninterruptibly();
} catch (Exception e) { } catch (Exception e) {
logger.warn("close connection to remote address {} exception", logger.warn("close connection to remote address {} exception",
NetworkUtil.parseRemoteAddress(channel), e); NetworkUtil.parseRemoteAddress(channel), e);
} }
} }
channel = null; channel = null;
} }
} }
/**
* A callback that will logging the result of {@link Channel#close()}.
*
* @param future The source {@link Future} which called this callback.
*/
@SuppressWarnings("rawtypes")
private void logCloseResult(Future future) { private void logCloseResult(Future future) {
if (future.isSuccess()) { if (future.isSuccess()) {
if (logger.isInfoEnabled()) { if (logger.isInfoEnabled()) {
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/ConnectionFactory.java
浏览文件 @ 1686e1dc
...@@ -25,31 +25,60 @@ import io.netty.util.AttributeKey; ...@@ -25,31 +25,60 @@ import io.netty.util.AttributeKey;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
/**
* This is a factory class of {@link Connection}.
*/
public class ConnectionFactory { public class ConnectionFactory {
/**
* A static class that holds the singleton instance of {@link ConnectionFactory}.
*/
private static class Singleton { private static class Singleton {
/**
* The singleton instance of {@link ConnectionFactory}.
*/
private static final ConnectionFactory INSTANCE = new ConnectionFactory(); private static final ConnectionFactory INSTANCE = new ConnectionFactory();
} }
/**
* Get the singleton instance of {@link ConnectionFactory}.
*
* @return The singleton instance of {@link ConnectionFactory}.
*/
public static ConnectionFactory instance() { public static ConnectionFactory instance() {
return Singleton.INSTANCE; return Singleton.INSTANCE;
} }
/**
* Sole constructor. It can only be used in {@link Singleton} class.
*/
private ConnectionFactory() { private ConnectionFactory() {
} }
/**
* Context key.
*/
public static final AttributeKey<StreamContext> CONTEXT_KEY = AttributeKey.valueOf("context"); public static final AttributeKey<StreamContext> CONTEXT_KEY = AttributeKey.valueOf("context");
/**
* Worker group in type of {@link EventLoopGroup}.
*/
private static final EventLoopGroup WORKER_GROUP = NettyEventLoopUtil.newEventLoopGroup(1, private static final EventLoopGroup WORKER_GROUP = NettyEventLoopUtil.newEventLoopGroup(1,
new NamedThreadFactory("log-proxy-client-worker", true)); new NamedThreadFactory("log-proxy-client-worker", true));
/**
* Create a {@link Bootstrap} instance.
*
* @param sslContext The {@link SslContext} used for encrypted communication.
* @return A {@link Bootstrap} instance.
*/
private Bootstrap initBootstrap(SslContext sslContext) { private Bootstrap initBootstrap(SslContext sslContext) {
Bootstrap bootstrap = new Bootstrap(); Bootstrap bootstrap = new Bootstrap();
bootstrap.group(WORKER_GROUP) bootstrap.group(WORKER_GROUP)
.channel(NettyEventLoopUtil.getClientSocketChannelClass()) .channel(NettyEventLoopUtil.getClientSocketChannelClass())
.option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_KEEPALIVE, true); .option(ChannelOption.SO_KEEPALIVE, true);
bootstrap.handler(new ChannelInitializer<SocketChannel>() { bootstrap.handler(new ChannelInitializer<SocketChannel>() {
...@@ -65,6 +94,13 @@ public class ConnectionFactory { ...@@ -65,6 +94,13 @@ public class ConnectionFactory {
return bootstrap; return bootstrap;
} }
/**
* Create a {@link Connection} with specific {@link StreamContext}.
*
* @param context Stream context.
* @return A {@link Connection}.
* @throws LogProxyClientException If exception occurs.
*/
public Connection createConnection(StreamContext context) throws LogProxyClientException { public Connection createConnection(StreamContext context) throws LogProxyClientException {
Bootstrap bootstrap = initBootstrap(context.getSslContext()); Bootstrap bootstrap = initBootstrap(context.getSslContext());
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, ClientConf.CONNECT_TIMEOUT_MS); bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, ClientConf.CONNECT_TIMEOUT_MS);
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/ConnectionParams.java
浏览文件 @ 1686e1dc
...@@ -16,19 +16,60 @@ import com.oceanbase.clogproxy.common.packet.ProtocolVersion; ...@@ -16,19 +16,60 @@ import com.oceanbase.clogproxy.common.packet.ProtocolVersion;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder; import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle; import org.apache.commons.lang3.builder.ToStringStyle;
/**
* This is a configuration class of connection parameters.
*/
public class ConnectionParams { public class ConnectionParams {
/**
* Log type.
*/
private final LogType logType; private final LogType logType;
/**
* Client id.
*/
private final String clientId; private final String clientId;
/**
* Log proxy host.
*/
private final String host; private final String host;
/**
* Log proxy port.
*/
private final int port; private final int port;
/**
* Connection config.
*/
private final ConnectionConfig connectionConfig; private final ConnectionConfig connectionConfig;
/**
* Generated configuration string.
*/
private String configurationString; private String configurationString;
/**
* Protocol version.
*/
private ProtocolVersion protocolVersion; private ProtocolVersion protocolVersion;
/**
* Flag of whether enable monitor.
*/
private boolean enableMonitor; private boolean enableMonitor;
/**
* Constructor.
*
* @param logType Log type.
* @param clientId Client id.
* @param host Log proxy host.
* @param port Log proxy port.
* @param connectionConfig Connection config.
*/
public ConnectionParams(LogType logType, String clientId, String host, int port, ConnectionConfig connectionConfig) { public ConnectionParams(LogType logType, String clientId, String host, int port, ConnectionConfig connectionConfig) {
this.logType = logType; this.logType = logType;
this.clientId = clientId; this.clientId = clientId;
...@@ -38,6 +79,11 @@ public class ConnectionParams { ...@@ -38,6 +79,11 @@ public class ConnectionParams {
this.configurationString = connectionConfig.generateConfigurationString(); this.configurationString = connectionConfig.generateConfigurationString();
} }
/**
* Update checkpoint in connection config.
*
* @param checkpoint Checkpoint of the last record put in queue.
*/
public void updateCheckpoint(String checkpoint) { public void updateCheckpoint(String checkpoint) {
connectionConfig.updateCheckpoint(checkpoint); connectionConfig.updateCheckpoint(checkpoint);
configurationString = connectionConfig.generateConfigurationString(); configurationString = connectionConfig.generateConfigurationString();
...@@ -48,42 +94,92 @@ public class ConnectionParams { ...@@ -48,42 +94,92 @@ public class ConnectionParams {
return ReflectionToStringBuilder.toString(this, ToStringStyle.SHORT_PREFIX_STYLE); return ReflectionToStringBuilder.toString(this, ToStringStyle.SHORT_PREFIX_STYLE);
} }
/**
* Get the basic info of connection, which contains client id and connection config.
*
* @return A string of client id and connection config.
*/
public String info() { public String info() {
return clientId + ": " + connectionConfig.toString(); return clientId + ": " + connectionConfig.toString();
} }
/**
* Get the log type.
*
* @return The log type.
*/
public LogType getLogType() { public LogType getLogType() {
return logType; return logType;
} }
/**
* Get the client id.
*
* @return The client id.
*/
public String getClientId() { public String getClientId() {
return clientId; return clientId;
} }
/**
* Get the host of log proxy.
*
* @return The host of log proxy.
*/
public String getHost() { public String getHost() {
return host; return host;
} }
/**
* Get the port of log proxy.
*
* @return The port of log proxy.
*/
public int getPort() { public int getPort() {
return port; return port;
} }
/**
* Get the generated configuration string.
*
* @return The configuration string.
*/
public String getConfigurationString() { public String getConfigurationString() {
return configurationString; return configurationString;
} }
/**
* Get the protocol version.
*
* @return Protocol version.
*/
public ProtocolVersion getProtocolVersion() { public ProtocolVersion getProtocolVersion() {
return protocolVersion; return protocolVersion;
} }
/**
* Set the protocol version.
*
* @param protocolVersion Protocol version.
*/
public void setProtocolVersion(ProtocolVersion protocolVersion) { public void setProtocolVersion(ProtocolVersion protocolVersion) {
this.protocolVersion = protocolVersion; this.protocolVersion = protocolVersion;
} }
/**
* Get the flag of whether enable monitor.
*
* @return The flag of whether enable monitor.
*/
public boolean isEnableMonitor() { public boolean isEnableMonitor() {
return enableMonitor; return enableMonitor;
} }
/**
* Set the flag of whether enable monitor.
*
* @param enableMonitor The flag of whether enable monitor.
*/
public void setEnableMonitor(boolean enableMonitor) { public void setEnableMonitor(boolean enableMonitor) {
this.enableMonitor = enableMonitor; this.enableMonitor = enableMonitor;
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/NamedThreadFactory.java
浏览文件 @ 1686e1dc
...@@ -13,22 +13,54 @@ package com.oceanbase.clogproxy.client.connection; ...@@ -13,22 +13,54 @@ package com.oceanbase.clogproxy.client.connection;
import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
/**
* This is a factory class for {@link ThreadFactory}.
*/
public class NamedThreadFactory implements ThreadFactory { public class NamedThreadFactory implements ThreadFactory {
/**
* Pool number.
*/
private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1); private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
private final AtomicInteger threadNumber = new AtomicInteger(1); /**
private final ThreadGroup group; * Thread number.
private final String namePrefix; */
private final boolean isDaemon; private final AtomicInteger threadNumber = new AtomicInteger(1);
/**
* Thread group.
*/
private final ThreadGroup group;
/**
* Prefix of thread name.
*/
private final String namePrefix;
/**
* Flag of whether the thread is daemon.
*/
private final boolean isDaemon;
/**
* Constructor with no arguments. It will take "ThreadPool" as its name.
*/
public NamedThreadFactory() { public NamedThreadFactory() {
this("ThreadPool"); this("ThreadPool");
} }
/**
* Constructor with name.
*
* @param name Name of thread factory.
*/
public NamedThreadFactory(String name) { public NamedThreadFactory(String name) {
this(name, false); this(name, false);
} }
/**
* Constructor with name prefix and daemon flag.
*
* @param prefix Name prefix of thread factory.
* @param daemon A flag of whether starting the thread on daemon mode.
*/
public NamedThreadFactory(String prefix, boolean daemon) { public NamedThreadFactory(String prefix, boolean daemon) {
SecurityManager s = System.getSecurityManager(); SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
...@@ -36,11 +68,6 @@ public class NamedThreadFactory implements ThreadFactory { ...@@ -36,11 +68,6 @@ public class NamedThreadFactory implements ThreadFactory {
isDaemon = daemon; isDaemon = daemon;
} }
/**
* Create a thread.
*
* @see ThreadFactory#newThread(Runnable)
*/
@Override @Override
public Thread newThread(Runnable r) { public Thread newThread(Runnable r) {
Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0); Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/NettyEventLoopUtil.java
浏览文件 @ 1686e1dc
...@@ -20,27 +20,34 @@ import io.netty.channel.socket.nio.NioSocketChannel; ...@@ -20,27 +20,34 @@ import io.netty.channel.socket.nio.NioSocketChannel;
import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadFactory;
/**
* This is a factory class of netty event loop.
*/
public class NettyEventLoopUtil { public class NettyEventLoopUtil {
/** check whether epoll enabled, and it would not be changed during runtime. */ /**
private static boolean epollEnabled = Epoll.isAvailable(); * Flag of whether epoll is enabled.
*/
private static final boolean EPOLL_ENABLED = Epoll.isAvailable();
/** /**
* Create the right event loop according to current platform and system property, fallback to NIO when epoll not enabled. * Create the right event loop according to current platform and system property, fallback to NIO when epoll not enabled.
* *
* @param nThreads number of threads * @param nThreads Number of threads.
* @param threadFactory ThreadFactory * @param threadFactory ThreadFactory instance.
* @return an EventLoopGroup suitable for the current platform * @return An EventLoopGroup suitable for the current platform.
*/ */
public static EventLoopGroup newEventLoopGroup(int nThreads, ThreadFactory threadFactory) { public static EventLoopGroup newEventLoopGroup(int nThreads, ThreadFactory threadFactory) {
return epollEnabled ? new EpollEventLoopGroup(nThreads, threadFactory) return EPOLL_ENABLED ? new EpollEventLoopGroup(nThreads, threadFactory)
: new NioEventLoopGroup(nThreads, threadFactory); : new NioEventLoopGroup(nThreads, threadFactory);
} }
/** /**
* @return a SocketChannel class suitable for the given EventLoopGroup implementation * Get the suitable {@link SocketChannel} for the given EventLoopGroup implementation.
*
* @return A {@link SocketChannel} class suitable for the given EventLoopGroup implementation.
*/ */
public static Class<? extends SocketChannel> getClientSocketChannelClass() { public static Class<? extends SocketChannel> getClientSocketChannelClass() {
return epollEnabled ? EpollSocketChannel.class : NioSocketChannel.class; return EPOLL_ENABLED ? EpollSocketChannel.class : NioSocketChannel.class;
} }
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/connection/StreamContext.java
浏览文件 @ 1686e1dc
...@@ -20,59 +20,140 @@ import java.util.concurrent.LinkedBlockingQueue; ...@@ -20,59 +20,140 @@ import java.util.concurrent.LinkedBlockingQueue;
import static com.oceanbase.clogproxy.common.packet.protocol.LogProxyProto.RuntimeStatus; import static com.oceanbase.clogproxy.common.packet.protocol.LogProxyProto.RuntimeStatus;
/**
* This class represents the context of client stream.
*/
public class StreamContext { public class StreamContext {
public static class TransferPacket { public static class TransferPacket {
private HeaderType type; /**
* Packet header type.
*/
private final HeaderType type;
/**
* Log message record.
*/
private LogMessage record; private LogMessage record;
/**
* Log proxy runtime status.
*/
private RuntimeStatus status; private RuntimeStatus status;
/**
* Constructor with a {@link LogMessage}.
*
* @param record A {@link LogMessage}.
*/
public TransferPacket(LogMessage record) { public TransferPacket(LogMessage record) {
this.type = HeaderType.DATA_CLIENT; this.type = HeaderType.DATA_CLIENT;
this.record = record; this.record = record;
} }
/**
* Constructor with a {@link RuntimeStatus}.
*
* @param status A {@link RuntimeStatus}.
*/
public TransferPacket(RuntimeStatus status) { public TransferPacket(RuntimeStatus status) {
this.type = HeaderType.STATUS; this.type = HeaderType.STATUS;
this.status = status; this.status = status;
} }
/**
* Get header type.
*
* @return Packet header type.
*/
public HeaderType getType() { public HeaderType getType() {
return type; return type;
} }
/**
* Get the log message record.
*
* @return Log message record.
*/
public LogMessage getRecord() { public LogMessage getRecord() {
return record; return record;
} }
/**
* Get the log proxy runtime status.
*
* @return Log proxy runtime status.
*/
public RuntimeStatus getStatus() { public RuntimeStatus getStatus() {
return status; return status;
} }
} }
/**
* Blocking queue which stores {@link TransferPacket}.
*/
private final BlockingQueue<TransferPacket> recordQueue = new LinkedBlockingQueue<>(ClientConf.TRANSFER_QUEUE_SIZE); private final BlockingQueue<TransferPacket> recordQueue = new LinkedBlockingQueue<>(ClientConf.TRANSFER_QUEUE_SIZE);
/**
* Client stream.
*/
private final ClientStream stream; private final ClientStream stream;
/**
* Connection params.
*/
ConnectionParams params; ConnectionParams params;
/**
* Netty ssl context.
*
* @see SslContext
*/
private final SslContext sslContext; private final SslContext sslContext;
/**
* Constructor of StreamContext.
*
* @param stream Client stream.
* @param params Connection params.
* @param sslContext Netty ssl context.
*/
public StreamContext(ClientStream stream, ConnectionParams params, SslContext sslContext) { public StreamContext(ClientStream stream, ConnectionParams params, SslContext sslContext) {
this.stream = stream; this.stream = stream;
this.params = params; this.params = params;
this.sslContext = sslContext; this.sslContext = sslContext;
} }
/**
* Get connection params.
*
* @return Connection params.
*/
public ConnectionParams getParams() { public ConnectionParams getParams() {
return params; return params;
} }
/**
* Get netty ssl context.
*
* @return Netty ssl context.
*/
public SslContext getSslContext() { public SslContext getSslContext() {
return sslContext; return sslContext;
} }
/**
* Get the client stream.
*
* @return Client stream.
*/
public ClientStream stream() { public ClientStream stream() {
return stream; return stream;
} }
/**
* Get the record queue.
*
* @return Record queue.
*/
public BlockingQueue<TransferPacket> recordQueue() { public BlockingQueue<TransferPacket> recordQueue() {
return recordQueue; return recordQueue;
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/constants/DataType.java
浏览文件 @ 1686e1dc
...@@ -10,6 +10,9 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,6 +10,9 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.constants; package com.oceanbase.clogproxy.client.constants;
/**
* The class that defines the constants that are used to identify data types.
*/
public class DataType { public class DataType {
public static final byte DT_UNKNOWN = 0x00; public static final byte DT_UNKNOWN = 0x00;
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/enums/DBType.java
浏览文件 @ 1686e1dc
...@@ -10,6 +10,9 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,6 +10,9 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.enums; package com.oceanbase.clogproxy.client.enums;
/**
* Database type enumeration.
*/
public enum DBType { public enum DBType {
MYSQL, MYSQL,
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/enums/ErrorCode.java
浏览文件 @ 1686e1dc
...@@ -10,66 +10,77 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,66 +10,77 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.enums; package com.oceanbase.clogproxy.client.enums;
/**
* Error code enumeration.
*/
public enum ErrorCode { public enum ErrorCode {
////////// 0~499: process error //////////// ////////// 0~499: process error ////////////
/** /**
* general error * General error.
*/ */
NONE(0), NONE(0),
/** /**
* inner error * Inner error
*/ */
E_INNER(1), E_INNER(1),
/** /**
* failed to connect * Failed to connect.
*/ */
E_CONNECT(2), E_CONNECT(2),
/** /**
* exceed max retry connect count * Exceed max retry connect count.
*/ */
E_MAX_RECONNECT(3), E_MAX_RECONNECT(3),
/** /**
* user callback throws exception * User callback throws exception.
*/ */
E_USER(4), E_USER(4),
////////// 500~: receive data error //////////// ////////// 500~: receive data error ////////////
/** /**
* unknown data protocol * Unknown data protocol.
*/ */
E_PROTOCOL(500), E_PROTOCOL(500),
/** /**
* unknown header type * Unknown header type.
*/ */
E_HEADER_TYPE(501), E_HEADER_TYPE(501),
/** /**
* failed to auth * Failed to auth.
*/ */
NO_AUTH(502), NO_AUTH(502),
/** /**
* unknown compress type * Unknown compress type.
*/ */
E_COMPRESS_TYPE(503), E_COMPRESS_TYPE(503),
/** /**
* length not match * Length not match.
*/ */
E_LEN(504), E_LEN(504),
/** /**
* failed to parse data * Failed to parse data.
*/ */
E_PARSE(505); E_PARSE(505);
/**
* The ordinal of this enumeration constant.
*/
int code; int code;
/**
* Constructor.
*
* @param code The ordinal of this enumeration constant.
*/
ErrorCode(int code) { ErrorCode(int code) {
this.code = code; this.code = code;
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/exception/DRCClientException.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.exception;
public class DRCClientException extends Exception {
private static final long serialVersionUID = 1L;
public DRCClientException() {
}
public DRCClientException(final String message) {
super(message);
}
}
\ No newline at end of file
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/exception/DRCClientRunTimeException.java → logproxy-client/src/main/java/com/oceanbase/clogproxy/client/exception/LogMessageException.java
浏览文件 @ 1686e1dc
...@@ -10,20 +10,43 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,20 +10,43 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.exception; package com.oceanbase.clogproxy.client.exception;
public class DRCClientRunTimeException extends RuntimeException { /**
public DRCClientRunTimeException(String errMessage) { * This is a subclasses of {@link RuntimeException} primarily used in process of parsing {@link com.oceanbase.clogproxy.client.message.LogMessage}.
*/
public class LogMessageException extends RuntimeException {
/**
* Constructor with message.
*
* @param errMessage Error message.
*/
public LogMessageException(String errMessage) {
super(errMessage); super(errMessage);
} }
public DRCClientRunTimeException() { /**
* Constructor with no arguments.
*/
public LogMessageException() {
super(); super();
} }
public DRCClientRunTimeException(Throwable cause) { /**
* Constructor with cause.
*
* @param cause Cause error or exception.
*/
public LogMessageException(Throwable cause) {
super(cause); super(cause);
} }
public DRCClientRunTimeException(String errMessage, Throwable cause) { /**
* Constructor with error message and cause.
*
* @param errMessage Error message.
* @param cause Cause error or exception.
*/
public LogMessageException(String errMessage, Throwable cause) {
super(errMessage, cause); super(errMessage, cause);
} }
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/exception/LogProxyClientException.java
浏览文件 @ 1686e1dc
...@@ -12,31 +12,67 @@ package com.oceanbase.clogproxy.client.exception; ...@@ -12,31 +12,67 @@ package com.oceanbase.clogproxy.client.exception;
import com.oceanbase.clogproxy.client.enums.ErrorCode; import com.oceanbase.clogproxy.client.enums.ErrorCode;
/**
* This is a subclasses of {@link RuntimeException} used to indicate the exception occurs in client with an error code.
*/
public class LogProxyClientException extends RuntimeException { public class LogProxyClientException extends RuntimeException {
/**
* Error code.
*/
private ErrorCode code = ErrorCode.NONE; private ErrorCode code = ErrorCode.NONE;
/**
* Constructor with error code and message.
*
* @param code Error code.
* @param message Error message.
*/
public LogProxyClientException(ErrorCode code, String message) { public LogProxyClientException(ErrorCode code, String message) {
super(message); super(message);
this.code = code; this.code = code;
} }
/**
* Constructor with error code and exception.
*
* @param code Error code.
* @param exception Exception instance.
*/
public LogProxyClientException(ErrorCode code, Exception exception) { public LogProxyClientException(ErrorCode code, Exception exception) {
super(exception.getMessage(), exception.getCause()); super(exception.getMessage(), exception.getCause());
this.code = code; this.code = code;
} }
/**
* Constructor with error code, error message and cause.
*
* @param code Error code.
* @param message Error message.
* @param throwable Cause.
*/
public LogProxyClientException(ErrorCode code, String message, Throwable throwable) { public LogProxyClientException(ErrorCode code, String message, Throwable throwable) {
super(message, throwable); super(message, throwable);
this.code = code; this.code = code;
} }
/**
* Identify whether the client should stop the stream.
*
* @return The flag of whether the client should stop the stream.
*/
public boolean needStop() { public boolean needStop() {
return (code == ErrorCode.E_MAX_RECONNECT) || (code == ErrorCode.E_PROTOCOL) || return (code == ErrorCode.E_MAX_RECONNECT) || (code == ErrorCode.E_PROTOCOL) ||
(code == ErrorCode.E_HEADER_TYPE) || (code == ErrorCode.NO_AUTH) || (code == ErrorCode.E_HEADER_TYPE) || (code == ErrorCode.NO_AUTH) ||
(code == ErrorCode.E_COMPRESS_TYPE) || (code == ErrorCode.E_LEN) || (code == ErrorCode.E_COMPRESS_TYPE) || (code == ErrorCode.E_LEN) ||
(code == ErrorCode.E_PARSE); (code == ErrorCode.E_PARSE);
} }
/**
* Get the error code.
*
* @return Error code.
*/
public ErrorCode getCode() { public ErrorCode getCode() {
return code; return code;
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/fliter/DataFilter.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.fliter;
import com.oceanbase.clogproxy.client.enums.DBType;
import com.oceanbase.clogproxy.client.exception.DRCClientException;
import com.oceanbase.clogproxy.client.util.DataFilterUtil;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class DataFilter implements DataFilterBase {
/* Used to be compatibility with old version */
private String oldBranchDb;
private String filterInfo;
// String save the source filter string user passed through
private String sourceFilter;
// String save the filter that will be sent to store to acquire data
// For ob1.0, that must be four columns, dbname like a.b.
private String connectStoreFilterConditions;
private final StringBuilder builder;
private final Map<String, Map<String, List<String>>> requires;
private final Map<String, Map<String, List<String>>> dbTableColsReflectionMap;
//If all cols needed is '*', then we don't need do filter operation.
//In that case, we can save a lot compute.
private boolean isAllMatch = true;
private String tenant;
public DataFilter() {
oldBranchDb = null;
filterInfo = null;
builder = new StringBuilder();
requires = new HashMap<String, Map<String, List<String>>>();
dbTableColsReflectionMap = new HashMap<String, Map<String, List<String>>>();
}
/**
* Initialize the filter using formatted string.
* @param tenant tenant name
* @param tableFields the formatted filter information such as
* "tableName1;fieldName1;fieldName2|tableName2;fieldName1". No ";" or "|" should be
* transfer to
* *.tableName1.fieldName1|*.tableName1.fieldName2|...
* added at the beginning or end of the string.
*/
public DataFilter(String tenant, String tableFields) {
this(tableFields);
this.tenant = tenant;
}
public DataFilter(String tableFields) {
oldBranchDb = null;
builder = new StringBuilder();
requires = new HashMap<String, Map<String, List<String>>>();
dbTableColsReflectionMap = new HashMap<String, Map<String, List<String>>>();
builder.append(tableFields);
this.sourceFilter = tableFields;
}
/**
* The current version uses topic instead of dbname, so use the
* method to be compatible with the older version.
* @param db is the original branched db name.
*/
@Override
public void setBranchDb(final String db) {
oldBranchDb = db;
}
/**
* Add more filter information after initializing, note that the user should
* make it consistent to the formatted parameters.
* @param tableFields consistent formatted filter information.
*/
public void addTablesFields(String tableFields) {
builder.append(tableFields);
}
@Override
public boolean getIsAllMatch() {
return isAllMatch;
}
@Override
public Map<String, Map<String, List<String>>> getReflectionMap() {
return dbTableColsReflectionMap;
}
@Override
public Map<String, Map<String, List<String>>> getRequireMap() {
return requires;
}
//Before validate function called, toString may return null;
//Yet, user should not care about this. That's inter behavior.
@Override
public String toString() {
return connectStoreFilterConditions;
}
/**
* The validate function will form mysql, ob0.5, oracle eg filter condition.
*/
private boolean validateNormalFilterString() {
if (filterInfo != null) {
return true;
}
String s = builder.toString();
String[] tbs = s.split("\\|");
int colStart;
StringBuilder builder1 = new StringBuilder();
for (String s1 : tbs) {
String[] tb = s1.split("[;,\\.]");
if (tb.length > 0) {
String itemDb;
String itemTb;
if (tb.length <= 2) {
if (oldBranchDb != null) {
itemDb = oldBranchDb;
} else {
itemDb = "*";
}
colStart = 1;
itemTb = tb[0];
} else {
colStart = 2;
itemDb = tb[0];
itemTb = tb[1];
}
if (tenant != null) {
builder1.append(tenant).append(".");
}
builder1.append(itemDb).append(".").append(itemTb).append("|");
if (tb.length > colStart) {
List<String> cols = new ArrayList<String>();
for (int i = colStart; i < tb.length; i++) {
cols.add(tb[i]);
//here, we don't use trim in case that " *" or "* " or " * " is kind of col names
if (!"*".equals(tb[i])) {
isAllMatch = false;
}
}
DataFilterUtil.putColNames(itemDb, itemTb, cols, this);
}
}
}
if (builder1.charAt(builder1.length() - 1) == '|') {
builder1.deleteCharAt(builder1.length() - 1);
}
filterInfo = builder1.toString();
connectStoreFilterConditions = filterInfo;
return true;
}
/**
* The validate function will reform the filter condition and cols info
*/
private boolean validateOB10FilterString() {
if (sourceFilter == null) {
return false;
}
String[] tenantAndDbAndTBAndCols = sourceFilter.split("\\|");
requires.clear();
StringBuilder filterConditionBuilder = new StringBuilder();
for (String s1 : tenantAndDbAndTBAndCols) {
String[] tb = s1.split("[;,\\.]");
if (tb.length < 4) {
// tenant dbname tableName columnName is strictly required for 0b1.0
return false;
}
String tenant = tb[0];
String dbname = (oldBranchDb != null) ? oldBranchDb : tb[1];
String tableName = tb[2];
List<String> cols = new ArrayList<String>();
for (int i = 3; i < tb.length; ++i) {
cols.add(tb[i]);
if (!"*".equals(tb[i])) {
isAllMatch = false;
}
}
//format string passed to store
String formatDBName = tenant + "." + dbname;
filterConditionBuilder.append(formatDBName).append(FILTER_SEPARATOR_INNER);
filterConditionBuilder.append(tableName).append(FILTER_SEPARATOR);
DataFilterUtil.putColNames(formatDBName, tableName, cols, this);
}
connectStoreFilterConditions = filterConditionBuilder.toString();
return true;
}
// When source type is ocean base 1.0, filter's content is like tenant.dbname.tablename.colvalues| ....
@Override
public boolean validateFilter(DBType dbType) throws DRCClientException {
switch (dbType) {
case OCEANBASE1: {
return validateOB10FilterString();
}
default: {
return validateNormalFilterString();
}
}
}
@Override
public String getConnectStoreFilterConditions() {
return this.connectStoreFilterConditions;
}
}
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/fliter/DataFilterBase.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.fliter;
import com.oceanbase.clogproxy.client.enums.DBType;
import com.oceanbase.clogproxy.client.exception.DRCClientException;
import java.util.List;
import java.util.Map;
public interface DataFilterBase {
String FILTER_SEPARATOR_INNER = ".";
String FILTER_SEPARATOR = "|";
/**
* Get the formatted filter string which will be delivered to store.
* Notice: before validate filter function called, getConnectStoreFilterConditions may return null.
* @return filter string
*/
String getConnectStoreFilterConditions();
/**
* Validate if the filter user passed is legal
* @param dbType database type which may be ob, mysql, oracle.
* For now, only ob1.0 need special handle which 4 tuple contains tenant, db, tb, cols is strictly required.
* @return true if filter is valid
* @throws DRCClientException if exception occurs
*/
boolean validateFilter(DBType dbType) throws DRCClientException;
/**
* This function is compatible for old usage.
* @param branchDb the original branched db name.
*/
void setBranchDb(String branchDb);
/**
* Fast match if cols are all needed.
* @return true if all cols are needed
*/
boolean getIsAllMatch();
Map<String, Map<String, List<String>>> getReflectionMap();
Map<String, Map<String, List<String>>> getRequireMap();
}
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/listener/FieldParseListener.java
浏览文件 @ 1686e1dc
...@@ -12,7 +12,17 @@ package com.oceanbase.clogproxy.client.listener; ...@@ -12,7 +12,17 @@ package com.oceanbase.clogproxy.client.listener;
import com.oceanbase.clogproxy.client.message.DataMessage; import com.oceanbase.clogproxy.client.message.DataMessage;
/**
* This interface defined a kind of listener for field parsing.
*/
public interface FieldParseListener { public interface FieldParseListener {
public void parseNotify(DataMessage.Record.Field prev, DataMessage.Record.Field next) throws Exception;
} /**
\ No newline at end of file * Handle the filed parsing result.
*
* @param prev The original field.
* @param next The field after parsing.
* @throws Exception When exception occurs.
*/
void parseNotify(DataMessage.Record.Field prev, DataMessage.Record.Field next) throws Exception;
}
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/listener/RecordListener.java
浏览文件 @ 1686e1dc
...@@ -14,9 +14,22 @@ package com.oceanbase.clogproxy.client.listener; ...@@ -14,9 +14,22 @@ package com.oceanbase.clogproxy.client.listener;
import com.oceanbase.clogproxy.client.exception.LogProxyClientException; import com.oceanbase.clogproxy.client.exception.LogProxyClientException;
import com.oceanbase.clogproxy.client.message.LogMessage; import com.oceanbase.clogproxy.client.message.LogMessage;
/**
* This interface defined a kind of listener for record response.
*/
public interface RecordListener { public interface RecordListener {
void notify(LogMessage record); /**
* Handle the {@link LogMessage}.
*
* @param logMessage A {@link LogMessage} instance.
*/
void notify(LogMessage logMessage);
/**
* Handle the exception.
*
* @param e An exception.
*/
void onException(LogProxyClientException e); void onException(LogProxyClientException e);
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/listener/StatusListener.java
浏览文件 @ 1686e1dc
...@@ -12,6 +12,14 @@ package com.oceanbase.clogproxy.client.listener; ...@@ -12,6 +12,14 @@ package com.oceanbase.clogproxy.client.listener;
import com.oceanbase.clogproxy.common.packet.protocol.LogProxyProto; import com.oceanbase.clogproxy.common.packet.protocol.LogProxyProto;
/**
* This interface defined a kind of listener for {@link LogProxyProto.RuntimeStatus} response.
*/
public interface StatusListener { public interface StatusListener {
/**
* Handle the response of {@link LogProxyProto.RuntimeStatus}.
*
* @param status A {@link LogProxyProto.RuntimeStatus} response.
*/
void notify(LogProxyProto.RuntimeStatus status); void notify(LogProxyProto.RuntimeStatus status);
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/message/DataMessage.java
浏览文件 @ 1686e1dc
...@@ -10,7 +10,6 @@ See the Mulan PSL v2 for more details. */ ...@@ -10,7 +10,6 @@ See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.message; package com.oceanbase.clogproxy.client.message;
import com.oceanbase.clogproxy.client.config.DRCConfig;
import com.oceanbase.clogproxy.client.enums.DBType; import com.oceanbase.clogproxy.client.enums.DBType;
import com.oceanbase.clogproxy.client.listener.FieldParseListener; import com.oceanbase.clogproxy.client.listener.FieldParseListener;
import com.oceanbase.clogproxy.client.util.StringUtils; import com.oceanbase.clogproxy.client.util.StringUtils;
...@@ -854,25 +853,6 @@ public class DataMessage extends Message { ...@@ -854,25 +853,6 @@ public class DataMessage extends Message {
return records; return records;
} }
/**
* Construct the message from DataInputStream.
*
* @param reader is the DataInputStream.
* @param drcConfig DRCConfig
* @throws IOException if an I/O error occurs
*/
public void mergeFrom(final DataInputStream reader, DRCConfig drcConfig) throws IOException {
do {
Record record = new Record();
record.mergeFrom(reader);
record.setRegionId(drcConfig.getRegionId());
if (record.isEnding()) {
break;
}
records.add(record);
} while (true);
}
@Override @Override
public void clear() { public void clear() {
super.clear(); super.clear();
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/message/LogMessage.java
浏览文件 @ 1686e1dc
...@@ -12,7 +12,7 @@ package com.oceanbase.clogproxy.client.message; ...@@ -12,7 +12,7 @@ package com.oceanbase.clogproxy.client.message;
import com.oceanbase.clogproxy.client.constants.DataType; import com.oceanbase.clogproxy.client.constants.DataType;
import com.oceanbase.clogproxy.client.enums.DBType; import com.oceanbase.clogproxy.client.enums.DBType;
import com.oceanbase.clogproxy.client.exception.DRCClientRunTimeException; import com.oceanbase.clogproxy.client.exception.LogMessageException;
import com.oceanbase.clogproxy.client.listener.FieldParseListener; import com.oceanbase.clogproxy.client.listener.FieldParseListener;
import com.oceanbase.clogproxy.client.util.BinaryMessageUtils; import com.oceanbase.clogproxy.client.util.BinaryMessageUtils;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
...@@ -224,7 +224,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -224,7 +224,7 @@ public class LogMessage extends DataMessage.Record {
try { try {
dbName = BinaryMessageUtils.getString(byteBuf.array(), (int) dbNameOffset, UTF8_ENCODING); dbName = BinaryMessageUtils.getString(byteBuf.array(), (int) dbNameOffset, UTF8_ENCODING);
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
} }
} }
...@@ -240,7 +240,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -240,7 +240,7 @@ public class LogMessage extends DataMessage.Record {
try { try {
tableName = BinaryMessageUtils.getString(byteBuf.array(), (int) tbNameOffset, UTF8_ENCODING); tableName = BinaryMessageUtils.getString(byteBuf.array(), (int) tbNameOffset, UTF8_ENCODING);
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
} }
} }
...@@ -267,7 +267,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -267,7 +267,7 @@ public class LogMessage extends DataMessage.Record {
serverId = BinaryMessageUtils.getString(byteBuf.array(), (int) instanceOffset, serverId = BinaryMessageUtils.getString(byteBuf.array(), (int) instanceOffset,
DEFAULT_ENCODING); DEFAULT_ENCODING);
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
} }
} }
...@@ -634,7 +634,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -634,7 +634,7 @@ public class LogMessage extends DataMessage.Record {
} catch (Exception e) { } catch (Exception e) {
fields = null; fields = null;
throw new DRCClientRunTimeException(e.getMessage(), e); throw new LogMessageException(e.getMessage(), e);
} }
return fields; return fields;
...@@ -660,7 +660,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -660,7 +660,7 @@ public class LogMessage extends DataMessage.Record {
} }
} }
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
return primaryKeyIndexList; return primaryKeyIndexList;
} }
...@@ -1014,7 +1014,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -1014,7 +1014,7 @@ public class LogMessage extends DataMessage.Record {
return pkValues; return pkValues;
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
} }
...@@ -1076,7 +1076,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -1076,7 +1076,7 @@ public class LogMessage extends DataMessage.Record {
} }
} }
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e); throw new LogMessageException(e);
} }
return tuples; return tuples;
...@@ -1092,7 +1092,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -1092,7 +1092,7 @@ public class LogMessage extends DataMessage.Record {
} }
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
} }
...@@ -1157,7 +1157,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -1157,7 +1157,7 @@ public class LogMessage extends DataMessage.Record {
} }
} }
} catch (Exception e) { } catch (Exception e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
return uniqueKeyList; return uniqueKeyList;
} }
...@@ -1234,7 +1234,7 @@ public class LogMessage extends DataMessage.Record { ...@@ -1234,7 +1234,7 @@ public class LogMessage extends DataMessage.Record {
return BinaryMessageUtils.getString(byteBuf.array(), (int) encoding, return BinaryMessageUtils.getString(byteBuf.array(), (int) encoding,
DEFAULT_ENCODING); DEFAULT_ENCODING);
} catch (UnsupportedEncodingException e) { } catch (UnsupportedEncodingException e) {
throw new DRCClientRunTimeException(e.getMessage(), e.getCause()); throw new LogMessageException(e.getMessage(), e.getCause());
} }
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/util/BinaryMessageUtils.java
浏览文件 @ 1686e1dc
...@@ -21,19 +21,24 @@ import java.nio.ByteOrder; ...@@ -21,19 +21,24 @@ import java.nio.ByteOrder;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
/**
* Utils class for binary message.
*/
public class BinaryMessageUtils { public class BinaryMessageUtils {
/**
*
*/
private static final int PREFIX_LENGTH = 12; private static final int PREFIX_LENGTH = 12;
/** /**
* get string begin with offset * Get string begin with offset.
* *
* @param data bytes array * @param data A bytes array.
* @param offset read offset * @param offset Reading offset.
* @param encoding string encoding * @param encoding String encoding.
* @return result string * @return Result string.
* @throws UnsupportedEncodingException when the encoding is not supported * @throws UnsupportedEncodingException When the encoding is not supported.
*/ */
public static String getString(byte[] data, int offset, String encoding) throws UnsupportedEncodingException { public static String getString(byte[] data, int offset, String encoding) throws UnsupportedEncodingException {
ByteBuf wrapByteBuf = Unpooled.wrappedBuffer(data).order(ByteOrder.LITTLE_ENDIAN); ByteBuf wrapByteBuf = Unpooled.wrappedBuffer(data).order(ByteOrder.LITTLE_ENDIAN);
...@@ -47,12 +52,12 @@ public class BinaryMessageUtils { ...@@ -47,12 +52,12 @@ public class BinaryMessageUtils {
} }
/** /**
* get list begin with offset * Get list begin with offset.
* *
* @param data bytes array * @param data A bytes array.
* @param offset read offset * @param offset Reading offset.
* @return result list * @return Result list.
* @throws IOException if data type is unsigned long * @throws IOException If data type is unsigned long.
*/ */
public static List getArray(byte[] data, int offset) throws IOException { public static List getArray(byte[] data, int offset) throws IOException {
ByteBuf wrapByteBuf = Unpooled.wrappedBuffer(data).order(ByteOrder.LITTLE_ENDIAN); ByteBuf wrapByteBuf = Unpooled.wrappedBuffer(data).order(ByteOrder.LITTLE_ENDIAN);
...@@ -107,11 +112,11 @@ public class BinaryMessageUtils { ...@@ -107,11 +112,11 @@ public class BinaryMessageUtils {
} }
/** /**
* get ByteString begin with offset * Get ByteString begin with offset.
* *
* @param data bytes array * @param data A bytes array.
* @param offset read offset * @param offset Reading offset.
* @return list of ByteString * @return A list of {@link ByteString}.
*/ */
public static List<ByteString> getByteStringList(byte[] data, long offset) { public static List<ByteString> getByteStringList(byte[] data, long offset) {
if (offset == -1) { if (offset == -1) {
...@@ -138,8 +143,8 @@ public class BinaryMessageUtils { ...@@ -138,8 +143,8 @@ public class BinaryMessageUtils {
lists.add(null); lists.add(null);
} else { } else {
lists.add(new ByteString(wrapByteBuf.array(), lists.add(new ByteString(wrapByteBuf.array(),
PREFIX_LENGTH + currentOffset + readBytes + (int) offset, PREFIX_LENGTH + currentOffset + readBytes + (int) offset,
nextOffset - currentOffset - 1)); nextOffset - currentOffset - 1));
} }
currentOffset = nextOffset; currentOffset = nextOffset;
} }
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/util/ClientIdGenerator.java
浏览文件 @ 1686e1dc
...@@ -14,18 +14,25 @@ import com.oceanbase.clogproxy.common.util.NetworkUtil; ...@@ -14,18 +14,25 @@ import com.oceanbase.clogproxy.common.util.NetworkUtil;
import java.lang.management.ManagementFactory; import java.lang.management.ManagementFactory;
/**
* The class used to generate client id.
*/
public class ClientIdGenerator { public class ClientIdGenerator {
/** /**
* LocalIP_PID_currentTimestamp * Generate a new client id in format "LocalIP"."PID"."currentTimestamp".
* pattern may be change, never depend on the content of this * Pattern may be changed, never depend on the content of this.
*
* @return client id string * @return Client id string.
*/ */
public static String generate() { public static String generate() {
return NetworkUtil.getLocalIp() + "_" + getProcessId() + "_" + (System.currentTimeMillis() / 1000); return NetworkUtil.getLocalIp() + "_" + getProcessId() + "_" + (System.currentTimeMillis() / 1000);
} }
/**
* Get the process id.
*
* @return Process id.
*/
private static String getProcessId() { private static String getProcessId() {
// Note: may fail in some JVM implementations // Note: may fail in some JVM implementations
// therefore fallback has to be provided // therefore fallback has to be provided
......
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/util/DataFilterUtil.java 已删除 100644 → 0
浏览文件 @ 8cae74a2
/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
oblogclient is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
package com.oceanbase.clogproxy.client.util;
import com.oceanbase.clogproxy.client.fliter.DataFilterBase;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class DataFilterUtil {
/**
*
* @param db db name
* @param tb table name
* @param cols column names
* @param dataFilterBase DataFilterBase
*/
public static void putColNames(String db, String tb, List<String> cols,
DataFilterBase dataFilterBase) {
if (tb == null) {
return;
}
Map<String, Map<String, List<String>>> dbAndTablePair = dataFilterBase.getRequireMap();
boolean founded = false;
for (Map.Entry<String, Map<String, List<String>>> dbEntry : dbAndTablePair.entrySet()) {
if (db == null || db.equalsIgnoreCase(dbEntry.getKey())) {
for (Map.Entry<String, List<String>> entry : dbEntry.getValue().entrySet()) {
if (tb.equalsIgnoreCase(entry.getKey())) {
founded = true;
entry.getValue().addAll(cols);
}
}
if (!founded) {
// db is already in the filter, but the table is not, so add the table
Map<String, List<String>> tabMap = dbEntry.getValue();
tabMap.put(tb, cols);
founded = true;
}
}
}
if (!founded) {
// db is not in the filter, so add two maps
Map<String, List<String>> tabMap = new HashMap<String, List<String>>();
tabMap.put(tb, cols);
dbAndTablePair.put(db, tabMap);
}
}
/**
* Use the give db and tb name to retrieve cols list
* @param db db name
* @param tb table name
* @param dataFilterBase DataFilterBase
* @return cols reference to corresponded db name and table name
* Note: this function get cols from map in old DataFilter implementation
*/
public static List<String> getColNamesWithMapping(String db, String tb,
DataFilterBase dataFilterBase) {
if (tb == null) {
return null;
}
Map<String, Map<String, List<String>>> dbAndTablePair = dataFilterBase.getReflectionMap();
Map<String, List<String>> tableAndCols = dbAndTablePair.get(db);
if (tableAndCols == null) {
//if we don't find tableAndCols, that mean this dbName appears for the first time,
//and we use getColNames to require the missing cols and update map;
tableAndCols = new HashMap<String, List<String>>();
List<String> cols = getColNames(db, tb, dataFilterBase);
tableAndCols.put(tb, cols);
dbAndTablePair.put(db, tableAndCols);
return cols;
} else {
List<String> needCols = tableAndCols.get(tb);
//we propose the cols can't be null, so we use null to determinate whether the cols we
//needed has existed in the map
if (needCols == null) {
//the cols we needed is missing ,use getColNames to require the missing cols
List<String> cols = getColNames(db, tb, dataFilterBase);
tableAndCols.put(tb, cols);
return cols;
} else {
//the cols existed, just return the value.
return needCols;
}
}
}
/**
* Use the give db and tb name to retrieve cols list
* @param db db name
* @param tb table name
* @param dataFilterBase DataFilterBase
* @return cols reference to corresponded db name and table name
*/
public static List<String> getColNames(String db, String tb, DataFilterBase dataFilterBase) {
if (tb == null) {
return null;
}
Map<String, Map<String, List<String>>> requireMap = dataFilterBase.getRequireMap();
for (Map.Entry<String, Map<String, List<String>>> dbEntry : requireMap.entrySet()) {
StringBuffer buf = new StringBuffer(dbEntry.getKey());
processStringToRegularExpress(buf);
if (db == null || db.toLowerCase().matches(buf.toString().toLowerCase())) {
for (Map.Entry<String, List<String>> entry : dbEntry.getValue().entrySet()) {
buf = new StringBuffer(entry.getKey());
processStringToRegularExpress(buf);
if (tb.toLowerCase().matches(buf.toString().toLowerCase())) {
return entry.getValue();
}
}
}
}
return null;
}
/**
* This function will first replace all "." to "\.", then replace all "*" to ".*"
*
* @param stringBuffer original string buffer
*/
public static void processStringToRegularExpress(StringBuffer stringBuffer) {
int index;
int beginIndex = 0;
while (-1 != (index = stringBuffer.indexOf(".", beginIndex))) {
stringBuffer.insert(index, '\\');
beginIndex = index + 2;
}
beginIndex = 0;
while (-1 != (index = stringBuffer.indexOf("*", beginIndex))) {
stringBuffer.insert(index, '.');
beginIndex = index + 2;
}
}
/**
* Judge if the given col name exists in col lists
* @param col col to be judged
* @param s cols list
* @return true if exists in, else false
*/
public static boolean isColInArray(final String col, final List<String> s) {
for (int i = 0; i < s.size(); i++) {
if ("*".equals(s.get(i)) || col.equalsIgnoreCase(s.get(i))) {
return true;
}
}
return false;
}
}
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/util/NettyEventLoopUtil.java
浏览文件 @ 1686e1dc
...@@ -20,27 +20,34 @@ import io.netty.channel.socket.nio.NioSocketChannel; ...@@ -20,27 +20,34 @@ import io.netty.channel.socket.nio.NioSocketChannel;
import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadFactory;
/**
* Utils class for netty.
*/
public class NettyEventLoopUtil { public class NettyEventLoopUtil {
/** check whether epoll enabled, and it would not be changed during runtime. */ /**
private static boolean epollEnabled = Epoll.isAvailable(); * Check whether epoll enabled, and it would not be changed during runtime.
*/
private static final boolean EPOLL_ENABLED = Epoll.isAvailable();
/** /**
* Create the right event loop according to current platform and system property, fallback to NIO when epoll not enabled. * Create a new {@link EventLoopGroup} according to current platform and system property, fallback to NIO when epoll not enabled.
* *
* @param nThreads number of threads * @param nThreads Number of threads.
* @param threadFactory ThreadFactory * @param threadFactory A {@link ThreadFactory} instance.
* @return an EventLoopGroup suitable for the current platform * @return An {@link EventLoopGroup} instance.
*/ */
public static EventLoopGroup newEventLoopGroup(int nThreads, ThreadFactory threadFactory) { public static EventLoopGroup newEventLoopGroup(int nThreads, ThreadFactory threadFactory) {
return epollEnabled ? new EpollEventLoopGroup(nThreads, threadFactory) return EPOLL_ENABLED ? new EpollEventLoopGroup(nThreads, threadFactory)
: new NioEventLoopGroup(nThreads, threadFactory); : new NioEventLoopGroup(nThreads, threadFactory);
} }
/** /**
* @return a SocketChannel class suitable for the given EventLoopGroup implementation * Get the suitable {@link SocketChannel} class according to current platform and system property.
*
* @return A {@link SocketChannel} implementation class.
*/ */
public static Class<? extends SocketChannel> getClientSocketChannelClass() { public static Class<? extends SocketChannel> getClientSocketChannelClass() {
return epollEnabled ? EpollSocketChannel.class : NioSocketChannel.class; return EPOLL_ENABLED ? EpollSocketChannel.class : NioSocketChannel.class;
} }
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/util/StringUtils.java
浏览文件 @ 1686e1dc
...@@ -13,14 +13,17 @@ package com.oceanbase.clogproxy.client.util; ...@@ -13,14 +13,17 @@ package com.oceanbase.clogproxy.client.util;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
/**
* Utils class for string.
*/
public class StringUtils { public class StringUtils {
/** /**
* Split a string by one separator character. The performance * Split a string by one separator character. The performance
* is better than Java String split. * is better than Java String split.
* *
* @param str is the string need be split. * @param str The string need be split.
* @param separatorChar the single separator character. * @param separatorChar The single separator character.
* @return the array of split items. * @return The array of split items.
*/ */
public static String[] split(String str, char separatorChar) { public static String[] split(String str, char separatorChar) {
if (str == null) { if (str == null) {
...@@ -57,6 +60,6 @@ public class StringUtils { ...@@ -57,6 +60,6 @@ public class StringUtils {
list.add(str.substring(start, i)); list.add(str.substring(start, i));
} }
return (String[]) list.toArray(new String[list.size()]); return list.toArray(new String[list.size()]);
} }
} }
logproxy-client/src/main/java/com/oceanbase/clogproxy/client/util/Validator.java
浏览文件 @ 1686e1dc
...@@ -12,29 +12,56 @@ package com.oceanbase.clogproxy.client.util; ...@@ -12,29 +12,56 @@ package com.oceanbase.clogproxy.client.util;
import java.util.Map; import java.util.Map;
/**
* Utils class used to validate arguments.
*/
public class Validator { public class Validator {
private static final int MINIMAL_VALID_PORT = 1; private static final int MINIMAL_VALID_PORT = 1;
private static final int MAXIMAL_VALID_PORT = 65535; private static final int MAXIMAL_VALID_PORT = 65535;
/**
* Validate the object is not null, otherwise throws an {@link NullPointerException}.
*
* @param obj Object to be verified.
* @param message Message in the NullPointerException.
*/
public static void notNull(Object obj, String message) { public static void notNull(Object obj, String message) {
if (obj == null) { if (obj == null) {
throw new NullPointerException(message); throw new NullPointerException(message);
} }
} }
/**
* Validate the port number is valid, otherwise throws an {@link IllegalArgumentException}.
*
* @param port Port number to be verified.
* @param message Message in the IllegalArgumentException.
*/
public static void validatePort(int port, String message) { public static void validatePort(int port, String message) {
if (port < MINIMAL_VALID_PORT || port >= MAXIMAL_VALID_PORT) { if (port < MINIMAL_VALID_PORT || port >= MAXIMAL_VALID_PORT) {
throw new IllegalArgumentException(message); throw new IllegalArgumentException(message);
} }
} }
/**
* Validate the string is not null or empty, otherwise throws an {@link IllegalArgumentException}.
*
* @param val String to be verified.
* @param message Message in the IllegalArgumentException.
*/
public static void notEmpty(String val, String message) { public static void notEmpty(String val, String message) {
if (val == null || val.isEmpty()) { if (val == null || val.isEmpty()) {
throw new IllegalArgumentException(message); throw new IllegalArgumentException(message);
} }
} }
/**
* Validate the map is not null or empty, otherwise throws an {@link IllegalArgumentException}.
*
* @param map Map to be verified.
* @param message Message in the IllegalArgumentException.
*/
public static void notEmpty(Map<String, String> map, String message) { public static void notEmpty(Map<String, String> map, String message) {
if (map == null || map.isEmpty()) { if (map == null || map.isEmpty()) {
throw new IllegalArgumentException(message); throw new IllegalArgumentException(message);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或