Minecraft/mc-server
Archived
0
Fork 0
Code Releases Activity

[WIP] try RxJava

Browse Source
This commit is contained in:
DmitriyMX
2020-05-21 00:28:10 +03:00
parent 980088b157
commit c9f396a999
14 changed files with 191 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
server-api/build.gradle Normal file
View File

@@ -0,0 +1,18 @@
ext {
netty_version = '4.1.22.Final';
library = [
rxjava: ['io.reactivex.rxjava3:rxjava:3.0.3'],
netty: [
"io.netty:netty-transport:$netty_version"
]
]
}
dependencies {
/* COMPONENTS */
compile project(':protocol')
compile library.rxjava
implementation library.netty
}

12
server-api/src/main/java/mc/server/reactive/NetworkContainer.java Normal file
View File

@@ -0,0 +1,12 @@
package mc.server.reactive;
import io.netty.channel.Channel;
import lombok.Data;
import mc.protocol.Packet;
@Data
public class NetworkContainer<T extends Packet> {
private final Channel channel;
private final T packet;
}

28
server-api/src/main/java/mc/server/reactive/RxProvider.java Normal file
View File

@@ -0,0 +1,28 @@
package mc.server.reactive;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.subjects.PublishSubject;
import io.reactivex.rxjava3.subjects.Subject;
import mc.protocol.Packet;
import java.util.HashMap;
import java.util.Map;
@SuppressWarnings({ "rawtypes", "unchecked" })
public class RxProvider {
private final Map<Class, Subject> subjectMap = new HashMap<>();
public <T extends Packet> Observable<NetworkContainer<T>> getPacketObservable(Class<T> packetType) {
return giveSubject(packetType);
}
public Observer<NetworkContainer<? extends Packet>> getPacketObserver(Class<? extends Packet> packetType) {
return giveSubject(packetType);
}
private Subject giveSubject(Class<? extends Packet> packetType) {
return subjectMap.computeIfAbsent(packetType, key -> PublishSubject.create());
}
}

1
server/build.gradle
View File

@@ -15,7 +15,6 @@ dependencies {
implementation library.logger
/* COMPONENTS */
implementation project(':protocol')
implementation project(':server-api')
implementation library.guice
implementation library.netty

28
server/src/main/java/mc/server/logic/HandshakeLogic.java Normal file
View File

@@ -0,0 +1,28 @@
package mc.server.logic;
import com.google.inject.Inject;
import io.netty.channel.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import mc.protocol.handshake.client.HandshakePacket;
import mc.server.network.impl.NettyConstants;
import mc.server.reactive.RxProvider;
@Slf4j
@RequiredArgsConstructor(onConstructor = @__({ @Inject }))
public class HandshakeLogic {
private final RxProvider rxProvider;
public void setup() {
log.info("HandshakeLogic setup");
rxProvider.getPacketObservable(HandshakePacket.class)
.doOnNext(container -> handle(container.getChannel(), container.getPacket()))
.subscribe();
}
private void handle(Channel channel, HandshakePacket packet) {
log.info("HandshakePacket handle");
channel.attr(NettyConstants.ATTR_STATE).set(packet.getNextState());
}
}

38
server/src/main/java/mc/server/logic/LoginLogic.java Normal file
View File

@@ -0,0 +1,38 @@
package mc.server.logic;
import com.google.inject.Inject;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import mc.protocol.login.client.LoginStartPacket;
import mc.protocol.login.server.DisconnectPacket;
import mc.protocol.text.Text;
import mc.protocol.text.TextColor;
import mc.protocol.text.TextStyle;
import mc.server.reactive.NetworkContainer;
import mc.server.reactive.RxProvider;
@Slf4j
@RequiredArgsConstructor(onConstructor = @__({ @Inject }))
public class LoginLogic {
private final RxProvider rxProvider;
public void setup() {
log.info("LoginLogic setup");
rxProvider.getPacketObservable(LoginStartPacket.class)
.map(NetworkContainer::getChannel)
.doOnNext(channel -> {
log.info("LoginLogic handle");
DisconnectPacket disconnectPacket = new DisconnectPacket();
disconnectPacket.setReason(Text.builder()
.append(Text.of(TextColor.WHITE, "Server is "))
.color(TextColor.RED).style(TextStyle.BOLD).append("NOT ")
.color(TextColor.WHITE).append("available.")
.build());
channel.writeAndFlush(disconnectPacket).channel().disconnect();
})
.subscribe();
}
}

32
server/src/main/java/mc/server/network/impl/handler/StatusHandler.java → server/src/main/java/mc/server/logic/StatusLogic.java
View File

@@ -1,12 +1,16 @@
package mc.server.network.impl.handler;
package mc.server.logic;
import io.netty.channel.ChannelHandlerContext;
import com.google.inject.Inject;
import io.netty.channel.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import mc.protocol.ProtocolConstant;
import mc.protocol.dto.ServerInfo;
import mc.protocol.status.PingPacket;
import mc.protocol.status.client.StatusServerRequest;
import mc.protocol.status.server.StatusServerResponse;
import mc.protocol.text.Text;
import mc.server.reactive.RxProvider;
import org.apache.commons.io.IOUtils;
import java.io.IOException;
@@ -14,12 +18,28 @@ import java.util.Base64;
import java.util.Collections;
@Slf4j
public class StatusHandler extends AbstractPacketHandler<StatusServerRequest> {
@RequiredArgsConstructor(onConstructor = @__({ @Inject }))
public class StatusLogic {
private static final String FAVICON_HEADER = "data:image/png;base64,";
private final RxProvider rxProvider;
@Override
protected void channelRead1(ChannelHandlerContext ctx, StatusServerRequest packet) throws Exception {
public void setup() {
log.info("StatusLogic setup");
rxProvider.getPacketObservable(StatusServerRequest.class)
.doOnNext(container -> status(container.getChannel(), container.getPacket()))
.subscribe();
rxProvider.getPacketObservable(PingPacket.class)
.doOnNext(container -> {
log.info("StatusLogic ping handle");
container.getChannel().writeAndFlush(container.getPacket()).channel().disconnect();
})
.subscribe();
}
private void status(Channel channel, StatusServerRequest packet) throws IOException {
log.info("StatusLogic handle");
log.info("{}", packet);
final ServerInfo.Version version = new ServerInfo.Version();
@@ -40,7 +60,7 @@ public class StatusHandler extends AbstractPacketHandler<StatusServerRequest> {
StatusServerResponse response = new StatusServerResponse();
response.setServerInfoDto(serverInfo);
ctx.channel().writeAndFlush(response);
channel.writeAndFlush(response);
}
private String getEmbeddedIconBase64() throws IOException {

12
server/src/main/java/mc/server/network/config/NetworkModule.java
View File

@@ -21,9 +21,10 @@ import mc.server.network.impl.NettyServer;
import mc.server.network.impl.codec.PacketDecoder;
import mc.server.network.impl.codec.PacketEncoder;
import mc.server.network.impl.codec.PacketSplitter;
import mc.server.network.impl.handler.HandshakeHandler;
import mc.server.network.impl.handler.CommonPacketHandler;
import mc.server.reactive.RxProvider;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
public class NetworkModule extends AbstractModule {
@@ -33,6 +34,7 @@ public class NetworkModule extends AbstractModule {
bind(Server.class).to(NettyServer.class).in(Singleton.class);
bind(EventLoopGroup.class).annotatedWith(Names.named("bossGroup")).toInstance(new NioEventLoopGroup(1));
bind(EventLoopGroup.class).annotatedWith(Names.named("workerGroup")).toInstance(new NioEventLoopGroup());
bind(RxProvider.class).toInstance(new RxProvider());
}
@Provides
@@ -51,14 +53,14 @@ public class NetworkModule extends AbstractModule {
@Provides
@Named("channelHandlerMap")
Map<String, ChannelHandler> channelHandlerMap(HandshakeHandler handshakeHandler) {
final Map<String, ChannelHandler> map = new HashMap<>();
Map<String, ChannelHandler> channelHandlerMap(CommonPacketHandler commonPacketHandler) {
final Map<String, ChannelHandler> map = new LinkedHashMap<>();
map.put("logger", new LoggingHandler());
map.put("packet_splitter", new PacketSplitter(new ProtocolSplitter()));
map.put("packet_decoder", new PacketDecoder(new ProtocolDecoder(PacketDirection.SERVER_BOUND)));
map.put("packet_encoder", new PacketEncoder(new ProtocolEncoder(PacketDirection.CLIENT_BOUND)));
map.put("handshake_handler", handshakeHandler);
map.put("common_handler", commonPacketHandler);
return map;
}

15
server/src/main/java/mc/server/network/impl/NettyServer.java
View File

@@ -3,17 +3,28 @@ package mc.server.network.impl;
import com.google.inject.Inject;
import com.google.inject.Provider;
import io.netty.bootstrap.ServerBootstrap;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import mc.server.logic.HandshakeLogic;
import mc.server.logic.LoginLogic;
import mc.server.logic.StatusLogic;
import mc.server.network.Server;
@Slf4j
@RequiredArgsConstructor(onConstructor = @__({ @Inject }))
public class NettyServer implements Server {
@Inject
private Provider<ServerBootstrap> serverBootstrapProvider;
private final Provider<ServerBootstrap> serverBootstrapProvider;
private final HandshakeLogic handshakeLogic;
private final StatusLogic statusLogic;
private final LoginLogic loginLogic;
@Override
public void start(String host, int port) {
handshakeLogic.setup();
statusLogic.setup();
loginLogic.setup();
try {
log.info("Network starting: {}:{}", host, port);
serverBootstrapProvider.get()

15
server/src/main/java/mc/server/network/impl/handler/AbstractPacketHandler.java
View File

@@ -1,15 +0,0 @@
package mc.server.network.impl.handler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import mc.protocol.Packet;
public abstract class AbstractPacketHandler<P extends Packet> extends SimpleChannelInboundHandler<P> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, P msg) throws Exception {
channelRead1(ctx, msg);
}
protected abstract void channelRead1(ChannelHandlerContext ctx, P packet) throws Exception;
}

21
server/src/main/java/mc/server/network/impl/handler/CommonPacketHandler.java Normal file
View File

@@ -0,0 +1,21 @@
package mc.server.network.impl.handler;
import com.google.inject.Inject;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.RequiredArgsConstructor;
import mc.protocol.Packet;
import mc.server.reactive.NetworkContainer;
import mc.server.reactive.RxProvider;
@RequiredArgsConstructor(onConstructor = @__({ @Inject }))
public class CommonPacketHandler extends SimpleChannelInboundHandler<Packet> {
private final RxProvider rxProvider;
@Override
protected void channelRead0(ChannelHandlerContext ctx, Packet packet) {
rxProvider.getPacketObserver(packet.getClass())
.onNext(new NetworkContainer<>(ctx.channel(), packet));
}
}

34
server/src/main/java/mc/server/network/impl/handler/HandshakeHandler.java
View File

@@ -1,34 +0,0 @@
package mc.server.network.impl.handler;
import com.google.inject.Inject;
import com.google.inject.Provider;
import io.netty.channel.ChannelHandlerContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import mc.protocol.State;
import mc.protocol.handshake.client.HandshakePacket;
import static mc.server.network.impl.NettyConstants.ATTR_STATE;
@Slf4j
@RequiredArgsConstructor(onConstructor = @__({ @Inject }))
public class HandshakeHandler extends AbstractPacketHandler<HandshakePacket> {
private final Provider<StatusHandler> statusHandlerProvider;
private final Provider<PingHandler> pingHandlerProvider;
private final Provider<LoginHandler> loginHandlerProvider;
@Override
protected void channelRead1(ChannelHandlerContext ctx, HandshakePacket packet) {
log.info("{}", packet);
ctx.channel().attr(ATTR_STATE).set(packet.getNextState());
if (packet.getNextState() == State.STATUS) {
ctx.channel().pipeline().replace("handshake_handler", "status_handler", statusHandlerProvider.get());
ctx.channel().pipeline().addAfter("status_handler", "ping_handler", pingHandlerProvider.get());
} else if (packet.getNextState() == State.LOGIN) {
ctx.channel().pipeline().replace("handshake_handler", "login_handler", loginHandlerProvider.get());
}
}
}

23
server/src/main/java/mc/server/network/impl/handler/LoginHandler.java
View File

@@ -1,23 +0,0 @@
package mc.server.network.impl.handler;
import io.netty.channel.ChannelHandlerContext;
import mc.protocol.login.client.LoginStartPacket;
import mc.protocol.login.server.DisconnectPacket;
import mc.protocol.text.Text;
import mc.protocol.text.TextColor;
import mc.protocol.text.TextStyle;
public class LoginHandler extends AbstractPacketHandler<LoginStartPacket> {
@Override
protected void channelRead1(ChannelHandlerContext ctx, LoginStartPacket packet) {
DisconnectPacket disconnectPacket = new DisconnectPacket();
disconnectPacket.setReason(Text.builder()
.append(Text.of(TextColor.WHITE, "Server is "))
.color(TextColor.RED).style(TextStyle.BOLD).append("NOT ")
.color(TextColor.WHITE).append("available.")
.build());
ctx.channel().writeAndFlush(disconnectPacket).channel().disconnect();
}
}

12
server/src/main/java/mc/server/network/impl/handler/PingHandler.java
View File

@@ -1,12 +0,0 @@
package mc.server.network.impl.handler;
import io.netty.channel.ChannelHandlerContext;
import mc.protocol.status.PingPacket;
public class PingHandler extends AbstractPacketHandler<PingPacket> {
@Override
protected void channelRead1(ChannelHandlerContext ctx, PingPacket packet) {
ctx.writeAndFlush(packet).channel().disconnect();
}
}