# README
## 项目功能
## 项目依赖
`Netty-socketio` 与`Socket.io-client-Java`的对应关系是：
| [`netty-socketio` ](https://github.com/mrniko/netty-socketio ) | [`Java client` ](https://github.com/socketio/socket.io-client-java ) |
| ------------------------------------------------------------ | ------------------------------------------------------------ |
| 1.7.19 | 1.0.x |
| 暂无 | [Document ](https://socketio.github.io/socket.io-client-java/installation.html ) |
以下用`Service`指代`Socket`连接中的`socket`服务器，它也是需求查询用户内网数据源的公网服务器。
用`Agent`指代`Socket`连接中的客户端，也是运行在用户`PC`上承担远程调用`JDBC`方法的代理服务。
具体结构见下文项目结构图。
## Quick Start
1. 运行`Service`模块下 `ServiceTest`
2. 运行`Agent`下 Test
## 实现方案
1. `Service` 启动`socket`服务与 `Agent` 建立连接后，可以开始使用代理进行查询。
2. `Service` 端通过自实现的`JDBC`驱动，进行`JDBC`操作。驱动中使用基于`CGlib`的动态代理，对`Service`端的所有`JDBC`相关驱动类进行增强，所有方法信息会被序列化传递到`Agent`执行，并有选择地将结果回送到`Service`
## 结构与流程
如上图，对于`Service` 端来讲，`Agent`对其的代理是无感知的。在`Service`来看，只是调用了一个自定义的`JDBC`驱动进行查询。
这得益于驱动内部方法地重写，自定义地实现类在`Agent`和`Service`中有相同的名字，但内部实现却不相同，这使得整个RPC的流程十分灵活。
## 方案细节
### 动态代理
动态代理是该项目中的核心，如在 `Driver` 类的 `connect` 方法中：返回的`Connection`就被替换为了动态代理增强过的`MyConnection`，实现对`Service`中调用的`JDBC`方法的完全代理。代理类会依靠`info`从缓存中找到命名空间（本项目中以`/dataSoure Name`来区别命名空间）对应的`socket`,将方法调用信息以`RPCReqquest`的方式序列化后发送出去。
```java
// In Service Source Code
@Override
public Connection connect(String url, Properties info) throws SQLException {
String agentID = info.getProperty("agentID");
String dbName = info.getProperty("agentDBName");
if(dbName == null){
dbName = url.split(":")[1];
info.setProperty("agentDBName", dbName);
}
MyConnection myConn = (MyConnection) ProxyFactory.getProxy(MyConnection.class, info);
myConn.setInfo(info);
return myConn;
}
```
RPC实体类包含如下信息：
```java
@Data
@Accessors (chain = true)
public class RpcRequest {
// Marks whether the method delivered need loopback data
private boolean reply;
// Marks whether the method will create an instance requeired to be cached.
private boolean binding;
private String ID;
private String IDtoInvoke;
private Class ServiceClass;
private String MethodName;
private Object[] args;
private Class[] argTypes;
}
```
在`Agent`收到`Request`的时候，会按照报文要求对方法进行调用，某些创建的实例会被缓存，以便之后调用。在本项目中，这些实例的类是：
```
Drive( MyDriver ), Connection( MyConnection ), Statement( MyStatement ), PreparedStatement( MyPreparedStatement ), ResultSet( MyResult )
```
```java
public Object invokeAsRequest(RpcRequest rpcRequest, BeanCache beanCache) {
...
// The ID of the rpcRequest could be save as the ID of an instance
// Because one instance can only been create just once for an unique rpcRequest
String IDtoCache = rpcRequest.getID();
String IDtoInvoke = rpcRequest.getIDtoInvoke();
...
```
### 注解解耦
在`Service`端，定义了以下注解。
```
annotation
- BindingParameter
- LocalMethod
- RemoteClass
- WithBindingParameter
```
如上文所说，注册了自定义驱动后。生成的增强类所调用的方法都会生成 RPC 请求，为了能够松散的实现对不同方法的请求控制，需要一些自定义的注解。
`LocalMethod` ：方法注释，标记不需要远程调用的方法。
`RemoteClass` ：类注释，标记该类远程调用时对应的类。
`WithBindingParameter` : 方法注释，标记包含绑定类方法。因为绑定类作为参数时，实际需要传递的实际是`Agentxx`类中持有的成员变量。
`BindingParameter` : 参数注释，标记参数中的绑定类。
在`Agent`端，定义了以下注解注解。
```java
annotation
- BindClass
- BindingParameter
- WithBindingParameter
```
`BindingPatameter` 和`WithBindingParameter`同上。
`BindClass` ：类注释，标记绑定类，在`Agent`中用于判断该实例是否需要缓存。
### RPC调用
在一次RPC调用流程中，`FutureTask` 异步获取返回结果，以“生产者-消费者”模型实现一次调用的同步管理。
`ClientWrapper` 持有着各个命名空间上的`socket`。在这些`socket`上的通信，每次调用，会在`wrapper`中注册一个工具类:`LockAndCondition`，发出消息后，等待`socket`上出现对应的响应报文唤醒`FutureTask` 线程。通过锁机制，保证逻辑的正确性。
```java
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ClientWrapper {
private SocketIOClient client;
private static Map< String , LockAndCondition > lockMap = new ConcurrentHashMap< >();
public SocketIOClient getClient(){
if(client == null) throw new RuntimeException("no such client");
return client;
}
public LockAndCondition getLockAndCondition(String messageID){
LockAndCondition lac = lockMap.get(messageID);
if(lac == null){
ReentrantLock lock = new ReentrantLock();
Condition condition = lock.newCondition();
lac = new LockAndCondition(lock, condition);
lockMap.put(messageID, lac);
}
return lac;
}
public void removeLockAndCondition(String messageID){
lockMap.remove(messageID);
}
}
```
```java
@Data
@NoArgsConstructor
@AllArgsConstructor
public class LockAndCondition{
private ReentrantLock lock;
private Condition condition;
private Object result;
private String BindingID;
LockAndCondition(ReentrantLock lock, Condition condition){
this.lock = lock;
this.condition = condition;
}
}
```
```java
FutureTask< Object > futureTask = new FutureTask< Object > (
new Callable< Object > () {
@Override
public Object call() throws Exception {
Object res = null;
ClientWrapper wrapper = ClientCache.getClientWrapper(agentID, dbName);
LockAndCondition lac = wrapper.getLockAndCondition(rpcRequest.getID());
ReentrantLock lock = lac.getLock();
Condition condition = lac.getCondition();
try{
byte[] bytes = ServerStater.serializer.serialize(rpcRequest);
lock.lock();
client.sendEvent("RPCRequest", bytes);
condition.await();
// get res from RPC response data
res = lac.getResult();
}catch (Exception e){
e.printStackTrace();
}finally {
lock.unlock();
}
return res;
}
}
);
ServerStater.threadPool.submit(futureTask);
Object res = futureTask.get();
```
`socket` 收到响应时解锁对应的线程。
```java
// rpcResponse
nameSpace.addEventListener("RPCResponse", byte[].class, ((client, data, ackRequest) -> {
RpcResponse rpcResponse = serializer.deserialize(data, RpcResponse.class);
logger.debug("RPCResponse: " + (rpcResponse.getStatus() ? "success" : "fail"));
String agentID = Commons.getAgentID(client);
String dbName = Commons.getDBName(client);
ClientWrapper wrapper = ClientCache.getClientWrapper(agentID, dbName);
LockAndCondition lac = wrapper.getLockAndCondition(rpcResponse.getID());
ReentrantLock lock = lac.getLock();
Condition condition = lac.getCondition();
// When a response is received, it notifies that the futuretask thread blocking on the lockandcondition
// If the response contains data, take it out.
try {
lock.lock();
Object resultData = rpcResponse.getResult();
if(!rpcResponse.getStatus()){
logger.error(resultData);
resultData = null;
}
if(resultData != null) lac.setResult(resultData);
condition.signal();
}catch (Exception e){
e.printStackTrace();
}finally {
lock.unlock();
}
wrapper.removeLockAndCondition(rpcResponse.getID());
logger.debug("received response message, signaled condition");
}));
```
`Service` 是使用`netty`实现的高效同步非阻塞`IO`，上文的同步机制可以很大程度上利用`socket`的并发效果。
### 绑定实例
确定`Agent`上缓存实例与`Service`端实例的一一对应关系是很必要，不然程序在反射调用方法时会产生问题。
例如，对于`createStatement()`方法必须由上一步生成的`Connection`类进行调用。为了达到这一点，这些`Service`端实例必须和`Agent`端具有相同的ID。
考虑到在进行`RPC`调用回调的时候，利用时间和随机数生成了一个唯一`ID`。
```java
public static String getID(){
return getTimeInMillis() + getRandom();
}
public static String getTimeInMillis() {
long timeInMillis = Calendar.getInstance().getTimeInMillis();
return timeInMillis+"";
}
public static String getRandom() {
Random random = new Random();
int nextInt = random.nextInt(9000000);
nextInt=nextInt+1000000;
String str=nextInt+"";
return str;
}
```
而`Agent`端的缓存实例是由某次调用产生的，所以只需将该次调用的`RPC`报文`ID`标记在实例上，并在收到`RPC`响应时为需要绑定的类型打上同样的标记即可。这样`Agent`方面，由于存储的实例都有了唯一的`ID`作为键，大大简化了缓存系统的复杂性。
标记实现：
```java
@Override
public Object intercept(Object o, Method method, Object[] objects, MethodProxy methodProxy) throws Throwable { ...
Object returnObj = methodProxy.invokeSuper(o, objects);
// If the return instance is corresponding with another instance in agent, set the binding ID.
if (InterceptorUtils.isInBindList(returnObj)){
InterceptorUtils.setInvokeHelper(returnObj, "setID", rpcRequest.getID());
}
```
# 测试
测试在`Intranet`中`test`内。
## 启动流程
1. 构造客户端与服务端实例
可见于`AbstractDriverTest`类
```java
static final String[][] dbNameAndDriver = new String[][]{
DBProperties.HSQL
};
// 服务端与客户端应注册同样的一个或多个驱动，但服务端中不需要加载，仅仅用于构造命名空间
static final ServerStater server = new ServerStater(dbNameAndDriver);
static final AgentStarter agent = new AgentStarter(dbNameAndDriver);
```
2. 此时 代理端/服务端 实例已经启动，服务端应优先注册代理端信息，否则会报错。
```java
info.setProperty("user", "sa");
info.setProperty("password", "");
info.setProperty("agentID", "1001");
info.setProperty("agentDBName", DBProperties.HSQL[0]);
```
3. 模拟服务端等待连接
```java
static void openSocket(){
while(ClientCache.getClient(
info.getProperty("agentID"),
info.getProperty("agentDBName"))
== null){
}
}
```
4. 连接就绪即可获取连接，此时运行`TestSuit`即可。
## 使用Junit5 和 HSQL 进行集成测试
这一次项目的测试记录，项目用 IDEA 编写，`Junit 5`进行单元测试，`HSQL DB`提供方便的内存数据库。
## Use Junit 5
### Configuring JUnit Platform
To get started with JUnit Platform, you need to add at least a single `TestEngine` implementation to your project. For example, if you want to write tests with Jupiter, add the test artifact `junit-jupiter-engine` to the dependencies in POM:
```xml
< dependencies >
[...]
< dependency >
< groupId > org.junit.jupiter< / groupId >
< artifactId > junit-jupiter-engine< / artifactId >
< version > 5.4.0< / version >
< scope > test< / scope >
< / dependency >
[...]
< / dependencies >
```
This will pull in all required dependencies. Among those dependencies is `junit-jupiter-api` which contains the classes and interfaces your test source requires to compile. `junit-platform-engine` is also resolved and added.
This is the only step that is required to get started - you can now create tests in your test source directory (`src/test/java`).
### 集成测试的细节
#### [生命周期](https://junit.org/junit5/docs/current/user-guide/#writing-tests-test-instance-lifecycle)
> In order to allow individual test methods to be executed in isolation and to avoid unexpected side effects due to mutable test instance state, JUnit creates a new instance of each test class before executing each *test method* (see [Test Classes and Methods](https://junit.org/junit5/docs/current/user-guide/#writing-tests-classes-and-methods)). This "per-method" test instance lifecycle is the default behavior in JUnit Jupiter and is analogous to all previous versions of JUnit.
>
> If you would prefer that JUnit Jupiter execute all test methods on the same test instance, annotate your test class with `@TestInstance(Lifecycle.PER_CLASS)`. When using this mode, a new test instance will be created once per test class. Thus, if your test methods rely on state stored in instance variables, you may need to reset that state in `@BeforeEach` or `@AfterEach` methods.
根据官方文档，Junit5的默认测试实例的生命周期为`pre-method`在集成测试时，应当选择`pre_class`，即一个测试类创建一个实例。
#### 成员变量
在`pre-class`的生命周期下，整个测试过程中，测试方法有时需要共享变量。比如进行`JDBC查询`的时候，只需要共用一个`connection`，甚至是一个`statement`.
#### [Method Order](https://junit.org/junit5/docs/current/user-guide/#writing-tests-test-execution-order-methods)
> Although true *unit tests* typically should not rely on the order in which they are executed, there are times when it is necessary to enforce a specific test method execution order
除上述两点之外，在集成测试中，必不可少的一项是使用Order控制方法的执行顺序。
例子：
```java
import org.junit.jupiter.api.MethodOrderer.OrderAnnotation;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
@TestMethodOrder (OrderAnnotation.class)
class OrderedTestsDemo {
@Test
@Order (1)
void nullValues() {
// perform assertions against null values
}
@Test
@Order (2)
void emptyValues() {
// perform assertions against empty values
}
@Test
@Order (3)
void validValues() {
// perform assertions against valid values
}
}
```
## 使用 HSQL DB
### 运行模式
HSQL DB 是一款纯Java编写的免费数据库，并且支持Memory-Only模型，所以每个单测使用独立的数据库就有了可能，Memory-Only模型数据不会持久化跑完即销毁，从根源上解决了测试难以复现和测试对数据的污染问题。
[Memory-Only 内存模式 ](https://blog.csdn.net/smmi/article/details/83779238 )
### 版本问题
maven 依赖
```xml
< dependency >
< groupId > org.hsqldb< / groupId >
< artifactId > hsqldb< / artifactId >
< version > 2.5.2< / version >
< scope > test< / scope >
< / dependency >
```
根据官网:
> Latest version 2.7.0 works with JDK 8 and above. Version for JDK 6 is also available.
但是这不代表在测试的时候可以直接调用 2.7.0 版本的代码，因为该版本使用 jdk 11 编译，在使用 IDEA 测试时，直接引用会报错，所以版本应该选择 2.5x .
代码：
```java
public class TestHSQL{
@BeforeAll
static void startHSQL(){
try {
Class.forName(DBProperties.HSQL_DTIVER_NAME);
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
}
...
}
```
错误信息：
```
java.lang.UnsupportedClassVersionError: org/hsqldb/jdbcDriver has been compiled by a more recent version of the Java Runtime (class file version 55.0), this version of the Java Runtime only recognizes class file versions up to 52.0
```
### 语法问题
HSQL 与 MySQL 语法并不相同，[HSQL语法](https://www.hsqldb.org/doc/2.0/guide/databaseobjects-chapt.html#dbc_schema_def_statements)，英文文档读起来稍微有点费劲。
```mysql
CREATE TABLE student1 (
student_id int(32) PRIMARY KEY AUTO_INCREMENT NOT NULL,
student_name varchar(100) NOT NULL,
student_address varchar(100) NOT NULL
);
```
如上的sql语句在程序中要这么写，差别还是很多的，对于大部分在标准 SQL 语句之外的语句都需要先检查是否使用了正确的语法。
```java
st.executeUpdate("CREATE TABLE student (" +
"student_id INTEGER GENERATED BY DEFAULT AS IDENTITY " +
"(START WITH 1, INCREMENT BY 1) NOT NULL," +
"student_name VARCHAR(100) NOT NULL," +
"student_address VARCHAR(100) NOT NULL," +
"PRIMARY KEY (student_id)" +
");");
```
其实还有个取巧的方法，可以使用参数直接开启`mysql`语法支持，但依然有些语法`MySQL`特点`HSQKDB`并不支持，此时应参考`HDBSQL`所实现的`SQL92`标准。
```
"jdbc:hsqldb:mem:test;sql.syntax_mys=true"
```
## 使用示例
```java
import com.fanruan.AgentStarter;
import com.fanruan.ServerStater;
import com.fanruan.service.jdbc.driver.ServiceDriver;
import com.fanruan.proxy.ProxyFactory;
import com.fanruan.utils.DBProperties;
import org.junit.jupiter.api.*;
import java.sql.*;
import java.util.Properties;
/**
* @author Yichen Dai
* @date 2022/8/18 15:27
*/
@TestInstance (TestInstance.Lifecycle.PER_CLASS)
@TestMethodOrder (MethodOrderer.OrderAnnotation.class)
public class TestUtil {
static Connection conn = null;
static Statement st = null;
static PreparedStatement pst = null;
static ResultSet rs = null;
static void configService(){
// 启动socket服务器
String[][] DBs = new String[][]{
DBProperties.HSQL,
};
new ServerStater(DBs);
}
static void configAgent(){
// 启动socket客户端
String[][] DBs = new String[][]{
DBProperties.HSQL,
};
new AgentStarter(DBs);
}
@BeforeAll
static void autoConfig(){
configService();
configAgent();
try {
// 等待socket连接
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Test
@Order (1)
void testConnect() throws SQLException {
// 建立连接
Properties info = new Properties();
info.setProperty("user", "sa");
info.setProperty("password", "");
info.setProperty("agentID", "1001");
info.setProperty("agentDBName", DBProperties.HSQL[0]);
Driver driver = (ServiceDriver) ProxyFactory.getProxy(ServiceDriver.class, null);
conn = driver.connect("jdbc:hsqldb:mem:test;sql.syntax_mys=true", info);
}
@Test
@Order (2)
void testCreateTable1() throws SQLException {
// 创建 statement
st = conn.createStatement();
// 创建表
int num = st.executeUpdate("DROP TABLE student IF EXISTS;");
Assertions.assertEquals(0, num);
num = st.executeUpdate("CREATE TABLE student (" +
"student_id INTEGER GENERATED BY DEFAULT AS IDENTITY " +
"(START WITH 1, INCREMENT BY 1) NOT NULL," +
"student_name VARCHAR(100) NOT NULL," +
"student_address VARCHAR(100) NOT NULL," +
"PRIMARY KEY (student_id)" +
");");
Assertions.assertEquals(0, num);
}
@Test
@Order (3)
void testCreateTable2() throws SQLException {
// 创建表
int num = st.executeUpdate("DROP TABLE student_score IF EXISTS;");
Assertions.assertEquals(0, num);
num = st.executeUpdate("CREATE TABLE score (" +
"student_id int(10) PRIMARY KEY NOT NULL," +
"score int(10) NOT NULL" +
");"
);
Assertions.assertEquals(0, num);
}
@Test
@Order (4)
void testInsert1() throws SQLException {
// 插入数据
int num = st.executeUpdate("INSERT INTO student VALUES" +
"(1, '张三', '上海')," +
"(2, '李四', '北京')," +
"(3, '王五', '成都');");
Assertions.assertEquals(3, num);
}
@Test
@Order (5)
void testInsert2() throws SQLException {
// 插入数据
int num = st.executeUpdate("INSERT INTO score VALUES" +
"(1, 645)," +
"(2, 627)," +
"(3, 591);");
Assertions.assertEquals(3, num);
}
@Test
@Order (6)
void testUpdate() throws SQLException {
// 预查询语句 删除指定 ID
pst = conn.prepareStatement("UPDATE student" +
" SET student_name = '李华', student_address = '杭州'"+
"WHERE student_id = ?");
Assertions.assertNotNull(pst);
pst.setInt(1, 1);
int num = pst.executeUpdate();
Assertions.assertEquals(1, num);
}
@Test
@Order (7)
void testDelete() throws SQLException {
// 预查询语句 删除指定 ID
pst = conn.prepareStatement("delete from student where student_id = ?");
Assertions.assertNotNull(pst);
pst.setInt(1, 3);
int num = pst.executeUpdate();
Assertions.assertEquals(1, num);
}
@Test
@Order (8)
void testSelect() throws SQLException {
rs = st.executeQuery("select * from student;");
String[] nameStrings = new String[]{"李华", "李四", "王五"};
String[] addressString = new String[]{"杭州", "北京", "成都"};
// 结果集断言
int num = 1;
while(rs.next()) {
Assertions.assertEquals(num, rs.getInt("student_id"));
Assertions.assertEquals(nameStrings[num-1], rs.getString("student_name"));
Assertions.assertEquals(addressString[num-1], rs.getString("student_address"));
num++;
}
}
@Test
@Order (9)
void testSubSelect() throws SQLException {
// 插入数据
rs = st.executeQuery(
"SELECT student_name FROM student " +
"WHERE student_id IN " +
"(SELECT student_id " +
"FROM score " +
"WHERE score > 600);"
);
String[] nameStrings = new String[]{"李华", "李四", "王五"};
// 结果集断言
int num = 1;
while(rs.next()) {
Assertions.assertEquals(nameStrings[num-1], rs.getString("student_name"));
num++;
}
}
@Test
@Order (10)
void testJoin() throws SQLException {
// 插入数据
rs = st.executeQuery(
"SELECT A.student_name " +
"FROM student A JOIN score B " +
"ON A.student_id = B.student_id " +
"WHERE score > 600;"
);
String[] nameStrings = new String[]{"李华", "李四", "王五"};
// 结果集断言
int num = 1;
while (rs.next()) {
Assertions.assertEquals(rs.getString("student_name"), nameStrings[num - 1]);
num++;
}
}
}
```
# 开发手册
## Service 添加`JDBC`类
### 实现接口
1. 以`Service` + `接口名` 命名一个类
2. 实现接口
3. 继承`BasedBind`类
4. 对于返回类属于绑定类的情况
1. 使用代理工厂返回替身类
2. 如果该类也有会返回绑定类的方法，对该类调用`setInfo(info)`
5. 为该类添加`RemoteClass(remoteClassName = Agent上对应类的全限定类名)`注解。
### 继承抽象类
1. 以`Service` + `抽象类名` 命名一个类
2. 继承抽象类
3. 粘贴所需方法和字段
```java
private String ID;
@LocalMethod
public String getID(){
return this.ID;
}
@LocalMethod
public void setID(String ID){
this.ID = ID;
}
@LocalMethod
@Override
public int hashCode(){
return super.hashCode();
}
@LocalMethod
@Override
public boolean equals(Object obj) {
return super.equals(obj);
}
@LocalMethod
@Override
protected Object clone() throws CloneNotSupportedException {
return super.clone();
}
@LocalMethod
@Override
public String toString() {
return super.toString();
}
@LocalMethod
@Override
protected void finalize() throws Throwable {
super.finalize();
}
```
4. 对于返回类属于绑定类的情况（包括抽象方法）
1. 使用代理工厂返回替身类
2. 如果该类也有会返回绑定类的方法，对该类调用`setInfo(info)`
5. 为该类添加`RemoteClass(remoteClassName = Agent上对应类的全限定类名)`注解。
## Agent 添加 `JDBC`类
### 实现接口
1. 以`Agent + 接口名`命名一个类
2. 实现接口
3. 组合一个接口类型的成员变量，创建一个入参为该接口类型的构造方法
4. 重写该接口的所有方法
1. 返回值为普通变量，直接用3中成员变量调用该方法
2. 返回值为绑定对象，返回包装对象：`new Agent+接口名(成员变量.方法(参数))`
3. 链式调用方法return this
5. 为该类添加`BindClass`注解
### 继承抽象类
```
1. 以`Agent + 继承类名`命名一个类
2. 继承类
3. 组合一个接口类型的成员变量，创建一个入参为该接口类型的构造方法
4. 重写该接口的所有方法(包括继承的抽象方法)
1. 返回值为普通变量，直接用3中成员变量调用该方法
2. 返回值为绑定对象，返回包装对象：`new Agent+接口名(成员变量.方法(参数))`
3. 链式调用方法return this
5. 为该类添加`BindClass`注解
```
# 项目参考
[nuzzle: A Simple RPC Project ](https://github.com/sakiila/nuzzle )
[CSV JDBC Driver ](https://github.com/peterborkuti/csv-jdbc-driver )
