Skip to content

Commit 51f9518

Browse files
JavaSaBrJavaSaBr
committed
add better way to handle invalid exceptions
1 parent b4c0b95 commit 51f9518

21 files changed

Lines changed: 159 additions & 59 deletions

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ repositories {
1414
}
1515
1616
ext {
17-
rlibVersion = "10.0.alpha7"
17+
rlibVersion = "10.0.alpha8"
1818
}
1919
2020
dependencies {

‎build.gradle‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
rootProject.version = "10.0.alpha7"
1+
rootProject.version = "10.0.alpha8"
22
group = 'javasabr.rlib'
33

44
allprojects {

‎rlib-network/src/loadTest/java/javasabr/rlib/network/StringNetworkLoadTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ void connectAndSendMessages(
6767
ThreadUtils.sleep(random.nextInt(5000));
6868
StringDataConnection connection = network.connect(serverAddress);
6969

70-
connection.onReceive((serverConnection, packet) -> statistics
70+
connection.onReceiveValidPacket((serverConnection, packet) -> statistics
7171
.receivedServerPackersPerSecond()
7272
.accumulate(1));
7373

@@ -141,7 +141,7 @@ void testServerWithMultiplyClients() {
141141
InetSocketAddress serverAddress = serverNetwork.start();
142142

143143
serverNetwork.onAccept(accepted -> accepted
144-
.onReceive((connection, packet) -> {
144+
.onReceiveValidPacket((connection, packet) -> {
145145
StringReadableNetworkPacket<StringDataConnection> receivedPacket = (StringReadableNetworkPacket<StringDataConnection>) packet;
146146
statistics
147147
.receivedClientPackersPerSecond()

‎rlib-network/src/loadTest/java/javasabr/rlib/network/StringSslNetworkLoadTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ void connectAndSendMessages(
7070
ThreadUtils.sleep(random.nextInt(5000));
7171
StringDataSslConnection connection = network.connect(serverAddress);
7272

73-
connection.onReceive((serverConnection, packet) -> statistics
73+
connection.onReceiveValidPacket((serverConnection, packet) -> statistics
7474
.receivedServerPackersPerSecond()
7575
.accumulate(1));
7676

@@ -150,7 +150,7 @@ void testServerWithMultiplyClients() {
150150
InetSocketAddress serverAddress = serverNetwork.start();
151151

152152
serverNetwork.onAccept(accepted -> accepted
153-
.onReceive((connection, packet) -> {
153+
.onReceiveValidPacket((connection, packet) -> {
154154
StringReadableNetworkPacket<StringDataSslConnection> receivedPacket = (StringReadableNetworkPacket<StringDataSslConnection>) packet;
155155
statistics
156156
.receivedClientPackersPerSecond()

‎rlib-network/src/main/java/javasabr/rlib/network/Connection.java‎

Lines changed: 28 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,10 +13,10 @@
1313
*/
1414
public interface Connection<C extends Connection<C>> {
1515

16-
record ReceivedPacketEvent<C, R>(C connection, R packet) {
16+
record ReceivedPacketEvent<C, R>(C connection, R packet, boolean valid) {
1717
@Override
1818
public String toString() {
19-
return "[" + connection + "|" + packet + ']';
19+
return "[" + connection + '|' + packet + '|' + valid + ']';
2020
}
2121
}
2222

@@ -53,9 +53,14 @@ public String toString() {
5353
CompletableFuture<Boolean> sendWithFeedback(WritableNetworkPacket<C> packet);
5454

5555
/**
56-
* Register a consumer to handle received packets.
56+
* Register a consumer to handle received valid packets.
5757
*/
58-
void onReceive(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer);
58+
void onReceiveValidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer);
59+
60+
/**
61+
* Register a consumer to handle received invalid packets.
62+
*/
63+
void onReceiveInvalidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer);
5964

6065
/**
6166
* Get a stream of received packet events.
@@ -72,15 +77,29 @@ default <R extends ReadableNetworkPacket<C>> Flux<ReceivedPacketEvent<C, R>> rec
7277
}
7378

7479
/**
75-
* Get a stream of received packets.
80+
* Get a stream of received valid packets.
81+
*/
82+
Flux<? extends ReadableNetworkPacket<C>> receivedValidPackets();
83+
84+
/**
85+
* Get a stream of received invalid packets.
7686
*/
77-
Flux<? extends ReadableNetworkPacket<C>> receivedPackets();
87+
Flux<? extends ReadableNetworkPacket<C>> receivedInvalidPackets();
88+
89+
/**
90+
* Get a stream of received valid packets with expected type.
91+
*/
92+
default <R extends ReadableNetworkPacket<C>> Flux<R> receivedValidPackets(Class<R> packetType) {
93+
return receivedValidPackets()
94+
.filter(packetType::isInstance)
95+
.map(networkPacket -> (R) networkPacket);
96+
}
7897

7998
/**
80-
* Get a stream of received packets with expected type.
99+
* Get a stream of received invalid packets with expected type.
81100
*/
82-
default <R extends ReadableNetworkPacket<C>> Flux<R> receivedPackets(Class<R> packetType) {
83-
return receivedPackets()
101+
default <R extends ReadableNetworkPacket<C>> Flux<R> receivedInvalidPackets(Class<R> packetType) {
102+
return receivedInvalidPackets()
84103
.filter(packetType::isInstance)
85104
.map(networkPacket -> (R) networkPacket);
86105
}

‎rlib-network/src/main/java/javasabr/rlib/network/exception/MalformedProtocolException.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package javasabr.rlib.network.exception;
22

3-
public class MalformedProtocolException extends RuntimeException {
3+
public class MalformedProtocolException extends NetworkException {
44
public MalformedProtocolException(String message) {
55
super(message);
66
}

‎rlib-network/src/main/java/javasabr/rlib/network/exception/NetworkException.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package javasabr.rlib.network.exception;
2+
3+
public class NetworkException extends RuntimeException {
4+
5+
protected NetworkException(String message) {
6+
super(message);
7+
}
8+
9+
protected NetworkException(String message, Throwable cause) {
10+
super(message, cause);
11+
}
12+
13+
protected NetworkException(Throwable cause) {
14+
super(cause);
15+
}
16+
}

‎rlib-network/src/main/java/javasabr/rlib/network/exception/UserDefinedNetworkException.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package javasabr.rlib.network.exception;
2+
3+
public class UserDefinedNetworkException extends NetworkException {
4+
protected UserDefinedNetworkException(String message) {
5+
super(message);
6+
}
7+
8+
protected UserDefinedNetworkException(String message, Throwable cause) {
9+
super(message, cause);
10+
}
11+
12+
protected UserDefinedNetworkException(Throwable cause) {
13+
super(cause);
14+
}
15+
}

‎rlib-network/src/main/java/javasabr/rlib/network/impl/AbstractConnection.java‎

Lines changed: 48 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,8 @@ public WritablePacketWithFeedback(CompletableFuture<Boolean> attachment, Writabl
6262
final StampedLock lock;
6363
final AtomicBoolean closed;
6464

65-
final MutableArray<BiConsumer<C, ? super ReadableNetworkPacket<C>>> subscribers;
65+
final MutableArray<BiConsumer<C, ? super ReadableNetworkPacket<C>>> validPacketSubscribers;
66+
final MutableArray<BiConsumer<C, ? super ReadableNetworkPacket<C>>> invalidPacketSubscribers;
6667

6768
final int maxPacketsByRead;
6869

@@ -81,7 +82,8 @@ public AbstractConnection(
8182
this.pendingPackets = DequeFactory.arrayBasedBased(WritableNetworkPacket.class);
8283
this.network = network;
8384
this.closed = new AtomicBoolean(false);
84-
this.subscribers = ArrayFactory.copyOnModifyArray(BiConsumer.class);
85+
this.validPacketSubscribers = ArrayFactory.copyOnModifyArray(BiConsumer.class);
86+
this.invalidPacketSubscribers = ArrayFactory.copyOnModifyArray(BiConsumer.class);
8587
this.remoteAddress = String.valueOf(NetworkUtils.getRemoteAddress(channel));
8688
}
8789

@@ -93,8 +95,14 @@ public void onConnected() {}
9395
protected abstract NetworkPacketWriter packetWriter();
9496

9597
@Override
96-
public void onReceive(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer) {
97-
subscribers.add(consumer);
98+
public void onReceiveValidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer) {
99+
validPacketSubscribers.add(consumer);
100+
packetReader().startRead();
101+
}
102+
103+
@Override
104+
public void onReceiveInvalidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer) {
105+
invalidPacketSubscribers.add(consumer);
98106
packetReader().startRead();
99107
}
100108

@@ -104,26 +112,43 @@ public void onReceive(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer)
104112
}
105113

106114
@Override
107-
public Flux<? extends ReadableNetworkPacket<C>> receivedPackets() {
108-
return Flux.create(this::registerFluxOnReceivedPackets);
115+
public Flux<? extends ReadableNetworkPacket<C>> receivedValidPackets() {
116+
return Flux.create(this::registerFluxOnReceivedValidPackets);
117+
}
118+
119+
@Override
120+
public Flux<? extends ReadableNetworkPacket<C>> receivedInvalidPackets() {
121+
return Flux.create(this::registerFluxOnReceivedInvalidPackets);
109122
}
110123

111124
protected void registerFluxOnReceivedEvents(
112125
FluxSink<ReceivedPacketEvent<C, ? extends ReadableNetworkPacket<C>>> sink) {
113126

114-
BiConsumer<C, ReadableNetworkPacket<C>> listener =
127+
BiConsumer<C, ReadableNetworkPacket<C>> validListener =
115128
(connection, packet) -> sink.next(new ReceivedPacketEvent<>(connection,
116-
packet));
129+
packet, true));
130+
BiConsumer<C, ReadableNetworkPacket<C>> invalidListener =
131+
(connection, packet) -> sink.next(new ReceivedPacketEvent<>(connection,
132+
packet, true));
133+
117134

118-
onReceive(listener);
135+
onReceiveValidPacket(validListener);
136+
onReceiveInvalidPacket(invalidListener);
119137

120-
sink.onDispose(() -> subscribers.remove(listener));
138+
sink.onDispose(() -> validPacketSubscribers.remove(validListener));
139+
sink.onDispose(() -> invalidPacketSubscribers.remove(invalidListener));
121140
}
122141

123-
protected void registerFluxOnReceivedPackets(FluxSink<? super ReadableNetworkPacket<C>> sink) {
142+
protected void registerFluxOnReceivedValidPackets(FluxSink<? super ReadableNetworkPacket<C>> sink) {
124143
BiConsumer<C, ReadableNetworkPacket<C>> listener = (connection, packet) -> sink.next(packet);
125-
onReceive(listener);
126-
sink.onDispose(() -> subscribers.remove(listener));
144+
onReceiveValidPacket(listener);
145+
sink.onDispose(() -> validPacketSubscribers.remove(listener));
146+
}
147+
148+
protected void registerFluxOnReceivedInvalidPackets(FluxSink<? super ReadableNetworkPacket<C>> sink) {
149+
BiConsumer<C, ReadableNetworkPacket<C>> listener = (connection, packet) -> sink.next(packet);
150+
onReceiveInvalidPacket(listener);
151+
sink.onDispose(() -> invalidPacketSubscribers.remove(listener));
127152
}
128153

129154
@Nullable
@@ -169,9 +194,16 @@ public boolean closed() {
169194

170195
protected void serializedPacket(WritableNetworkPacket<?> packet) {}
171196

172-
protected void handleReceivedPacket(ReadableNetworkPacket<C> packet) {
173-
log.debug(packet, remoteAddress, "Handle received packet:[%s] from:[%s]"::formatted);
174-
subscribers
197+
protected void handleReceivedValidPacket(ReadableNetworkPacket<C> packet) {
198+
log.debug(packet, remoteAddress, "Handle received valid packet:[%s] from:[%s]"::formatted);
199+
validPacketSubscribers
200+
.iterations()
201+
.forEach((C) this, packet, BiConsumer::accept);
202+
}
203+
204+
protected void handleReceivedInvalidPacket(ReadableNetworkPacket<C> packet) {
205+
log.debug(packet, remoteAddress, "Handle failed received packet:[%s] from:[%s]"::formatted);
206+
invalidPacketSubscribers
175207
.iterations()
176208
.forEach((C) this, packet, BiConsumer::accept);
177209
}

‎rlib-network/src/main/java/javasabr/rlib/network/impl/DefaultDataConnection.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,8 @@ protected NetworkPacketReader createPacketReader() {
4444
return new DefaultNetworkPacketReader<>(
4545
(C) this,
4646
this::updateLastActivity,
47-
this::handleReceivedPacket,
47+
this::handleReceivedValidPacket,
48+
this::handleReceivedInvalidPacket,
4849
value -> createReadablePacket(),
4950
packetLengthHeaderSize,
5051
maxPacketsByRead);

0 commit comments

Comments
 (0)