qq27800734/taoyao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[*] 日常优化

Browse Source
This commit is contained in:
acgist
2023-07-20 08:32:14 +08:00
parent 2c2449fd3a
commit eb84cc3d8d
11 changed files with 627 additions and 585 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
taoyao-signal-server/taoyao-boot/src/main/java/com/acgist/taoyao/boot/utils/CloseableUtils.java
View File

@@ -1,6 +1,7 @@
package com.acgist.taoyao.boot.utils; package com.acgist.taoyao.boot.utils;
import java.io.Closeable; import java.io.Closeable;
import java.nio.channels.AsynchronousChannelGroup;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@@ -45,4 +46,19 @@ public final class CloseableUtils {
} }
} }
/**
* 关闭通道线程池
*
* @param group 通道线程池
*/
public static final void shutdown(AsynchronousChannelGroup group) {
try {
if(group != null) {
group.shutdown();
}
} catch (Exception e) {
log.error("关闭通道线程池异常", e);
}
}
} }

11
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/ClientAdapter.java
View File

@@ -58,6 +58,7 @@ public abstract class ClientAdapter<T extends AutoCloseable> implements Client {
* @param instance 终端实例 * @param instance 终端实例
*/ */
protected ClientAdapter(long timeout, T instance) { protected ClientAdapter(long timeout, T instance) {
this.ip = this.getClientIP(instance);
this.time = System.currentTimeMillis(); this.time = System.currentTimeMillis();
this.timeout = timeout; this.timeout = timeout;
this.instance = instance; this.instance = instance;
@@ -151,9 +152,19 @@ public abstract class ClientAdapter<T extends AutoCloseable> implements Client {
@Override @Override
public void close() throws Exception { public void close() throws Exception {
log.info("关闭终端实例：{} - {}", this.ip, this.clientId);
this.instance.close(); this.instance.close();
} }
/**
* 解析终端IP
*
* @param instance 终端实例
*
* @return 终端IP
*/
protected abstract String getClientIP(T instance);
@Override @Override
public String toString() { public String toString() {
return this.getClass().getSimpleName() + " - " + this.ip + " - " + this.clientId; return this.getClass().getSimpleName() + " - " + this.ip + " - " + this.clientId;

408
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/ClientManager.java
View File

@@ -22,214 +22,212 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j @Slf4j
@Manager @Manager
public class ClientManager { public class ClientManager {
private final ApplicationContext applicationContext; private final ApplicationContext applicationContext;
/** /**
* 终端列表 * 终端列表
*/ */
private final List<Client> clients; private final List<Client> clients;
public ClientManager(ApplicationContext applicationContext) { public ClientManager(ApplicationContext applicationContext) {
this.applicationContext = applicationContext; this.applicationContext = applicationContext;
this.clients = new CopyOnWriteArrayList<>(); this.clients = new CopyOnWriteArrayList<>();
}
@Scheduled(cron = "${taoyao.scheduled.client:0 * * * * ?}")
public void scheduled() {
this.closeTimeout();
}
/**
* 终端打开加入管理
*
* @param client 终端
*/
public void open(Client client) {
this.clients.add(client);
}
/**
* 授权终端单播消息
*
* @param to 接收终端
* @param message 消息
*/
public void unicast(String to, Message message) {
this.clients().stream()
.filter(v -> Objects.equals(to, v.getClientId()))
.forEach(v -> v.push(message));
}
/**
* 授权终端单播消息
*
* @param to 接收终端
* @param message 消息
*/
public void unicast(Client to, Message message) {
this.clients().stream()
.filter(v -> v == to)
.forEach(v -> v.push(message));
}
/**
* 授权终端广播消息
*
* @param message 消息
* @param clientTypes 终端类型
*/
public void broadcast(Message message, ClientType ... clientTypes) {
this.clients(clientTypes).forEach(v -> v.push(message));
}
/**
* 授权终端广播消息
*
* @param from 发送终端
* @param message 消息
* @param clientTypes 终端类型
*/
public void broadcast(String from, Message message, ClientType ... clientTypes) {
this.clients(clientTypes).stream()
.filter(v -> !Objects.equals(from, v.getClientId()))
.forEach(v -> v.push(message));
}
/**
* 授权终端广播消息
*
* @param from 发送终端
* @param message 消息
* @param clientTypes 终端类型
*/
public void broadcast(Client from, Message message, ClientType ... clientTypes) {
this.clients(clientTypes).stream()
.filter(v -> v != from)
.forEach(v -> v.push(message));
}
/**
* @param instance 终端实例
*
* @return 终端（包含授权和未授权）
*/
public Client clients(AutoCloseable instance) {
return this.clients.stream()
.filter(v -> v.getInstance() == instance)
.findFirst()
.orElse(null);
}
/**
* @param clientId 终端ID
*
* @return 授权终端
*/
public Client clients(String clientId) {
return this.clients().stream()
.filter(v -> Objects.equals(clientId, v.getClientId()))
.findFirst()
.orElse(null);
}
/**
* @param clientTypes 终端类型
*
* @return 授权终端列表
*/
public List<Client> clients(ClientType ... clientTypes) {
return this.clients.stream()
.filter(Client::authorized)
.filter(client -> ArrayUtils.isEmpty(clientTypes) || ArrayUtils.contains(clientTypes, client.getClientType()))
.toList();
}
/**
* @param instance 终端实例
*
* @return 终端状态
*/
public ClientStatus status(AutoCloseable instance) {
final Client client = this.clients(instance);
return client == null ? null : client.getStatus();
}
/**
* @param clientId 终端ID
*
* @return 授权终端状态
*/
public ClientStatus status(String clientId) {
final Client client = this.clients(clientId);
return client == null ? null : client.getStatus();
} }
@Scheduled(cron = "${taoyao.scheduled.client:0 * * * * ?}")
public void scheduled() {
this.closeTimeout();
}
/**
* 终端打开加入管理
*
* @param client 终端
*/
public void open(Client client) {
this.clients.add(client);
}
/**
* 授权终端单播消息
*
* @param to 接收终端
* @param message 消息
*/
public void unicast(String to, Message message) {
this.clients().stream()
.filter(v -> Objects.equals(to, v.getClientId()))
.forEach(v -> v.push(message));
}
/**
* 授权终端单播消息
*
* @param to 接收终端
* @param message 消息
*/
public void unicast(Client to, Message message) {
this.clients().stream()
.filter(v -> v.getInstance() == to)
.forEach(v -> v.push(message));
}
/**
* 授权终端广播消息
*
* @param message 消息
* @param clientTypes 终端类型
*/
public void broadcast(Message message, ClientType ... clientTypes) {
this.clients(clientTypes).forEach(v -> v.push(message));
}
/**
* 授权终端广播消息
*
* @param from 发送终端
* @param message 消息
* @param clientTypes 终端类型
*/
public void broadcast(String from, Message message, ClientType ... clientTypes) {
this.clients(clientTypes).stream()
.filter(v -> !Objects.equals(from, v.getClientId()))
.forEach(v -> v.push(message));
}
/**
* 授权终端广播消息
*
* @param from 发送终端
* @param message 消息
* @param clientTypes 终端类型
*/
public void broadcast(Client from, Message message, ClientType ... clientTypes) {
this.clients(clientTypes).stream()
.filter(v -> v.getInstance() != from)
.forEach(v -> v.push(message));
}
/**
* @param instance 终端实例
*
* @return 终端
*/
public Client clients(AutoCloseable instance) {
return this.clients.stream()
.filter(v -> v.getInstance() == instance)
.findFirst()
.orElse(null);
}
/**
* @param clientId 终端标识
*
* @return 授权终端
*/
public Client clients(String clientId) {
return this.clients().stream()
.filter(v -> Objects.equals(clientId, v.getClientId()))
.findFirst()
.orElse(null);
}
/**
* @param clientTypes 终端类型
*
* @return 所有授权终端列表
*/
public List<Client> clients(ClientType ... clientTypes) {
return this.clients.stream()
.filter(Client::authorized)
.filter(client -> ArrayUtils.isEmpty(clientTypes) || ArrayUtils.contains(clientTypes, client.getClientType()))
.toList();
}
/**
* @param instance 终端实例
*
* @return 终端状态
*/
public ClientStatus status(AutoCloseable instance) {
final Client client = this.clients(instance);
return client == null ? null : client.getStatus();
}
/**
* @param clientId 终端标识
*
* @return 授权终端状态
*/
public ClientStatus status(String clientId) {
final Client client = this.clients(clientId);
return client == null ? null : client.getStatus();
}
/** /**
* @param clientTypes 终端类型 * @param clientTypes 终端类型
* *
* @return 所有授权终端状态列表 * @return 授权终端状态列表
*/ */
public List<ClientStatus> status(ClientType ... clientTypes) { public List<ClientStatus> status(ClientType ... clientTypes) {
return this.clients(clientTypes).stream() return this.clients(clientTypes).stream()
.map(Client::getStatus) .map(Client::getStatus)
.toList(); .toList();
} }
/** /**
* 推送消息 * 推送消息
* *
* @param instance 终端实例 * @param instance 终端实例
* @param message 消息 * @param message 消息
*/ */
public void push(AutoCloseable instance, Message message) { public void push(AutoCloseable instance, Message message) {
final Client client = this.clients(instance); final Client client = this.clients(instance);
if(client == null) { if(client == null) {
log.warn("推送消息终端无效：{}-{}", instance, message); log.warn("推送消息终端无效：{} - {}", instance, message);
return; return;
} }
client.push(message); client.push(message);
} }
/** /**
* 关闭终端 * 关闭终端
* *
* @param instance 终端实例 * @param instance 终端实例
*/ */
public void close(AutoCloseable instance) { public void close(AutoCloseable instance) {
final Client client = this.clients(instance); final Client client = this.clients(instance);
try { try {
if(client != null) { if(client != null) {
client.close(); client.close();
} else { } else {
instance.close(); instance.close();
} }
} catch (Exception e) { } catch (Exception e) {
log.error("关闭终端异常：{}", instance, e); log.error("关闭终端异常：{}", instance, e);
} finally { } finally {
if(client != null) { if(client != null) {
// 移除管理 this.clients.remove(client);
this.clients.remove(client); this.applicationContext.publishEvent(new ClientCloseEvent(client));
// 关闭事件 }
this.applicationContext.publishEvent(new ClientCloseEvent(client)); }
} }
}
} /**
* 定时关闭超时终端
/** */
* 定时关闭超时终端 private void closeTimeout() {
*/ final int oldSize = this.clients.size();
private void closeTimeout() { this.clients.stream()
final int oldSize = this.clients.size(); .filter(v -> v.unauthorized())
this.clients.stream() .filter(v -> v.timeout())
.filter(v -> v.unauthorized()) .forEach(v -> {
.filter(v -> v.timeout()) log.debug("关闭超时终端：{}", v);
.forEach(v -> { this.close(v);
log.debug("关闭超时终端：{}", v); });
this.close(v); final int newSize = this.clients.size();
}); log.debug("定时关闭超时终端：{}", newSize - oldSize);
final int newSize = this.clients.size(); }
log.debug("定时关闭超时终端：{}", newSize - oldSize);
}
} }

102
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/ClientStatus.java
View File

@@ -37,31 +37,31 @@ public class ClientStatus {
private Double humidity; private Double humidity;
@Schema(title = "温度", description = "温度") @Schema(title = "温度", description = "温度")
private Double temperature; private Double temperature;
@Schema(title = "信号强度（0~100）", description = "信号强度（0~100）") @Schema(title = "信号强度（0~100）", description = "信号强度（0~100）")
private Integer signal; private Integer signal;
@Schema(title = "电池电量（0~100）", description = "电池电量（0~100）") @Schema(title = "电池电量（0~100）", description = "电池电量（0~100）")
private Integer battery; private Integer battery;
@Schema(title = "是否发生告警", description = "是否发生告警") @Schema(title = "是否发生告警", description = "是否发生告警")
private Boolean alarming; private Boolean alarming;
@Schema(title = "是否正在充电", description = "是否正在充电") @Schema(title = "是否正在充电", description = "是否正在充电")
private Boolean charging; private Boolean charging;
@Schema(title = "终端是否正在录像", description = "终端是否正在录像") @Schema(title = "终端是否正在录像", description = "终端是否正在录像")
private Boolean clientRecording; private Boolean clientRecording;
@Schema(title = "服务端是否正在录像", description = "服务端是否正在录像") @Schema(title = "服务端是否正在录像", description = "服务端是否正在录像")
private Boolean serverRecording; private Boolean serverRecording;
@Schema(title = "最后心跳时间", description = "最后心跳时间") @Schema(title = "终端状态", description = "其他扩展终端状态")
private LocalDateTime lastHeartbeat; private Map<String, Object> status = new HashMap<>();
@Schema(title = "终端状态", description = "其他扩展终端状态") @Schema(title = "终端配置", description = "其他扩展终端配置")
private Map<String, Object> status = new HashMap<>(); private Map<String, Object> config = new HashMap<>();
@Schema(title = "终端配置", description = "其他扩展终端配置") @Schema(title = "最后心跳时间", description = "最后心跳时间")
private Map<String, Object> config = new HashMap<>(); private LocalDateTime lastHeartbeat;
/** /**
* 拷贝属性 * 拷贝属性
* *
* @param body 消息主体 * @param body 消息主体
*/ */
public void copy(Map<String, Object> body) { public void copy(Map<String, Object> body) {
this.setLatitude(MapUtils.getDouble(body, Constant.LATITUDE)); this.setLatitude(MapUtils.getDouble(body, Constant.LATITUDE));
this.setLongitude(MapUtils.getDouble(body, Constant.LONGITUDE)); this.setLongitude(MapUtils.getDouble(body, Constant.LONGITUDE));
this.setHumidity(MapUtils.getDouble(body, Constant.HUMIDITY)); this.setHumidity(MapUtils.getDouble(body, Constant.HUMIDITY));
@@ -74,30 +74,30 @@ public class ClientStatus {
this.status(MapUtils.get(body, Constant.STATUS)); this.status(MapUtils.get(body, Constant.STATUS));
this.config(MapUtils.get(body, Constant.CONFIG)); this.config(MapUtils.get(body, Constant.CONFIG));
this.setLastHeartbeat(LocalDateTime.now()); this.setLastHeartbeat(LocalDateTime.now());
} }
/** /**
* 拷贝状态 * 拷贝状态
* *
* @param map 状态 * @param map 状态
*/ */
public void status(Map<String, Object> map) { public void status(Map<String, Object> map) {
if(map == null) { if(map == null) {
return; return;
} }
map.forEach(this.status::put); map.forEach(this.status::put);
} }
/** /**
* 拷贝配置 * 拷贝配置
* *
* @param map 配置 * @param map 配置
*/ */
public void config(Map<String, Object> map) { public void config(Map<String, Object> map) {
if(map == null) { if(map == null) {
return; return;
} }
map.forEach(this.config::put); map.forEach(this.config::put);
} }
} }

36
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/ClientType.java
View File

@@ -14,10 +14,25 @@ import lombok.Getter;
@Getter @Getter
public enum ClientType { public enum ClientType {
/**
* 通过浏览器接入的终端
*/
WEB("Web"), WEB("Web"),
/**
* 媒体服务终端
*/
MEDIA("媒体服务"), MEDIA("媒体服务"),
/**
* 没有界面的摄像头
*/
CAMERA("摄像头"), CAMERA("摄像头"),
/**
* 手机APP、平板APP
*/
MOBILE("移动端"), MOBILE("移动端"),
/**
* 其他智能终端
*/
OTHER("其他终端"); OTHER("其他终端");
/** /**
@@ -33,7 +48,10 @@ public enum ClientType {
* @return 是否是媒体终端 * @return 是否是媒体终端
*/ */
public boolean mediaClient() { public boolean mediaClient() {
return this == WEB || this == CAMERA || this == MOBILE; return
this == WEB ||
this == CAMERA ||
this == MOBILE;
} }
/** /**
@@ -49,7 +67,8 @@ public enum ClientType {
* @return 类型 * @return 类型
*/ */
public static final ClientType of(String value) { public static final ClientType of(String value) {
for (ClientType type : ClientType.values()) { final ClientType[] types = ClientType.values();
for (ClientType type : types) {
if(type.name().equalsIgnoreCase(value)) { if(type.name().equalsIgnoreCase(value)) {
return type; return type;
} }
@@ -58,12 +77,17 @@ public enum ClientType {
} }
/** /**
* 媒体终端 * 媒体终端类型列表
*/ */
public static final ClientType[] MEDIA_CLIENT = Stream.of(ClientType.values()).filter(ClientType::mediaClient).toArray(ClientType[]::new); public static final ClientType[] MEDIA_CLIENT_TYPE
=
Stream.of(ClientType.values()).filter(ClientType::mediaClient).toArray(ClientType[]::new);
/** /**
* 媒体服务 * 媒体服务类型列表
*/ */
public static final ClientType[] MEDIA_SERVER = Stream.of(ClientType.values()).filter(ClientType::mediaServer).toArray(ClientType[]::new); public static final ClientType[] MEDIA_SERVER_TYPE
=
Stream.of(ClientType.values()).filter(ClientType::mediaServer).toArray(ClientType[]::new);
} }

89
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/socket/SocketClient.java
View File

@@ -36,55 +36,50 @@ public class SocketClient extends ClientAdapter<AsynchronousSocketChannel> {
*/ */
private final Cipher cipher; private final Cipher cipher;
public SocketClient(SocketProperties socketProperties, AsynchronousSocketChannel instance) { public SocketClient(SocketProperties socketProperties, AsynchronousSocketChannel instance) {
super(socketProperties.getTimeout(), instance); super(socketProperties.getTimeout(), instance);
this.ip = this.clientIp(instance); this.cipher = CipherUtils.buildCipher(Cipher.ENCRYPT_MODE, socketProperties.getEncrypt(), socketProperties.getEncryptSecret());
this.cipher = CipherUtils.buildCipher(Cipher.ENCRYPT_MODE, socketProperties.getEncrypt(), socketProperties.getEncryptSecret()); }
}
@Override @Override
public void push(Message message) { public void push(Message message) {
synchronized (this.instance) { synchronized (this.instance) {
try { try {
if(this.instance.isOpen()) { if(this.instance.isOpen()) {
// 加密 // 加密
final byte[] bytes = this.encrypt(message); final byte[] bytes = this.encrypt(message);
// 发送 // 发送
final ByteBuffer buffer = ByteBuffer.allocateDirect(Short.BYTES + bytes.length); final ByteBuffer buffer = ByteBuffer.allocateDirect(Short.BYTES + bytes.length);
buffer.putShort((short) bytes.length); buffer.putShort((short) bytes.length);
buffer.put(bytes); buffer.put(bytes);
buffer.flip(); buffer.flip();
final Future<Integer> future = this.instance.write(buffer); final Future<Integer> future = this.instance.write(buffer);
future.get(this.timeout, TimeUnit.MILLISECONDS); future.get(this.timeout, TimeUnit.MILLISECONDS);
} else { } else {
log.error("Socket终端已经关闭：{}", this.instance); log.error("Socket终端已经关闭：{}", this.instance);
} }
} catch (Exception e) { } catch (Exception e) {
log.error("Socket终端发送消息异常：{}", message, e); log.error("Socket终端发送消息异常：{}", message, e);
} }
} }
} }
/** @Override
* @param instance 终端实例 protected String getClientIP(AsynchronousSocketChannel instance) {
* try {
* @return 终端IP
*/
private String clientIp(AsynchronousSocketChannel instance) {
try {
return ((InetSocketAddress) instance.getRemoteAddress()).getHostString(); return ((InetSocketAddress) instance.getRemoteAddress()).getHostString();
} catch (IOException e) { } catch (IOException e) {
throw MessageCodeException.of(e, "无效终端（IP）：" + instance); throw MessageCodeException.of(e, "无效终端IP：" + instance);
} }
} }
/** /**
* @param message 消息 * @param message 消息
* *
* @return 加密消息 * @return 加密消息
*/ */
private byte[] encrypt(Message message) { private byte[] encrypt(Message message) {
final byte[] bytes = message.toString().getBytes(); final byte[] bytes = message.toString().getBytes();
if(this.cipher != null) { if(this.cipher != null) {
try { try {
return this.cipher.doFinal(bytes); return this.cipher.doFinal(bytes);
@@ -93,6 +88,6 @@ public class SocketClient extends ClientAdapter<AsynchronousSocketChannel> {
} }
} }
return bytes; return bytes;
} }
} }

190
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/socket/SocketSignal.java
View File

@@ -22,106 +22,102 @@ import lombok.extern.slf4j.Slf4j;
/** /**
* Socket信令 * Socket信令
* *
* TODO：加密
*
* @author acgist * @author acgist
*/ */
@Slf4j @Slf4j
public class SocketSignal { public class SocketSignal {
private ClientManager clientManager; private final ClientManager clientManager;
private ProtocolManager protocolManager; private final ProtocolManager protocolManager;
private SocketProperties socketProperties; private final SocketProperties socketProperties;
private PlatformErrorProtocol platformErrorProtocol; private final PlatformErrorProtocol platformErrorProtocol;
/** /**
* 线程序号 * 线程序号
*/ */
private int index = 0; private int index = 0;
/** /**
* 通道线程池 * 通道线程池
*/ */
private AsynchronousChannelGroup group; private AsynchronousChannelGroup group;
/** /**
* 服务端通道 * 服务端通道
*/ */
private AsynchronousServerSocketChannel channel; private AsynchronousServerSocketChannel server;
public SocketSignal( public SocketSignal(
ClientManager clientManager, ClientManager clientManager,
ProtocolManager protocolManager, ProtocolManager protocolManager,
SocketProperties socketProperties, SocketProperties socketProperties,
PlatformErrorProtocol platformErrorProtocol PlatformErrorProtocol platformErrorProtocol
) { ) {
this.clientManager = clientManager; this.clientManager = clientManager;
this.protocolManager = protocolManager; this.protocolManager = protocolManager;
this.socketProperties = socketProperties; this.socketProperties = socketProperties;
this.platformErrorProtocol = platformErrorProtocol; this.platformErrorProtocol = platformErrorProtocol;
} }
/** /**
* 初始化服务端 * 初始化服务端
*/ */
public void init() { public void init() {
boolean success = true; boolean success = true;
final String host = this.socketProperties.getHost(); final String host = this.socketProperties.getHost();
final Integer port = this.socketProperties.getPort(); final Integer port = this.socketProperties.getPort();
try { try {
final ExecutorService executor = new ThreadPoolExecutor( final ExecutorService executor = new ThreadPoolExecutor(
this.socketProperties.getMinThread(), this.socketProperties.getMinThread(),
this.socketProperties.getMaxThread(), this.socketProperties.getMaxThread(),
this.socketProperties.getKeepAliveTime(), this.socketProperties.getKeepAliveTime(),
TimeUnit.MILLISECONDS, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(this.socketProperties.getQueueSize()), new LinkedBlockingQueue<>(this.socketProperties.getQueueSize()),
this.newThreadFactory() this.newThreadFactory()
); );
this.group = AsynchronousChannelGroup.withThreadPool(executor); this.group = AsynchronousChannelGroup.withThreadPool(executor);
this.channel = AsynchronousServerSocketChannel.open(this.group); this.server = AsynchronousServerSocketChannel.open(this.group);
this.channel.bind(new InetSocketAddress(host, port)); this.server.bind(new InetSocketAddress(host, port));
this.channel.accept(this.channel, new SocketSignalAcceptHandler( this.server.accept(this.server, new SocketSignalAcceptHandler(
this.clientManager, this.clientManager,
this.protocolManager, this.protocolManager,
this.socketProperties, this.socketProperties,
this.platformErrorProtocol this.platformErrorProtocol
)); ));
} catch (IOException e) { } catch (IOException e) {
log.error("启动Socket信令服务异常", e); log.error("启动Socket信令服务异常", e);
success = false; success = false;
} finally { } finally {
if(success) { if(success) {
log.info("启动Socket信令服务成功：{}-{}", host, port); log.info("启动Socket信令服务成功：{} - {}", host, port);
} else { } else {
this.destroy(); this.destroy();
} }
} }
} }
/** /**
* @return 线程池工厂 * @return 线程池工厂
*/ */
private ThreadFactory newThreadFactory() { private ThreadFactory newThreadFactory() {
return (runnable) -> { return (runnable) -> {
final Thread thread = new Thread(runnable); final Thread thread = new Thread(runnable);
// 线程名称 // 线程名称
synchronized(this) { synchronized(this) {
if(++this.index > this.socketProperties.getMaxThread()) { if(++this.index > this.socketProperties.getMaxThread()) {
this.index = 0; this.index = 0;
} }
thread.setName(this.socketProperties.getThreadNamePrefix() + this.index); thread.setName(this.socketProperties.getThreadNamePrefix() + this.index);
} }
// 守护线程 // 守护线程
thread.setDaemon(true); thread.setDaemon(true);
return thread; return thread;
}; };
} }
@PreDestroy @PreDestroy
public void destroy() { public void destroy() {
log.debug("关闭Socket信令服务：{}", this.channel); log.debug("关闭Socket信令服务：{}", this.server);
CloseableUtils.close(this.channel); CloseableUtils.close(this.server);
if(this.group != null) { CloseableUtils.shutdown(this.group);
this.group.shutdown(); }
}
}
} }

84
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/socket/SocketSignalAcceptHandler.java
View File

@@ -21,47 +21,47 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j @Slf4j
public final class SocketSignalAcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel> { public final class SocketSignalAcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel> {
private ClientManager clientManager; private final ClientManager clientManager;
private ProtocolManager protocolManager; private final ProtocolManager protocolManager;
private SocketProperties socketProperties; private final SocketProperties socketProperties;
private PlatformErrorProtocol platformErrorProtocol; private final PlatformErrorProtocol platformErrorProtocol;
public SocketSignalAcceptHandler( public SocketSignalAcceptHandler(
ClientManager clientManager, ClientManager clientManager,
ProtocolManager protocolManager, ProtocolManager protocolManager,
SocketProperties socketProperties, SocketProperties socketProperties,
PlatformErrorProtocol platformErrorProtocol PlatformErrorProtocol platformErrorProtocol
) { ) {
this.clientManager = clientManager; this.clientManager = clientManager;
this.protocolManager = protocolManager; this.protocolManager = protocolManager;
this.socketProperties = socketProperties; this.socketProperties = socketProperties;
this.platformErrorProtocol = platformErrorProtocol; this.platformErrorProtocol = platformErrorProtocol;
} }
@Override @Override
public void completed(AsynchronousSocketChannel channel, AsynchronousServerSocketChannel server) { public void completed(AsynchronousSocketChannel channel, AsynchronousServerSocketChannel server) {
try { try {
channel.setOption(StandardSocketOptions.SO_KEEPALIVE, Boolean.TRUE); channel.setOption(StandardSocketOptions.SO_KEEPALIVE, Boolean.TRUE);
this.clientManager.open(new SocketClient(this.socketProperties, channel)); this.clientManager.open(new SocketClient(this.socketProperties, channel));
final SocketSignalMessageHandler messageHandler = new SocketSignalMessageHandler( final SocketSignalMessageHandler messageHandler = new SocketSignalMessageHandler(
this.clientManager, this.clientManager,
this.protocolManager, this.protocolManager,
this.socketProperties, this.socketProperties,
channel, this.platformErrorProtocol,
this.platformErrorProtocol channel
); );
messageHandler.loopMessage(); messageHandler.loopMessage();
log.debug("Socket信令终端连接成功：{}", channel); log.debug("Socket信令终端连接成功：{}", channel);
} catch (IOException e) { } catch (IOException e) {
log.error("Socket信令终端连接异常：", channel, e); log.error("Socket信令终端连接异常：", channel, e);
} finally { } finally {
server.accept(server, this); server.accept(server, this);
} }
} }
@Override @Override
public void failed(Throwable throwable, AsynchronousServerSocketChannel server) { public void failed(Throwable throwable, AsynchronousServerSocketChannel server) {
log.error("Socket信令终端连接异常：{}", server, throwable); log.error("Socket信令终端连接异常：{}", server, throwable);
} }
} }

268
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/socket/SocketSignalMessageHandler.java
View File

@@ -10,7 +10,6 @@ import javax.crypto.IllegalBlockSizeException;
import com.acgist.taoyao.boot.config.SocketProperties; import com.acgist.taoyao.boot.config.SocketProperties;
import com.acgist.taoyao.boot.model.MessageCodeException; import com.acgist.taoyao.boot.model.MessageCodeException;
import com.acgist.taoyao.boot.utils.CloseableUtils;
import com.acgist.taoyao.signal.client.ClientManager; import com.acgist.taoyao.signal.client.ClientManager;
import com.acgist.taoyao.signal.protocol.ProtocolManager; import com.acgist.taoyao.signal.protocol.ProtocolManager;
import com.acgist.taoyao.signal.protocol.platform.PlatformErrorProtocol; import com.acgist.taoyao.signal.protocol.platform.PlatformErrorProtocol;
@@ -27,134 +26,133 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j @Slf4j
public final class SocketSignalMessageHandler implements CompletionHandler<Integer, ByteBuffer> { public final class SocketSignalMessageHandler implements CompletionHandler<Integer, ByteBuffer> {
private ClientManager clientManager; private final ClientManager clientManager;
private ProtocolManager protocolManager; private final ProtocolManager protocolManager;
private PlatformErrorProtocol platformErrorProtocol; private final PlatformErrorProtocol platformErrorProtocol;
/** /**
* 消息长度 * 消息长度
*/ */
private short messageLength; private short messageLength;
/** /**
* 缓冲大小 * 缓冲大小
*/ */
private final int bufferSize; private final int bufferSize;
/** /**
* 最大缓存大小 * 最大缓存大小
*/ */
private final int maxBufferSize; private final int maxBufferSize;
/** /**
* 加密工具 * 加密工具
*/ */
private final Cipher cipher; private final Cipher cipher;
/** /**
* 消息处理 * 消息处理
*/ */
private final ByteBuffer buffer; private final ByteBuffer buffer;
/** /**
* 终端通道 * 终端通道
*/ */
private final AsynchronousSocketChannel channel; private final AsynchronousSocketChannel channel;
public SocketSignalMessageHandler( public SocketSignalMessageHandler(
ClientManager clientManager, ClientManager clientManager,
ProtocolManager protocolManager, ProtocolManager protocolManager,
SocketProperties socketProperties, SocketProperties socketProperties,
AsynchronousSocketChannel channel, PlatformErrorProtocol platformErrorProtocol,
PlatformErrorProtocol platformErrorProtocol AsynchronousSocketChannel channel
) { ) {
this.messageLength = 0; this.clientManager = clientManager;
this.bufferSize = socketProperties.getBufferSize(); this.protocolManager = protocolManager;
this.maxBufferSize = socketProperties.getMaxBufferSize(); this.platformErrorProtocol = platformErrorProtocol;
this.cipher = CipherUtils.buildCipher(Cipher.DECRYPT_MODE, socketProperties.getEncrypt(), socketProperties.getEncryptSecret()); this.channel = channel;
this.buffer = ByteBuffer.allocateDirect(maxBufferSize); this.messageLength = 0;
this.channel = channel; this.bufferSize = socketProperties.getBufferSize();
this.clientManager = clientManager; this.maxBufferSize = socketProperties.getMaxBufferSize();
this.protocolManager = protocolManager; this.cipher = CipherUtils.buildCipher(Cipher.DECRYPT_MODE, socketProperties.getEncrypt(), socketProperties.getEncryptSecret());
this.platformErrorProtocol = platformErrorProtocol; this.buffer = ByteBuffer.allocateDirect(maxBufferSize);
} }
/** /**
* 消息轮询 * 消息轮询
*/ */
public void loopMessage() { public void loopMessage() {
if(this.channel.isOpen()) { if(this.channel.isOpen()) {
final ByteBuffer buffer = ByteBuffer.allocateDirect(this.bufferSize); final ByteBuffer buffer = ByteBuffer.allocateDirect(this.bufferSize);
this.channel.read(buffer, buffer, this); this.channel.read(buffer, buffer, this);
} else { } else {
log.debug("Socket信令消息轮询退出（通道已经关闭）"); log.debug("Socket信令消息轮询退出（通道已经关闭）");
this.close(); this.close();
} }
} }
/** /**
* 关闭通道 * 关闭通道
*/ */
private void close() { private void close() {
log.debug("Socket信令终端关闭：{}", this.channel); log.debug("Socket信令终端关闭：{}", this.channel);
CloseableUtils.close(this.channel); this.clientManager.close(this.channel);
this.clientManager.close(this.channel); }
}
@Override
@Override public void completed(Integer result, ByteBuffer buffer) {
public void completed(Integer result, ByteBuffer buffer) { if (result == null || result < 0) {
if (result == null || result < 0) { log.warn("Socket信令接收消息失败关闭通道：{}", result);
log.warn("Socket信令接收消息失败关闭通道：{}", result); this.close();
this.close(); } else if(result == 0) {
} else if(result == 0) { // 消息空轮询
// 消息空轮询 log.debug("Socket信令接收消息失败（长度）：{}", result);
log.debug("Socket信令接收消息失败（长度）：{}", result); } else {
} else { buffer.flip();
buffer.flip(); this.buffer.put(buffer);
this.buffer.put(buffer); while(this.buffer.position() > 0) {
while(this.buffer.position() > 0) { if(this.messageLength <= 0) {
if(this.messageLength <= 0) { if(this.buffer.position() < Short.BYTES) {
if(this.buffer.position() < Short.BYTES) { // 不够消息长度
// 不够消息长度 break;
break; } else {
} else { this.buffer.flip();
this.buffer.flip(); this.messageLength = this.buffer.getShort();
this.messageLength = this.buffer.getShort(); this.buffer.compact();
this.buffer.compact(); if(this.messageLength < 0 || this.messageLength > this.maxBufferSize) {
if(this.messageLength < 0 || this.messageLength > this.maxBufferSize) { throw MessageCodeException.of("信令消息长度错误：" + this.messageLength);
throw MessageCodeException.of("信令消息长度错误：" + this.messageLength); }
} }
} } else {
} else { if(this.buffer.position() < this.messageLength) {
if(this.buffer.position() < this.messageLength) { // 不够消息长度
// 不够消息长度 break;
break; } else {
} else { // 拆包
// 拆包 final byte[] bytes = new byte[this.messageLength];
final byte[] bytes = new byte[this.messageLength]; this.messageLength = 0;
this.messageLength = 0; this.buffer.flip();
this.buffer.flip(); this.buffer.get(bytes);
this.buffer.get(bytes); this.buffer.compact();
this.buffer.compact(); // 解密
// 解密 final String message = this.decrypt(bytes);
final String message = this.decrypt(bytes); log.debug("Socket信令消息：{} - {}", this.channel, message);
log.debug("Socket信令消息：{} - {}", this.channel, message); // 处理
// 处理 this.execute(message.strip());
this.execute(message.strip()); }
} }
} }
} }
} this.loopMessage();
this.loopMessage(); }
}
@Override
@Override public void failed(Throwable throwable, ByteBuffer buffer) {
public void failed(Throwable throwable, ByteBuffer buffer) { log.error("Socket信令终端异常：{}", this.channel, throwable);
log.error("Socket信令终端异常：{}", this.channel, throwable); this.close();
this.close(); }
}
/**
/** * @param bytes 加密消息
* @param bytes 加密消息 *
* * @return 消息
* @return 消息 */
*/ private String decrypt(byte[] bytes) {
private String decrypt(byte[] bytes) {
if(this.cipher != null) { if(this.cipher != null) {
try { try {
return new String(this.cipher.doFinal(bytes)); return new String(this.cipher.doFinal(bytes));
@@ -163,18 +161,18 @@ public final class SocketSignalMessageHandler implements CompletionHandler<Integ
} }
} }
return new String(bytes); return new String(bytes);
} }
/** /**
* @param message 消息 * @param message 消息
*/ */
private void execute(String message) { private void execute(String message) {
try { try {
this.protocolManager.execute(message, this.channel); this.protocolManager.execute(message, this.channel);
} catch (Exception e) { } catch (Exception e) {
log.error("处理Socket信令消息异常：{} - {}", this.clientManager.clients(this.channel), message, e); log.error("处理Socket信令消息异常：{} - {}", this.clientManager.clients(this.channel), message, e);
this.clientManager.push(this.channel, this.platformErrorProtocol.build(e)); this.clientManager.push(this.channel, this.platformErrorProtocol.build(e));
} }
} }
} }

6
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/client/websocket/WebSocketClient.java
View File

@@ -21,7 +21,6 @@ public class WebSocketClient extends ClientAdapter<Session> {
public WebSocketClient(long timeout, Session instance) { public WebSocketClient(long timeout, Session instance) {
super(timeout, instance); super(timeout, instance);
this.ip = (String) instance.getUserProperties().get(Constant.IP);
} }
@Override @Override
@@ -38,5 +37,10 @@ public class WebSocketClient extends ClientAdapter<Session> {
} }
} }
} }
@Override
protected String getClientIP(Session instance) {
return (String) instance.getUserProperties().get(Constant.IP);
}
} }

2
taoyao-signal-server/taoyao-signal/src/main/java/com/acgist/taoyao/signal/protocol/room/RoomCreateProtocol.java
View File

@@ -67,7 +67,7 @@ public class RoomCreateProtocol extends ProtocolClientAdapter implements Applica
); );
message.setBody(room.getRoomStatus()); message.setBody(room.getRoomStatus());
// 通知媒体终端 // 通知媒体终端
this.clientManager.broadcast(message, ClientType.MEDIA_CLIENT); this.clientManager.broadcast(message, ClientType.MEDIA_CLIENT_TYPE);
} else { } else {
this.logNoAdapter(clientType); this.logNoAdapter(clientType);
} }