alyenc/comm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

把复杂的协议改成json

Browse Source
This commit is contained in:
alyenc 2025-04-01 01:55:41 +08:00
parent ca6a44bbed
commit 086979ef33
55 changed files with 358 additions and 665 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/main/java/org/codenil/comm/Communication.java → src/main/java/dev/xiushen/andes/comm/Communication.java
View File

@ -1,21 +1,19 @@
package org.codenil.comm;
package dev.xiushen.andes.comm;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.codenil.comm.callback.ConnectCallback;
import org.codenil.comm.callback.DisconnectCallback;
import org.codenil.comm.callback.MessageCallback;
import org.codenil.comm.connections.ConnectionInitializer;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import org.codenil.comm.message.DisconnectReason;
import org.codenil.comm.message.RawMessage;
import dev.xiushen.andes.comm.callback.ConnectCallback;
import dev.xiushen.andes.comm.callback.DisconnectCallback;
import dev.xiushen.andes.comm.callback.MessageCallback;
import dev.xiushen.andes.comm.connections.ConnectionInitializer;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.message.DisconnectReason;
import dev.xiushen.andes.comm.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@ -57,17 +55,17 @@ public class Communication {
new IllegalStateException("Unable to start an already started " + getClass().getSimpleName()));
}
//注册回调监听
//注册连接成功回调
connectionEvents.subscribeConnect(this::dispatchConnect);
//启动连接初始化
return connectionInitializer
.start()
.thenApply((socketAddress) -> {
logger.info("P2P RLPx agent started and listening on {}.", socketAddress);
logger.info("Communication started and listening on {}.", socketAddress);
return socketAddress.getPort();
})
.whenComplete((_, err) -> {
.whenComplete((integer, err) -> {
if (err != null) {
logger.error("Failed to start Communication. Check for port conflicts.");
}
@ -179,14 +177,14 @@ public class Communication {
/**
* 收到消息后调用注册的回调
*/
private void dispatchMessage(final PeerConnection connection, final RawMessage message) {
private void dispatchMessage(final PeerConnection connection, final String message) {
connectionEvents.dispatchMessage(connection, message);
}
/**
* 收到指定code消息后调用指定回调
*/
private void dispatchMessageByCode(final int code, final PeerConnection connection, final RawMessage message) {
private void dispatchMessageByCode(final int code, final PeerConnection connection, final String message) {
connectionEvents.dispatchMessageByCode(code, connection, message);
}
}

2
src/main/java/org/codenil/comm/NetworkConfig.java → src/main/java/dev/xiushen/andes/comm/NetworkConfig.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm;
package dev.xiushen.andes.comm;
public class NetworkConfig {

44
src/main/java/org/codenil/comm/NetworkService.java → src/main/java/dev/xiushen/andes/comm/NetworkService.java
View File

@ -1,17 +1,22 @@
package org.codenil.comm;
package dev.xiushen.andes.comm;
import org.codenil.comm.callback.ConnectCallback;
import org.codenil.comm.callback.DisconnectCallback;
import org.codenil.comm.callback.MessageCallback;
import org.codenil.comm.connections.*;
import org.codenil.comm.message.DisconnectReason;
import org.codenil.comm.message.Message;
import org.codenil.comm.netty.NettyConnectionInitializer;
import dev.xiushen.andes.comm.callback.ConnectCallback;
import dev.xiushen.andes.comm.callback.DisconnectCallback;
import dev.xiushen.andes.comm.callback.MessageCallback;
import dev.xiushen.andes.comm.connections.ConnectionInitializer;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.message.DisconnectReason;
import dev.xiushen.andes.comm.message.Message;
import dev.xiushen.andes.comm.netty.NettyConnectionInitializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Clock;
import java.util.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
@ -58,7 +63,7 @@ public class NetworkService {
if (stopped.compareAndSet(false, true)) {
logger.info("Stopping Network.");
CompletableFuture<Void> stop = communication.stop();
return stop.whenComplete((result, throwable) -> {
return stop.whenComplete((Void, throwable) -> {
shutdown.countDown();
});
} else {
@ -77,8 +82,8 @@ public class NetworkService {
/**
* 断开连接
*/
public void disconnect(final String pkiId, final DisconnectReason reason) {
PeerConnection connection = aliveConnections.get(pkiId);
public void disconnect(final RemotePeer remotePeer, final DisconnectReason reason) {
PeerConnection connection = aliveConnections.get(remotePeer.identifier());
if(Objects.nonNull(connection)) {
try {
connection.disconnect(reason);
@ -90,12 +95,12 @@ public class NetworkService {
/**
* 发送消息
* @param pkiId 远程节点的PKIID
* @param remotePeer 远程节点RemotePeer
* @param message 发送的信息
*/
public void send(final String pkiId, final Message message) {
public void send(final RemotePeer remotePeer, final Message message) {
try {
PeerConnection connection = aliveConnections.get(pkiId);
PeerConnection connection = aliveConnections.get(remotePeer.identifier());
if (Objects.nonNull(connection)) {
connection.send(message);
}
@ -124,8 +129,8 @@ public class NetworkService {
communication.subscribeConnect(newConnection -> {
synchronized (this) {
callback.onConnect(newConnection);
aliveConnections.put(newConnection.pkiId(), newConnection);
deadConnections.remove(newConnection.pkiId(), newConnection);
aliveConnections.put(newConnection.remoteIdentifier(), newConnection);
deadConnections.remove(newConnection.remoteIdentifier(), newConnection);
}
});
}
@ -138,8 +143,8 @@ public class NetworkService {
communication.subscribeDisconnect(connection -> {
synchronized (this) {
callback.onDisconnect(connection);
aliveConnections.remove(connection.pkiId(), connection);
deadConnections.put(connection.pkiId(), connection);
aliveConnections.remove(connection.remoteIdentifier(), connection);
deadConnections.put(connection.remoteIdentifier(), connection);
}
});
}
@ -161,5 +166,4 @@ public class NetworkService {
public List<PeerConnection> aliveConnections() {
return new ArrayList<>(aliveConnections.values());
}
}

12
src/main/java/org/codenil/comm/RemotePeer.java → src/main/java/dev/xiushen/andes/comm/RemotePeer.java
View File

@ -1,8 +1,8 @@
package org.codenil.comm;
package dev.xiushen.andes.comm;
public class RemotePeer {
private String pkiId;
private String identifier;
private final String endpoint;
@ -11,8 +11,8 @@ public class RemotePeer {
this.endpoint = endpoint;
}
public String pkiId() {
return pkiId;
public String identifier() {
return identifier;
}
public String endpoint() {
@ -20,11 +20,11 @@ public class RemotePeer {
}
public void setPkiId(String pkiId) {
this.pkiId = pkiId;
this.identifier = pkiId;
}
public String toString() {
return String.format("%s, pkiId:%s", this.endpoint(), this.pkiId);
return String.format("%s, identifier:%s", this.endpoint(), this.identifier);
}
}

21
src/main/java/org/codenil/comm/Test.java → src/main/java/dev/xiushen/andes/comm/Test.java
View File

@ -1,8 +1,7 @@
package org.codenil.comm;
package dev.xiushen.andes.comm;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.message.MessageCodes;
import org.codenil.comm.message.RawMessage;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.message.MessageCodes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -22,7 +21,7 @@ public class Test {
private static void server() throws Exception {
NetworkConfig config = new NetworkConfig();
config.setBindHost("192.168.8.30");
config.setBindHost("192.168.0.26");
config.setBindPort(8080);
NetworkService service = new NetworkService(config, "");
@ -31,29 +30,27 @@ public class Test {
start.whenComplete((res, err) -> {
service.subscribeMessageByCode(MessageCodes.GOSSIP, message -> {
logger.info("接收到消息：" + message.message().code());
logger.info("接收到消息：{}", message.message());
});
});
}
private static void client() throws Exception {
NetworkConfig config = new NetworkConfig();
config.setBindHost("192.168.8.30");
config.setBindPort(8090);
config.setBindHost("192.168.31.58");
config.setBindPort(9091);
NetworkService service = new NetworkService(config, "");
CompletableFuture<Integer> start = service.start();
start.whenComplete((res, err) -> {
RemotePeer remotePeer = new RemotePeer("192.168.8.30:8080");
RemotePeer remotePeer = new RemotePeer("192.168.31.58:9090");
CompletableFuture<PeerConnection> conn = service.connect(remotePeer);
conn.whenComplete((cres, cerr) -> {
if (cerr == null) {
try {
RawMessage rawMessage = RawMessage.create(MessageCodes.GOSSIP);
rawMessage.setData("test".getBytes(StandardCharsets.UTF_8));
service.send(remotePeer.pkiId(), rawMessage);
} catch (Exception e) {
throw new RuntimeException(e);
}

4
src/main/java/org/codenil/comm/callback/ConnectCallback.java → src/main/java/dev/xiushen/andes/comm/callback/ConnectCallback.java
View File

@ -1,6 +1,6 @@
package org.codenil.comm.callback;
package dev.xiushen.andes.comm.callback;
import org.codenil.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnection;
/**
* 连接回调

4
src/main/java/org/codenil/comm/callback/DisconnectCallback.java → src/main/java/dev/xiushen/andes/comm/callback/DisconnectCallback.java
View File

@ -1,6 +1,6 @@
package org.codenil.comm.callback;
package dev.xiushen.andes.comm.callback;
import org.codenil.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnection;
@FunctionalInterface
public interface DisconnectCallback {

5
src/main/java/org/codenil/comm/callback/MessageCallback.java → src/main/java/dev/xiushen/andes/comm/callback/MessageCallback.java
View File

@ -1,9 +1,8 @@
package org.codenil.comm.callback;
package dev.xiushen.andes.comm.callback;
import org.codenil.comm.message.DefaultMessage;
import dev.xiushen.andes.comm.message.DefaultMessage;
@FunctionalInterface
public interface MessageCallback {
void onMessage(final DefaultMessage message);
}

9
src/main/java/dev/xiushen/andes/comm/callback/ResponseCallback.java Normal file
View File

@ -0,0 +1,9 @@
package dev.xiushen.andes.comm.callback;
import dev.xiushen.andes.comm.message.DefaultMessage;
import dev.xiushen.andes.comm.message.Message;
@FunctionalInterface
public interface ResponseCallback {
Message response(DefaultMessage message);
}

15
src/main/java/org/codenil/comm/connections/AbstractPeerConnection.java → src/main/java/dev/xiushen/andes/comm/connections/AbstractPeerConnection.java
View File

@ -1,12 +1,11 @@
package org.codenil.comm.connections;
package dev.xiushen.andes.comm.connections;
import dev.xiushen.andes.comm.RemotePeer;
import dev.xiushen.andes.comm.message.Message;
import io.netty.channel.ChannelHandler;
import org.codenil.comm.RemotePeer;
import org.codenil.comm.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;
public abstract class AbstractPeerConnection implements PeerConnection {
@ -26,14 +25,6 @@ public abstract class AbstractPeerConnection implements PeerConnection {
this.remoteIdentifier = remoteIdentifier;
}
@Override
public String pkiId() {
if(Objects.isNull(remotePeer)) {
throw new IllegalStateException("connection not complated yet");
}
return remotePeer.pkiId();
}
@Override
public void send(final Message message) {
doSendMessage(message);

4
src/main/java/org/codenil/comm/connections/ConnectionInitializer.java → src/main/java/dev/xiushen/andes/comm/connections/ConnectionInitializer.java
View File

@ -1,6 +1,6 @@
package org.codenil.comm.connections;
package dev.xiushen.andes.comm.connections;
import org.codenil.comm.RemotePeer;
import dev.xiushen.andes.comm.RemotePeer;
import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

9
src/main/java/org/codenil/comm/connections/KeepAlive.java → src/main/java/dev/xiushen/andes/comm/connections/KeepAlive.java
View File

@ -1,12 +1,11 @@
package org.codenil.comm.connections;
package dev.xiushen.andes.comm.connections;
import dev.xiushen.andes.comm.message.DisconnectReason;
import dev.xiushen.andes.comm.message.PingMessage;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.codenil.comm.message.DisconnectReason;
import org.codenil.comm.message.PingMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -46,7 +45,7 @@ public class KeepAlive extends ChannelDuplexHandler {
try {
logger.debug("Idle connection detected, sending Wire PING to peer.");
connection.send(PingMessage.get());
connection.send(new PingMessage());
waitingForPong.set(true);
} catch (Exception e) {
logger.trace("PING not sent because peer is already disconnected");

10
src/main/java/org/codenil/comm/connections/PeerConnection.java → src/main/java/dev/xiushen/andes/comm/connections/PeerConnection.java
View File

@ -1,14 +1,12 @@
package org.codenil.comm.connections;
package dev.xiushen.andes.comm.connections;
import dev.xiushen.andes.comm.RemotePeer;
import dev.xiushen.andes.comm.message.DisconnectReason;
import dev.xiushen.andes.comm.message.Message;
import io.netty.channel.ChannelHandler;
import org.codenil.comm.RemotePeer;
import org.codenil.comm.message.DisconnectReason;
import org.codenil.comm.message.Message;
public interface PeerConnection {
String pkiId();
String remoteIdentifier();
boolean disconnected();

17
src/main/java/org/codenil/comm/connections/PeerConnectionEvents.java → src/main/java/dev/xiushen/andes/comm/connections/PeerConnectionEvents.java
View File

@ -1,10 +1,9 @@
package org.codenil.comm.connections;
package dev.xiushen.andes.comm.connections;
import org.codenil.comm.callback.ConnectCallback;
import org.codenil.comm.callback.DisconnectCallback;
import org.codenil.comm.callback.MessageCallback;
import org.codenil.comm.message.DefaultMessage;
import org.codenil.comm.message.RawMessage;
import dev.xiushen.andes.comm.callback.ConnectCallback;
import dev.xiushen.andes.comm.callback.DisconnectCallback;
import dev.xiushen.andes.comm.callback.MessageCallback;
import dev.xiushen.andes.comm.message.DefaultMessage;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ -36,7 +35,7 @@ public class PeerConnectionEvents {
public void subscribeByCode(final int messageCode, final MessageCallback callback) {
subscribersByCode
.computeIfAbsent(messageCode, _ -> Subscribers.create())
.computeIfAbsent(messageCode, integer -> Subscribers.create())
.subscribe(callback);
}
@ -50,12 +49,12 @@ public class PeerConnectionEvents {
disconnectSubscribers.forEach(s -> s.onDisconnect(connection));
}
public void dispatchMessage(final PeerConnection connection, final RawMessage message) {
public void dispatchMessage(final PeerConnection connection, final String message) {
final DefaultMessage msg = new DefaultMessage(connection, message);
messageSubscribers.forEach(s -> s.onMessage(msg));
}
public void dispatchMessageByCode(final int code, final PeerConnection connection, final RawMessage message) {
public void dispatchMessageByCode(final int code, final PeerConnection connection, final String message) {
final DefaultMessage msg = new DefaultMessage(connection, message);
subscribersByCode.get(code).forEach(s -> s.onMessage(msg));
}

2
src/main/java/org/codenil/comm/connections/Subscribers.java → src/main/java/dev/xiushen/andes/comm/connections/Subscribers.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.connections;
package dev.xiushen.andes.comm.connections;
import com.google.common.collect.ImmutableSet;

28
src/main/java/org/codenil/comm/handler/CommonHandler.java → src/main/java/dev/xiushen/andes/comm/handler/CommonHandler.java
View File

@ -1,19 +1,19 @@
package org.codenil.comm.handler;
package dev.xiushen.andes.comm.handler;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.message.MessageCodes;
import dev.xiushen.andes.comm.message.PongMessage;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import org.codenil.comm.message.MessageCodes;
import org.codenil.comm.message.PongMessage;
import org.codenil.comm.message.RawMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicBoolean;
public class CommonHandler extends SimpleChannelInboundHandler<RawMessage> {
public class CommonHandler extends SimpleChannelInboundHandler<String> {
private static final Logger logger = LoggerFactory.getLogger(CommonHandler.class);
@ -31,13 +31,15 @@ public class CommonHandler extends SimpleChannelInboundHandler<RawMessage> {
}
@Override
protected void channelRead0(final ChannelHandlerContext ctx, final RawMessage originalMessage) {
logger.debug("Received a message from {}", originalMessage.code());
switch (originalMessage.code()) {
protected void channelRead0(final ChannelHandlerContext ctx, final String message) {
JsonObject jsonObject = new Gson().fromJson(message, JsonObject.class);
int code = jsonObject.getAsJsonPrimitive("code").getAsInt();
logger.debug("Received a message from {}", code);
switch (code) {
case MessageCodes.PING:
logger.trace("Received Wire PING");
try {
connection.send(PongMessage.get());
connection.send(new PongMessage());
} catch (Exception e) {
// Nothing to do
}
@ -55,7 +57,7 @@ public class CommonHandler extends SimpleChannelInboundHandler<RawMessage> {
connection.terminateConnection();
}
connectionEvents.dispatchMessage(connection, originalMessage);
connectionEvents.dispatchMessage(connection, message);
}
@Override

90
src/main/java/org/codenil/comm/handler/MessageFrameDecoder.java → src/main/java/dev/xiushen/andes/comm/handler/MessageFrameDecoder.java
View File

@ -1,32 +1,31 @@
package org.codenil.comm.handler;
package dev.xiushen.andes.comm.handler;
import io.netty.buffer.ByteBuf;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import dev.xiushen.andes.comm.connections.KeepAlive;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.message.DisconnectMessage;
import dev.xiushen.andes.comm.message.DisconnectReason;
import dev.xiushen.andes.comm.message.HelloMessage;
import dev.xiushen.andes.comm.message.MessageCodes;
import dev.xiushen.andes.comm.netty.NettyPeerConnection;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DecoderException;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.codenil.comm.connections.KeepAlive;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import org.codenil.comm.message.DisconnectMessage;
import org.codenil.comm.message.DisconnectReason;
import org.codenil.comm.message.MessageCodes;
import org.codenil.comm.message.RawMessage;
import org.codenil.comm.netty.NettyPeerConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
public class MessageFrameDecoder extends ByteToMessageDecoder {
public class MessageFrameDecoder extends MessageToMessageDecoder<String> {
private static final Logger logger = LoggerFactory.getLogger(MessageFrameDecoder.class);
private static final int LENGTH_FIELD_LENGTH = 4; // 长度字段占4字节
private final CompletableFuture<PeerConnection> connectionFuture;
private final PeerConnectionEvents connectionEvents;
@ -41,46 +40,17 @@ public class MessageFrameDecoder extends ByteToMessageDecoder {
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> out) throws Exception {
if (byteBuf.readableBytes() < 4) {
return; // 不足4字节长度字段，等待更多数据
}
byteBuf.readerIndex(0);
// 读取协议头：消息总长度
int totalLength = byteBuf.readInt();
if (byteBuf.readableBytes() < totalLength - 4) {
return; // 不足消息总长度，等待更多数据
}
// 读取payload
// 读取id
int idLength = byteBuf.readInt();
byte[] idBytes = new byte[idLength];
byteBuf.readBytes(idBytes);
String id = new String(idBytes, StandardCharsets.UTF_8);
// 读取code
int code = byteBuf.readInt();
// 读取data
int dataLength = byteBuf.readInt();;
byte[] data = new byte[dataLength];
byteBuf.readBytes(data);
protected void decode(ChannelHandlerContext ctx, String message, List<Object> out) {
// 创建消息对象
RawMessage message = RawMessage.create(code);
message.setRequestId(id);
message.setData(data);
JsonObject jsonObject = new Gson().fromJson(message, JsonObject.class);
int code = jsonObject.getAsJsonPrimitive("code").getAsInt();
if (hellosExchanged) {
out.add(message);
} else if (message.code() == MessageCodes.HELLO) {
} else if (code == MessageCodes.HELLO) {
hellosExchanged = true;
String remoteIdentifier = new String(message.data());
HelloMessage helloMessage = new Gson().fromJson(message, HelloMessage.class);
String remoteIdentifier = helloMessage.getData();
final PeerConnection connection = new NettyPeerConnection(ctx, remoteIdentifier, connectionEvents);
/*
@ -98,22 +68,20 @@ public class MessageFrameDecoder extends ByteToMessageDecoder {
.addLast("Common", new CommonHandler(connection, connectionEvents, waitingForPong))
.addLast("FrameEncoder", new MessageFrameEncoder());
connectionFuture.complete(connection);
} else if (message.code() == MessageCodes.DISCONNECT) {
} else if (code == MessageCodes.DISCONNECT) {
logger.debug("Disconnected before sending HELLO.");
ctx.close();
connectionFuture.completeExceptionally(new RuntimeException("Disconnect"));
} else {
logger.debug(
"Message received before HELLO's exchanged, disconnecting. Code: {}, Data: {}",
message.code(), Arrays.toString(message.data()));
if(code != MessageCodes.PONG) {
logger.debug(
"Message received before HELLO's exchanged, disconnecting. Code: {}",
code);
DisconnectMessage disconnectMessage = DisconnectMessage.create(DisconnectReason.UNKNOWN);
RawMessage rawMessage = RawMessage.create(disconnectMessage.code());
rawMessage.setData(disconnectMessage.data());
ctx.writeAndFlush(rawMessage)
.addListener(_ -> ctx.close());
connectionFuture.completeExceptionally(new RuntimeException("Message received before HELLO's exchanged"));
DisconnectMessage disconnectMessage = new DisconnectMessage(DisconnectReason.UNKNOWN.message());
ctx.writeAndFlush(disconnectMessage).addListener(Void -> ctx.close());
connectionFuture.completeExceptionally(new RuntimeException("Message received before HELLO's exchanged"));
}
}
}

18
src/main/java/dev/xiushen/andes/comm/handler/MessageFrameEncoder.java Normal file
View File

@ -0,0 +1,18 @@
package dev.xiushen.andes.comm.handler;
import com.google.gson.Gson;
import dev.xiushen.andes.comm.message.Message;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;
import java.util.List;
public class MessageFrameEncoder extends MessageToMessageEncoder<Message> {
public MessageFrameEncoder() {}
@Override
protected void encode(ChannelHandlerContext ctx, Message message, List<Object> list) {
list.add(new Gson().toJson(message));
}
}

4
src/main/java/org/codenil/comm/handler/TimeoutHandler.java → src/main/java/dev/xiushen/andes/comm/handler/TimeoutHandler.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.handler;
package dev.xiushen.andes.comm.handler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
@ -21,7 +21,7 @@ public class TimeoutHandler<C extends Channel> extends ChannelInitializer<C> {
}
@Override
protected void initChannel(final C ch) throws Exception {
protected void initChannel(final C ch) {
ch.eventLoop().schedule(() -> {
if (!condition.get()) {
callback.invoke();

58
src/main/java/org/codenil/comm/handshake/AbstractHandshakeHandler.java → src/main/java/dev/xiushen/andes/comm/handshake/AbstractHandshakeHandler.java
View File

@ -1,25 +1,23 @@
package org.codenil.comm.handshake;
package dev.xiushen.andes.comm.handshake;
import io.netty.buffer.ByteBuf;
import com.google.gson.Gson;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.handler.MessageFrameDecoder;
import dev.xiushen.andes.comm.message.HelloMessage;
import dev.xiushen.andes.comm.message.Message;
import dev.xiushen.andes.comm.message.MessageCodes;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.MessageToByteEncoder;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import org.codenil.comm.handler.MessageFrameDecoder;
import org.codenil.comm.message.HelloMessage;
import org.codenil.comm.message.MessageCodes;
import org.codenil.comm.message.RawMessage;
import org.codenil.comm.serialize.SerializeHelper;
import io.netty.handler.codec.MessageToMessageEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
public abstract class AbstractHandshakeHandler extends SimpleChannelInboundHandler<ByteBuf> {
public abstract class AbstractHandshakeHandler extends SimpleChannelInboundHandler<String> {
private static final Logger logger = LoggerFactory.getLogger(AbstractHandshakeHandler.class);
@ -43,8 +41,8 @@ public abstract class AbstractHandshakeHandler extends SimpleChannelInboundHandl
}
@Override
protected void channelRead0(final ChannelHandlerContext ctx, final ByteBuf msg) {
final Optional<ByteBuf> nextMsg = nextHandshakeMessage(msg);
protected void channelRead0(final ChannelHandlerContext ctx, final String msg) {
final Optional<String> nextMsg = nextHandshakeMessage(msg);
if (nextMsg.isPresent()) {
ctx.writeAndFlush(nextMsg.get());
@ -67,18 +65,13 @@ public abstract class AbstractHandshakeHandler extends SimpleChannelInboundHandl
* 替换完编解码器后发送Hello消息
* hello消息需要带一些数据
*/
HelloMessage helloMessage = HelloMessage.create(selfIdentifier.getBytes(StandardCharsets.UTF_8));
RawMessage rawMessage = RawMessage.create(helloMessage.code());
rawMessage.setData(helloMessage.data());
rawMessage.setRequestId(helloMessage.requestId());
ctx.writeAndFlush(rawMessage)
HelloMessage helloMessage = new HelloMessage(selfIdentifier);
ctx.writeAndFlush(helloMessage)
.addListener(ff -> {
if (ff.isSuccess()) {
logger.trace("Successfully wrote hello message");
}
});
msg.retain();
ctx.fireChannelRead(msg);
}
}
@ -90,31 +83,22 @@ public abstract class AbstractHandshakeHandler extends SimpleChannelInboundHandl
ctx.close();
}
protected abstract Optional<ByteBuf> nextHandshakeMessage(ByteBuf msg);
protected abstract Optional<String> nextHandshakeMessage(String msg);
/** Ensures that wire hello message is the first message written. */
private static class FirstMessageFrameEncoder extends MessageToByteEncoder<RawMessage> {
private static class FirstMessageFrameEncoder extends MessageToMessageEncoder<Message> {
private FirstMessageFrameEncoder() {}
@Override
protected void encode(
final ChannelHandlerContext context,
final RawMessage msg,
final ByteBuf out) {
if (msg.code() != MessageCodes.HELLO) {
final Message msg,
final List<Object> list) {
if (msg.getCode() != MessageCodes.HELLO) {
throw new IllegalStateException("First wire message sent wasn't a HELLO.");
}
byte[] idBytes = Optional.ofNullable(msg.requestId()).orElse("").getBytes(StandardCharsets.UTF_8);
SerializeHelper builder = new SerializeHelper();
ByteBuf buf = builder.writeBytes(idBytes)
.writeInt(msg.code())
.writeBytes(msg.data())
.build();
out.writeBytes(buf);
buf.release();
list.add(new Gson().toJson(msg));
context.pipeline().remove(this);
}
}

11
src/main/java/org/codenil/comm/handshake/HandshakeHandlerInbound.java → src/main/java/dev/xiushen/andes/comm/handshake/HandshakeHandlerInbound.java
View File

@ -1,8 +1,7 @@
package org.codenil.comm.handshake;
package dev.xiushen.andes.comm.handshake;
import io.netty.buffer.ByteBuf;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@ -22,8 +21,8 @@ public class HandshakeHandlerInbound extends AbstractHandshakeHandler {
}
@Override
protected Optional<ByteBuf> nextHandshakeMessage(ByteBuf msg) {
final Optional<ByteBuf> nextMsg;
protected Optional<String> nextHandshakeMessage(String msg) {
final Optional<String> nextMsg;
if (handshaker.getStatus() == HandshakeStatus.IN_PROGRESS) {
nextMsg = handshaker.handleMessage(msg);
} else {

16
src/main/java/org/codenil/comm/handshake/HandshakeHandlerOutbound.java → src/main/java/dev/xiushen/andes/comm/handshake/HandshakeHandlerOutbound.java
View File

@ -1,10 +1,8 @@
package org.codenil.comm.handshake;
package dev.xiushen.andes.comm.handshake;
import io.netty.buffer.ByteBuf;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import io.netty.channel.ChannelHandlerContext;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -16,9 +14,9 @@ import java.util.concurrent.CompletableFuture;
*/
public class HandshakeHandlerOutbound extends AbstractHandshakeHandler {
private static final Logger logger = LoggerFactory.getLogger(AbstractHandshakeHandler.class);
private static final Logger logger = LoggerFactory.getLogger(HandshakeHandlerOutbound.class);
private final ByteBuf first;
private final String first;
public HandshakeHandlerOutbound(
final String selfIdentifier,
@ -32,8 +30,8 @@ public class HandshakeHandlerOutbound extends AbstractHandshakeHandler {
}
@Override
protected Optional<ByteBuf> nextHandshakeMessage(ByteBuf msg) {
final Optional<ByteBuf> nextMsg;
protected Optional<String> nextHandshakeMessage(String msg) {
final Optional<String> nextMsg;
if (handshaker.getStatus() == HandshakeStatus.IN_PROGRESS) {
nextMsg = handshaker.handleMessage(msg);
} else {

2
src/main/java/org/codenil/comm/handshake/HandshakeStatus.java → src/main/java/dev/xiushen/andes/comm/handshake/HandshakeStatus.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.handshake;
package dev.xiushen.andes.comm.handshake;
public enum HandshakeStatus {
UNINITIALIZED,

8
src/main/java/org/codenil/comm/handshake/Handshaker.java → src/main/java/dev/xiushen/andes/comm/handshake/Handshaker.java
View File

@ -1,6 +1,4 @@
package org.codenil.comm.handshake;
import io.netty.buffer.ByteBuf;
package dev.xiushen.andes.comm.handshake;
import java.util.Optional;
@ -12,7 +10,7 @@ public interface Handshaker {
HandshakeStatus getStatus();
ByteBuf firstMessage();
String firstMessage();
Optional<ByteBuf> handleMessage(ByteBuf buf);
Optional<String> handleMessage(String buf);
}

43
src/main/java/org/codenil/comm/handshake/PlainHandshaker.java → src/main/java/dev/xiushen/andes/comm/handshake/PlainHandshaker.java
View File

@ -1,10 +1,10 @@
package org.codenil.comm.handshake;
package dev.xiushen.andes.comm.handshake;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.codenil.comm.handler.MessageHandler;
import org.codenil.comm.message.MessageType;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import dev.xiushen.andes.comm.message.MessageCodes;
import dev.xiushen.andes.comm.message.PingMessage;
import dev.xiushen.andes.comm.message.PongMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -15,14 +15,14 @@ import static com.google.common.base.Preconditions.checkState;
public class PlainHandshaker implements Handshaker {
private static final Logger logger = LoggerFactory.getLogger(AbstractHandshakeHandler.class);
private static final Logger logger = LoggerFactory.getLogger(PlainHandshaker.class);
private final AtomicReference<HandshakeStatus> status =
new AtomicReference<>(HandshakeStatus.UNINITIALIZED);
private boolean initiator;
private byte[] initiatorMsg;
private byte[] responderMsg;
private String initiatorMsg;
private String responderMsg;
@Override
public void prepareInitiator() {
@ -46,42 +46,41 @@ public class PlainHandshaker implements Handshaker {
}
@Override
public ByteBuf firstMessage() {
public String firstMessage() {
checkState(initiator, "illegal invocation of firstMessage on non-initiator end of handshake");
checkState(status.compareAndSet(HandshakeStatus.PREPARED, HandshakeStatus.IN_PROGRESS),
"illegal invocation of firstMessage, handshake had already started");
initiatorMsg = MessageHandler.buildMessage(MessageType.PING, MessageType.PING.getValue(), new byte[0]);
logger.trace("First plain handshake message under INITIATOR role");
return Unpooled.wrappedBuffer(initiatorMsg);
return new Gson().toJson(new PingMessage());
}
@Override
public Optional<ByteBuf> handleMessage(ByteBuf buf) {
public Optional<String> handleMessage(String message) {
checkState(status.get() == HandshakeStatus.IN_PROGRESS,
"illegal invocation of onMessage on handshake that is not in progress");
PlainMessage message = MessageHandler.parseMessage(buf);
Optional<byte[]> nextMsg = Optional.empty();
JsonObject jsonObject = new Gson().fromJson(message, JsonObject.class);
int code = jsonObject.getAsJsonPrimitive("code").getAsInt();
Optional<String> nextMsg = Optional.empty();
if (initiator) {
checkState(responderMsg == null,
"unexpected message: responder message had " + "already been received");
checkState(message.messageType().equals(MessageType.PONG),
checkState(code == MessageCodes.PONG,
"unexpected message: needs to be a pong");
responderMsg = message.data();
responderMsg = message;
} else {
checkState(initiatorMsg == null,
"unexpected message: initiator message " + "had already been received");
checkState(message.messageType().equals(MessageType.PING),
checkState(code == MessageCodes.PING,
"unexpected message: needs to be a ping");
initiatorMsg = message.data();
responderMsg = MessageHandler.buildMessage(MessageType.PONG, MessageType.PONG.getValue(), new byte[0]);
initiatorMsg = message;
responderMsg = new Gson().toJson(new PongMessage());
nextMsg = Optional.of(responderMsg);
}
status.set(HandshakeStatus.SUCCESS);
logger.trace("Handshake status set to {}", status.get());
return nextMsg.map(Unpooled::wrappedBuffer);
return nextMsg;
}
}

14
src/main/java/dev/xiushen/andes/comm/message/AbstractMessage.java Normal file
View File

@ -0,0 +1,14 @@
package dev.xiushen.andes.comm.message;
public abstract class AbstractMessage implements Message {
private final int code;
public AbstractMessage(final int code) {
this.code = code;
}
public int getCode() {
return code;
}
}

15
src/main/java/dev/xiushen/andes/comm/message/DataMessage.java Normal file
View File

@ -0,0 +1,15 @@
package dev.xiushen.andes.comm.message;
public abstract class DataMessage extends AbstractMessage {
private final String data;
public DataMessage(final int code, final String data) {
super(code);
this.data = data;
}
public String getData() {
return data;
}
}

10
src/main/java/org/codenil/comm/message/DefaultMessage.java → src/main/java/dev/xiushen/andes/comm/message/DefaultMessage.java
View File

@ -1,21 +1,21 @@
package org.codenil.comm.message;
package dev.xiushen.andes.comm.message;
import org.codenil.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnection;
public class DefaultMessage {
private final RawMessage message;
private final String message;
private final PeerConnection connection;
public DefaultMessage(
final PeerConnection connection,
final RawMessage message) {
final String message) {
this.message = message;
this.connection = connection;
}
public RawMessage message() {
public String message() {
return message;
}

4
src/main/java/org/codenil/comm/message/DisconnectCallback.java → src/main/java/dev/xiushen/andes/comm/message/DisconnectCallback.java
View File

@ -1,6 +1,6 @@
package org.codenil.comm.message;
package dev.xiushen.andes.comm.message;
import org.codenil.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnection;
@FunctionalInterface
public interface DisconnectCallback {

8
src/main/java/dev/xiushen/andes/comm/message/DisconnectMessage.java Normal file
View File

@ -0,0 +1,8 @@
package dev.xiushen.andes.comm.message;
public class DisconnectMessage extends DataMessage {
public DisconnectMessage(final String data) {
super(MessageCodes.DISCONNECT, data);
}
}

2
src/main/java/org/codenil/comm/message/DisconnectReason.java → src/main/java/dev/xiushen/andes/comm/message/DisconnectReason.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.message;
package dev.xiushen.andes.comm.message;
public enum DisconnectReason {

8
src/main/java/dev/xiushen/andes/comm/message/EmptyMessage.java Normal file
View File

@ -0,0 +1,8 @@
package dev.xiushen.andes.comm.message;
public abstract class EmptyMessage extends AbstractMessage {
public EmptyMessage(int code) {
super(code);
}
}

8
src/main/java/dev/xiushen/andes/comm/message/HelloMessage.java Normal file
View File

@ -0,0 +1,8 @@
package dev.xiushen.andes.comm.message;
public class HelloMessage extends DataMessage {
public HelloMessage(final String data) {
super(MessageCodes.HELLO, data);
}
}

6
src/main/java/dev/xiushen/andes/comm/message/Message.java Normal file
View File

@ -0,0 +1,6 @@
package dev.xiushen.andes.comm.message;
public interface Message {
int getCode();
}

2
src/main/java/org/codenil/comm/message/MessageCallback.java → src/main/java/dev/xiushen/andes/comm/message/MessageCallback.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.message;
package dev.xiushen.andes.comm.message;
@FunctionalInterface
public interface MessageCallback {

2
src/main/java/org/codenil/comm/message/MessageCodes.java → src/main/java/dev/xiushen/andes/comm/message/MessageCodes.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.message;
package dev.xiushen.andes.comm.message;
public class MessageCodes {
public static final int HELLO = 0x00;

13
src/main/java/dev/xiushen/andes/comm/message/PingMessage.java Normal file
View File

@ -0,0 +1,13 @@
package dev.xiushen.andes.comm.message;
public class PingMessage extends EmptyMessage {
public PingMessage() {
super(MessageCodes.PING);
}
@Override
public String toString() {
return "PingMessage{data=''}";
}
}

8
src/main/java/dev/xiushen/andes/comm/message/PongMessage.java Normal file
View File

@ -0,0 +1,8 @@
package dev.xiushen.andes.comm.message;
public class PongMessage extends EmptyMessage {
public PongMessage() {
super(MessageCodes.PONG);
}
}

62
src/main/java/org/codenil/comm/netty/NettyConnectionInitializer.java → src/main/java/dev/xiushen/andes/comm/netty/NettyConnectionInitializer.java
View File

@ -1,5 +1,16 @@
package org.codenil.comm.netty;
package dev.xiushen.andes.comm.netty;
import dev.xiushen.andes.comm.NetworkConfig;
import dev.xiushen.andes.comm.RemotePeer;
import dev.xiushen.andes.comm.callback.ConnectCallback;
import dev.xiushen.andes.comm.connections.ConnectionInitializer;
import dev.xiushen.andes.comm.connections.PeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.connections.Subscribers;
import dev.xiushen.andes.comm.handler.TimeoutHandler;
import dev.xiushen.andes.comm.handshake.HandshakeHandlerInbound;
import dev.xiushen.andes.comm.handshake.HandshakeHandlerOutbound;
import dev.xiushen.andes.comm.handshake.PlainHandshaker;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
@ -7,22 +18,15 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.codenil.comm.NetworkConfig;
import org.codenil.comm.RemotePeer;
import org.codenil.comm.callback.ConnectCallback;
import org.codenil.comm.connections.ConnectionInitializer;
import org.codenil.comm.connections.PeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import org.codenil.comm.connections.Subscribers;
import org.codenil.comm.handler.TimeoutHandler;
import org.codenil.comm.handshake.HandshakeHandlerInbound;
import org.codenil.comm.handshake.HandshakeHandlerOutbound;
import org.codenil.comm.handshake.PlainHandshaker;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import javax.annotation.Nonnull;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.security.GeneralSecurityException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
@ -34,6 +38,11 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class NettyConnectionInitializer implements ConnectionInitializer {
private static final int TIMEOUT_SECONDS = 10;
private static final int MAX_FRAME_LENGTH = 1024 * 1024; // 最大消息长度
private static final int LENGTH_FIELD_OFFSET = 0; // 长度字段起始位置
private static final int LENGTH_FIELD_LENGTH = 4; // 长度字段占4字节
private static final int LENGTH_ADJUSTMENT = 0; // 长度字段后数据的偏移量
private static final int INITIAL_BYTES_TO_STRIP = 4; // 跳过长度字段
private final Subscribers<ConnectCallback> connectSubscribers = Subscribers.create();
private final PeerConnectionEvents eventDispatcher;
@ -76,9 +85,8 @@ public class NettyConnectionInitializer implements ConnectionInitializer {
.channel(NioServerSocketChannel.class)
.childHandler(inboundChannelInitializer())
.bind(config.bindHost(), config.bindPort());
server.addListener(future -> {
final InetSocketAddress socketAddress =
(InetSocketAddress) server.channel().localAddress();
this.server.addListener(future -> {
final InetSocketAddress socketAddress = (InetSocketAddress) server.channel().localAddress();
if (!future.isSuccess() || socketAddress == null) {
final String message =
String.format("Unable to start listening on %s:%s. Check for port conflicts.",
@ -154,9 +162,17 @@ public class NettyConnectionInitializer implements ConnectionInitializer {
protected void initChannel(final SocketChannel ch) throws Exception {
final CompletableFuture<PeerConnection> connectionFuture = new CompletableFuture<>();
connectionFuture.thenAccept(connection -> connectSubscribers.forEach(c -> c.onConnect(connection)));
//连接处理器
//连接超时处理器
ch.pipeline().addLast(timeoutHandler(connectionFuture, "Timed out waiting to fully establish incoming connection"));
//其他处理器，TLS之类的
ch.pipeline().addLast(new LengthFieldPrepender(4)); // 编码器：添加4字节长度头
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(
MAX_FRAME_LENGTH, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH,
LENGTH_ADJUSTMENT, INITIAL_BYTES_TO_STRIP));
// JSON 编解码（基于String）
ch.pipeline().addLast(new StringDecoder(StandardCharsets.UTF_8)); // 解码ByteBuf -> String
ch.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8)); // 编码String -> ByteBuf
addAdditionalInboundHandlers(ch);
//握手消息处理器
ch.pipeline().addLast(inboundHandler(selfIdentifier, connectionFuture));
@ -172,8 +188,18 @@ public class NettyConnectionInitializer implements ConnectionInitializer {
protected void initChannel(final SocketChannel ch) throws Exception {
//连接处理器
ch.pipeline().addLast(timeoutHandler(connectionFuture, "Timed out waiting to establish connection with peer: " + remotePeer.toString()));
ch.pipeline().addLast(new LengthFieldPrepender(4)); // 编码器：添加4字节长度头
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(
MAX_FRAME_LENGTH, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH,
LENGTH_ADJUSTMENT, INITIAL_BYTES_TO_STRIP));
// JSON 编解码（基于String）
ch.pipeline().addLast(new StringDecoder(StandardCharsets.UTF_8)); // 解码ByteBuf -> String
ch.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8)); // 编码String -> ByteBuf
//其他处理器
addAdditionalOutboundHandlers(ch, remotePeer);
//握手消息处理器
ch.pipeline().addLast(outboundHandler(selfIdentifier, remotePeer, connectionFuture));
}

22
src/main/java/org/codenil/comm/netty/NettyPeerConnection.java → src/main/java/dev/xiushen/andes/comm/netty/NettyPeerConnection.java
View File

@ -1,15 +1,13 @@
package org.codenil.comm.netty;
package dev.xiushen.andes.comm.netty;
import dev.xiushen.andes.comm.connections.AbstractPeerConnection;
import dev.xiushen.andes.comm.connections.PeerConnectionEvents;
import dev.xiushen.andes.comm.message.DisconnectMessage;
import dev.xiushen.andes.comm.message.DisconnectReason;
import dev.xiushen.andes.comm.message.Message;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import org.codenil.comm.connections.AbstractPeerConnection;
import org.codenil.comm.connections.PeerConnectionEvents;
import org.codenil.comm.message.DisconnectMessage;
import org.codenil.comm.message.DisconnectReason;
import org.codenil.comm.message.Message;
import org.codenil.comm.message.RawMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -19,7 +17,7 @@ import static java.util.concurrent.TimeUnit.SECONDS;
public class NettyPeerConnection extends AbstractPeerConnection {
private static final Logger logger = LoggerFactory.getLogger(AbstractPeerConnection.class);
private static final Logger logger = LoggerFactory.getLogger(NettyPeerConnection.class);
private final ChannelHandlerContext ctx;
@ -53,16 +51,14 @@ public class NettyPeerConnection extends AbstractPeerConnection {
public void disconnect(DisconnectReason reason) {
if (disconnected.compareAndSet(false, true)) {
connectionEvents.dispatchDisconnect(this);
doSendMessage(DisconnectMessage.create(reason));
doSendMessage(new DisconnectMessage(reason.message()));
closeConnection();
}
}
@Override
protected void doSendMessage(final Message message) {
RawMessage rawMessage = RawMessage.create(message.code());
rawMessage.setData(message.data());
ctx.channel().writeAndFlush(rawMessage);
ctx.channel().writeAndFlush(message);
}
@Override

2
src/main/java/org/codenil/comm/serialize/SerializeHelper.java → src/main/java/dev/xiushen/andes/comm/serialize/SerializeHelper.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.serialize;
package dev.xiushen.andes.comm.serialize;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

2
src/main/java/org/codenil/comm/serialize/Version.java → src/main/java/dev/xiushen/andes/comm/serialize/Version.java
View File

@ -1,4 +1,4 @@
package org.codenil.comm.serialize;
package dev.xiushen.andes.comm.serialize;
public class Version {

9
src/main/java/org/codenil/comm/callback/ResponseCallback.java
View File

@ -1,9 +0,0 @@
package org.codenil.comm.callback;
import org.codenil.comm.message.DefaultMessage;
import org.codenil.comm.message.Message;
@FunctionalInterface
public interface ResponseCallback {
Message response(DefaultMessage message);
}

31
src/main/java/org/codenil/comm/handler/MessageFrameEncoder.java
View File

@ -1,31 +0,0 @@
package org.codenil.comm.handler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.codenil.comm.message.RawMessage;
import org.codenil.comm.serialize.SerializeHelper;
import java.nio.charset.StandardCharsets;
import java.util.Optional;
public class MessageFrameEncoder extends MessageToByteEncoder<RawMessage> {
public MessageFrameEncoder() {}
@Override
protected void encode(
final ChannelHandlerContext ctx,
final RawMessage msg,
final ByteBuf out) {
byte[] idBytes = Optional.ofNullable(msg.requestId()).orElse("").getBytes(StandardCharsets.UTF_8);
SerializeHelper builder = new SerializeHelper();
ByteBuf buf = builder.writeBytes(idBytes)
.writeInt(msg.code())
.writeBytes(msg.data())
.build();
out.writeBytes(buf);
buf.release();
}
}

51
src/main/java/org/codenil/comm/handler/MessageHandler.java
View File

@ -1,51 +0,0 @@
package org.codenil.comm.handler;
import io.netty.buffer.ByteBuf;
import org.codenil.comm.handshake.PlainMessage;
import org.codenil.comm.message.MessageType;
import org.codenil.comm.serialize.SerializeHelper;
public class MessageHandler {
public static byte[] buildMessage(final PlainMessage message) {
SerializeHelper builder = new SerializeHelper();
ByteBuf buf = builder.writeInt(message.messageType().getValue())
.writeInt(message.code())
.writeBytes(message.data()).build();
byte[] result = new byte[buf.readableBytes()];
buf.readBytes(result);
buf.release();
return result;
}
public static byte[] buildMessage(
final MessageType messageType,
final int code,
final byte[] data) {
return buildMessage(new PlainMessage(messageType, code, data));
}
public static PlainMessage parseMessage(final ByteBuf buf) {
PlainMessage ret = null;
buf.readerIndex(0);
//跳过版本
int versionLength = buf.readInt();
buf.skipBytes(versionLength);
int payloadLength = buf.readInt();
if(payloadLength < 8) {
return ret;
}
int messageType = buf.readInt();
int code = buf.readInt();
int dataLength = buf.readInt();
byte[] data = new byte[dataLength];
buf.readBytes(data);
ret = new PlainMessage(MessageType.forNumber(messageType), code, data);
return ret;
}
}

31
src/main/java/org/codenil/comm/handshake/PlainMessage.java
View File

@ -1,31 +0,0 @@
package org.codenil.comm.handshake;
import org.codenil.comm.message.MessageType;
public class PlainMessage {
private final MessageType messageType;
private final int code;
private final byte[] data;
public PlainMessage(final MessageType messageType, final byte[] data) {
this(messageType, -1, data);
}
public PlainMessage(final MessageType messageType, final int code, final byte[] data) {
this.messageType = messageType;
this.code = code;
this.data = data;
}
public MessageType messageType() {
return messageType;
}
public byte[] data() {
return data;
}
public int code() {
return code;
}
}

31
src/main/java/org/codenil/comm/message/AbstractMessage.java
View File

@ -1,31 +0,0 @@
package org.codenil.comm.message;
public abstract class AbstractMessage implements Message {
private String requestId;
private byte[] data;
@Override
public String requestId() {
return requestId;
}
@Override
public final int size() {
return data.length;
}
@Override
public byte[] data() {
return data;
}
public void setRequestId(String requestId) {
this.requestId = requestId;
}
public void setData(byte[] data) {
this.data = data;
}
}

29
src/main/java/org/codenil/comm/message/DisconnectMessage.java
View File

@ -1,29 +0,0 @@
package org.codenil.comm.message;
public class DisconnectMessage extends AbstractMessage {
private DisconnectMessage(final byte[] data) {
super.setData(data);
}
public static DisconnectMessage create(final DisconnectReason reason) {
return new DisconnectMessage(new byte[]{reason.code()});
}
public static DisconnectMessage readFrom(final Message message) {
if (message instanceof DisconnectMessage) {
return (DisconnectMessage) message;
}
final int code = message.code();
if (code != MessageCodes.DISCONNECT) {
throw new IllegalArgumentException(
String.format("Message has code %d and thus is not a DisconnectMessage.", code));
}
return new DisconnectMessage(message.data());
}
@Override
public int code() {
return MessageCodes.DISCONNECT;
}
}

19
src/main/java/org/codenil/comm/message/EmptyMessage.java
View File

@ -1,19 +0,0 @@
package org.codenil.comm.message;
public abstract class EmptyMessage implements Message {
@Override
public final int size() {
return 0;
}
@Override
public byte[] data() {
return new byte[]{};
}
@Override
public String toString() {
return getClass().getSimpleName() + "{ code=" + code() + ", size=0}";
}
}

17
src/main/java/org/codenil/comm/message/HelloMessage.java
View File

@ -1,17 +0,0 @@
package org.codenil.comm.message;
public class HelloMessage extends AbstractMessage {
public HelloMessage(final byte[] data) {
super.setData(data);
}
public static HelloMessage create(byte[] bytes) {
return new HelloMessage(new byte[0]);
}
@Override
public int code() {
return MessageCodes.HELLO;
}
}

19
src/main/java/org/codenil/comm/message/Message.java
View File

@ -1,19 +0,0 @@
package org.codenil.comm.message;
public interface Message {
String requestId();
int size();
int code();
byte[] data();
default RawMessage wrapMessage(final String requestId) {
RawMessage rawMessage = RawMessage.create(code());
rawMessage.setRequestId(requestId);
rawMessage.setData(data());
return rawMessage;
}
}

29
src/main/java/org/codenil/comm/message/MessageType.java
View File

@ -1,29 +0,0 @@
package org.codenil.comm.message;
public enum MessageType {
PING(0),
PONG(1),
DATA(2),
UNRECOGNIZED(-1),
;
private final int value;
MessageType(final int value) {
this.value = value;
}
public int getValue() {
return value;
}
public static MessageType forNumber(final int value) {
return switch (value) {
case 0 -> PING;
case 1 -> PONG;
case 2 -> DATA;
case -1 -> UNRECOGNIZED;
default -> null;
};
}
}

26
src/main/java/org/codenil/comm/message/PingMessage.java
View File

@ -1,26 +0,0 @@
package org.codenil.comm.message;
public class PingMessage extends EmptyMessage {
private static final PingMessage INSTANCE = new PingMessage();
public static PingMessage get() {
return INSTANCE;
}
private PingMessage() {}
@Override
public String requestId() {
return "";
}
@Override
public int code() {
return MessageCodes.PING;
}
@Override
public String toString() {
return "PingMessage{data=''}";
}
}

21
src/main/java/org/codenil/comm/message/PongMessage.java
View File

@ -1,21 +0,0 @@
package org.codenil.comm.message;
public class PongMessage extends EmptyMessage {
private static final PongMessage INSTANCE = new PongMessage();
public static PongMessage get() {
return INSTANCE;
}
private PongMessage() {}
@Override
public String requestId() {
return "";
}
@Override
public int code() {
return MessageCodes.PONG;
}
}

56
src/main/java/org/codenil/comm/message/RawMessage.java
View File

@ -1,56 +0,0 @@
package org.codenil.comm.message;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.codenil.comm.serialize.SerializeHelper;
public class RawMessage extends AbstractMessage {
private final int code;
protected RawMessage(final int code) {
this.code = code;
}
public static RawMessage create(
final int code) {
return new RawMessage(code);
}
public static RawMessage decode(byte[] messageBytes) {
ByteBuf buf = Unpooled.wrappedBuffer(messageBytes);
buf.readerIndex(0);
int code = buf.readInt();
int requestIdLength = buf.readInt();
byte[] requestIdBytes = new byte[requestIdLength];
buf.readBytes(requestIdBytes);
int dataLength = buf.readInt();
byte[] dataBytes = new byte[dataLength];
buf.readBytes(dataBytes);
RawMessage rawMessage = create(code);
rawMessage.setRequestId(new String(requestIdBytes));
rawMessage.setData(dataBytes);
return rawMessage;
}
public static byte[] encode(RawMessage rawMessage) {
SerializeHelper converter = new SerializeHelper();
ByteBuf byteBuf = converter
.writeVersion("1.0")
.writeInt(rawMessage.code())
.writeString(rawMessage.requestId())
.writeBytes(rawMessage.data())
.build();
return converter.readPayload(byteBuf);
}
@Override
public int code() {
return code;
}
}