linbin
/
tianhe
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

开发 web 端的控制通道代码中

master
linbin 2024-09-06 17:43:49 +08:00
parent b01671d7f3
commit 96eacc4b79
13 changed files with 471 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
pom.xml
View File

@ -1,25 +1,168 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>tianhe</artifactId>
<version>1.0-SNAPSHOT</version>
<packaging>jar</packaging>
<groupId>org.example</groupId>
<artifactId>tianhe</artifactId>
<version>1.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>tianhe</name>
<url>http://maven.apache.org</url>
<name>tianhe</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
</dependencies>
<dependencies>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<version>2.2.224</version>
</dependency>
<dependency>
<groupId>org.duckdb</groupId>
<artifactId>duckdb_jdbc</artifactId>
<version>0.10.3</version>
</dependency>
<dependency>
<groupId>com.jfirer</groupId>
<artifactId>JfireBoot</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.jfirer</groupId>
<artifactId>baseutil</artifactId>
<version>1.1.7-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.jfirer</groupId>
<artifactId>jfire-jsql-starter</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP-java6</artifactId>
<version>2.3.13</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.20.0</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j2-impl</artifactId>
<version>2.20.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>2.0.6</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.30</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.49</version>
</dependency>
</dependencies>
<distributionManagement>
<repository>
<id>nexus3</id>
<name>nexus</name>
<url>http://yynas.cn:8081/repository/maven-releases/</url>
</repository>
<snapshotRepository>
<id>nexus3</id>
<name>nexus</name>
<url>http://yynas.cn:8081/repository/maven-snapshots/</url>
</snapshotRepository>
</distributionManagement>
<build>
<resources>
<resource>
<directory>src/main/resources</directory>
<filtering>true</filtering>
<includes>
<include>version.yml</include>
</includes>
</resource>
<resource>
<directory>src/main/resources</directory>
<filtering>false</filtering>
<!--include下的文件会被添加到打包后的 jar 包中。filtering为 false 代表这些文件不会过滤，不会替换文件中的{}包围起来的变量。因此也就避免了二进制文件在打包的时候的错误-->
<includes>
<include>init.sql</include>
<include>log4j2.xml</include>
<include>dist/*</include>
<include>dist/assets/*</include>
</includes>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<archive>
<manifest>
<mainClass>com.ylz.DrgOperation.OperationMaster.OperationMaster</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
<version>3.2.0</version>
<executions>
<execution>
<id>timestamp-property</id>
<goals>
<goal>timestamp-property</goal>
</goals>
<configuration>
<name>buildTime</name>
<pattern>yyyy-MM-dd HH:mm:ss</pattern>
<locale>Asia/Shanghai</locale>
<timeZone>Asia/Shanghai</timeZone>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>14</source>
<target>14</target>
</configuration>
</plugin>
</plugins>
</build>
</project>

21
src/main/java/org/tianhe/common/CommandType.java 100644
View File

@ -0,0 +1,21 @@
package org.tianhe.common;
public enum CommandType
{
REGISTER_SELF_CONTROL_CHANNEL(1),//
REGISTER_SELF_DATA_CHANNEL(2),//
CONTROL_CHANNEL_HEART_BEAT(3),//
LOG_REPORT(4),
;
private byte command;
CommandType(int i)
{
command = (byte) i;
}
public byte getCommand()
{
return command;
}
}

40
src/main/java/org/tianhe/common/packet/LogReport.java 100644
View File

@ -0,0 +1,40 @@
package org.tianhe.common.packet;
import com.jfirer.jnet.common.buffer.buffer.IoBuffer;
import org.tianhe.common.CommandType;
import java.nio.charset.StandardCharsets;
public class LogReport implements Packet
{
private String date;
private String content;
@Override
public int write(IoBuffer buffer)
{
buffer.put(CommandType.LOG_REPORT.getCommand());
byte[] dateBytes = date.getBytes(StandardCharsets.UTF_8);
buffer.putShort((short) dateBytes.length);
buffer.put(dateBytes);
byte[] contentBytes = content.getBytes(StandardCharsets.UTF_8);
buffer.putInt(contentBytes.length);
buffer.put(contentBytes);
return 1 + 2 + dateBytes.length + 4 + contentBytes.length;
}
@Override
public Packet read(IoBuffer buffer)
{
short len = buffer.getShort();
byte[] bytes = new byte[len];
buffer.get(bytes);
date = new String(bytes, StandardCharsets.UTF_8);
int contentLen = buffer.getInt();
bytes = new byte[contentLen];
buffer.get(bytes);
content = new String(bytes, StandardCharsets.UTF_8);
return this;
}
}

10
src/main/java/org/tianhe/common/packet/Packet.java 100644
View File

@ -0,0 +1,10 @@
package org.tianhe.common.packet;
import com.jfirer.jnet.common.buffer.buffer.IoBuffer;
public interface Packet
{
int write(IoBuffer buffer);
Packet read(IoBuffer buffer);
}

36
src/main/java/org/tianhe/common/packet/RegisterSelfControlChannel.java 100644
View File

@ -0,0 +1,36 @@
package org.tianhe.common.packet;
import com.jfirer.jnet.common.buffer.buffer.IoBuffer;
import lombok.Data;
import org.tianhe.common.CommandType;
import java.nio.charset.StandardCharsets;
@Data
public class RegisterSelfControlChannel implements Packet
{
private String hosId;
private byte agentId;
public int write(IoBuffer buffer)
{
buffer.put(CommandType.REGISTER_SELF_CONTROL_CHANNEL.getCommand());
byte[] bytes = hosId.getBytes(StandardCharsets.UTF_8);
buffer.putShort((short) bytes.length);
buffer.put(bytes);
buffer.put(agentId);
return 1 + 2 + bytes.length + 1;
}
@Override
public RegisterSelfControlChannel read(IoBuffer buffer)
{
short len = buffer.getShort();
byte[] bytes = new byte[len];
buffer.get(bytes);
RegisterSelfControlChannel registerSelfControlChannel = new RegisterSelfControlChannel();
registerSelfControlChannel.setHosId(new String(bytes, StandardCharsets.UTF_8));
registerSelfControlChannel.setAgentId(buffer.get());
return this;
}
}

6
src/main/java/org/tianhe/web/service/LogReportService.java 100644
View File

@ -0,0 +1,6 @@
package org.tianhe.web.service;
public interface LogReportService
{
void report(String content, String date);
}

19
src/main/java/org/tianhe/web/tcp/ControlChannelStruct.java 100644
View File

@ -0,0 +1,19 @@
package org.tianhe.web.tcp;
import com.jfirer.jnet.common.api.Pipeline;
import lombok.Data;
import java.time.LocalDateTime;
@Data
public class ControlChannelStruct
{
private final Pipeline pipeline;
private volatile LocalDateTime lastActiveTime;
public ControlChannelStruct(Pipeline pipeline, LocalDateTime lastActiveTime)
{
this.pipeline = pipeline;
this.lastActiveTime = lastActiveTime;
}
}

28
src/main/java/org/tianhe/web/tcp/HeartBeatHandler.java 100644
View File

@ -0,0 +1,28 @@
package org.tianhe.web.tcp;
import com.jfirer.jnet.common.api.ReadProcessor;
import com.jfirer.jnet.common.api.ReadProcessorNode;
import com.jfirer.jnet.common.buffer.buffer.IoBuffer;
import org.tianhe.common.CommandType;
import java.time.LocalDateTime;
public class HeartBeatHandler implements ReadProcessor<IoBuffer>
{
private ControlChannelStruct controlChannelStruct;
@Override
public void read(IoBuffer data, ReadProcessorNode next)
{
byte b = data.get();
if (b == CommandType.CONTROL_CHANNEL_HEART_BEAT.getCommand())
{
controlChannelStruct.setLastActiveTime(LocalDateTime.now());
}
else
{
data.addReadPosi(-1);
next.fireRead(data);
}
}
}

27
src/main/java/org/tianhe/web/tcp/LogReportHandler.java 100644
View File

@ -0,0 +1,27 @@
package org.tianhe.web.tcp;
import com.jfirer.jnet.common.api.ReadProcessor;
import com.jfirer.jnet.common.api.ReadProcessorNode;
import com.jfirer.jnet.common.buffer.buffer.IoBuffer;
import org.tianhe.common.CommandType;
import org.tianhe.common.packet.LogReport;
import org.tianhe.web.service.LogReportService;
public class LogReportHandler implements ReadProcessor<IoBuffer>
{
private LogReportService logReportService;
@Override
public void read(IoBuffer data, ReadProcessorNode next)
{
if (data.get() == CommandType.LOG_REPORT.getCommand())
{
LogReport read = (LogReport) new LogReport().read(data);
}
else
{
data.addReadPosi(-1);
next.fireRead(data);
}
}
}

48
src/main/java/org/tianhe/web/tcp/RegisterSelfControlChannelHandler.java 100644
View File

@ -0,0 +1,48 @@
package org.tianhe.web.tcp;
import com.jfirer.jnet.common.api.ReadProcessor;
import com.jfirer.jnet.common.api.ReadProcessorNode;
import com.jfirer.jnet.common.buffer.buffer.IoBuffer;
import lombok.extern.slf4j.Slf4j;
import org.tianhe.common.CommandType;
import org.tianhe.common.packet.RegisterSelfControlChannel;
import java.time.LocalDateTime;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
public class RegisterSelfControlChannelHandler implements ReadProcessor<IoBuffer>
{
private final ConcurrentHashMap<String, ControlChannelStruct> control_channels;
private boolean unDefined = true;
private final HeartBeatHandler heartBeatHandler;
public RegisterSelfControlChannelHandler(ConcurrentHashMap<String, ControlChannelStruct> controlChannels, HeartBeatHandler heartBeatHandler)
{
control_channels = controlChannels;
this.heartBeatHandler = heartBeatHandler;
}
@Override
public void read(IoBuffer data, ReadProcessorNode next)
{
if (unDefined)
{
if (data.get() == CommandType.REGISTER_SELF_CONTROL_CHANNEL.getCommand())
{
unDefined = false;
RegisterSelfControlChannel from =new RegisterSelfControlChannel().read(data);
control_channels.put(from.getHosId() + "_" + from.getAgentId(), new ControlChannelStruct(next.pipeline(), LocalDateTime.now()));
}
else
{
log.error("未能识别的初始命令，关闭连接");
next.pipeline().channelContext().close();
}
}
else
{
next.fireRead(data);
}
}
}

38
src/main/java/org/tianhe/web/tcp/WerbTcpServer.java 100644
View File

@ -0,0 +1,38 @@
package org.tianhe.web.tcp;
import com.jfirer.jfire.core.ApplicationContext;
import com.jfirer.jfire.core.AwareContextInited;
import com.jfirer.jnet.common.api.Pipeline;
import com.jfirer.jnet.common.decoder.TotalLengthFieldBasedFrameDecoder;
import com.jfirer.jnet.common.util.ChannelConfig;
import com.jfirer.jnet.server.AioServer;
import lombok.extern.slf4j.Slf4j;
import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;
@Resource
@Slf4j
public class WerbTcpServer implements AwareContextInited
{
@Override
public void aware(ApplicationContext applicationContext)
{
ChannelConfig channelConfig = new ChannelConfig();
channelConfig.setMsOfReadTimeout(1000*30);
channelConfig.setWorkerGroup(ChannelConfig.DEFAULT_WORKER_GROUP);
channelConfig.setChannelGroup(ChannelConfig.DEFAULT_CHANNEL_GROUP);
channelConfig.setPort(39999);
ConcurrentHashMap<String,ControlChannelStruct> controlChannels = new ConcurrentHashMap<>();
AioServer aioServer = AioServer.newAioServer(channelConfig, channelContext -> {
Pipeline pipeline = channelContext.pipeline();
pipeline.addReadProcessor(new TotalLengthFieldBasedFrameDecoder(0, 4, 4, Integer.MAX_VALUE, channelConfig.getAllocator()));
HeartBeatHandler heartBeatHandler = new HeartBeatHandler();
RegisterSelfControlChannelHandler registerSelfControlChannelHandler = new RegisterSelfControlChannelHandler(controlChannels, heartBeatHandler);
pipeline.addReadProcessor(registerSelfControlChannelHandler);
pipeline.addReadProcessor(heartBeatHandler);
});
aioServer.start();
log.debug("运维服务器穿透模块启动");
}
}

35
src/main/resources/log4j2.xml 100644
View File

@ -0,0 +1,35 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration agentStatus="ERROR">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout
pattern="%-5level %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] - %X{traceId} %-80msg %l{%n%ex}%n"/>
</Console>
<RollingRandomAccessFile name="AllFile"
fileName="target/logs/Operation/app.log"
filePattern="target/logs/Operation/$${date:yyyy-MM}/app_%d{yyyy-MM-dd}_%i.log.gz">
<PatternLayout
pattern="%-5level %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] - %X{traceId} %-80msg %l{%n%ex}%n"/>
<Policies>
<TimeBasedTriggeringPolicy/>
<SizeBasedTriggeringPolicy size="20 MB"/>
</Policies>
<DefaultRolloverStrategy max="20"/>
</RollingRandomAccessFile>
</Appenders>
<Loggers>
<Root level="error">
<AppenderRef ref="AllFile"/>
</Root>
<Logger name="com.ylz" level="debug" additivity="false">
<AppenderRef ref="Console"/>
<AppenderRef ref="AllFile"/>
</Logger>
<Logger name="com.jfirer" level="info" additivity="false">
<AppenderRef ref="Console"/>
<AppenderRef ref="AllFile"/>
</Logger>
</Loggers>
</Configuration>

1
src/main/resources/version.yml 100644
View File

@ -0,0 +1 @@
version: ${buildTime}