Skip to content

Commit ae3f7e2

Browse files
JavaSaBrJavaSaBr
committed
implement ssl prototype
1 parent 64af36b commit ae3f7e2

36 files changed

Lines changed: 1495 additions & 226 deletions

‎rlib-common/src/main/java/com/ss/rlib/common/function/NotNullSafeTriFunction.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
package com.ss.rlib.common.function;
2+
3+
import org.jetbrains.annotations.NotNull;
4+
5+
@FunctionalInterface
6+
public interface NotNullSafeTriFunction<F, S, T, R> extends SafeTriFunction<F, S, T, R> {
7+
8+
@Override
9+
@NotNull R apply(@NotNull F first, @NotNull S second, @NotNull T third) throws Exception;
10+
}

‎rlib-common/src/main/java/com/ss/rlib/common/function/NullableSupplier.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
package com.ss.rlib.common.function;
2+
3+
import org.jetbrains.annotations.Nullable;
4+
5+
import java.util.function.Supplier;
6+
7+
@FunctionalInterface
8+
public interface NullableSupplier<T> extends Supplier<T> {
9+
10+
@Override
11+
@Nullable T get();
12+
}

‎rlib-common/src/main/java/com/ss/rlib/common/function/SafeTriFunction.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
package com.ss.rlib.common.function;
2+
3+
import org.jetbrains.annotations.Nullable;
4+
5+
@FunctionalInterface
6+
public interface SafeTriFunction<F, S, T, R> {
7+
8+
@Nullable R apply(@Nullable F first, @Nullable S second, @Nullable T third) throws Exception;
9+
}

‎rlib-common/src/main/java/com/ss/rlib/common/util/Utils.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -264,6 +264,34 @@ public static <F, S> void unchecked(
264264
}
265265
}
266266

267+
/**
268+
* Execute the function with auto-converting a checked exception to an unchecked.
269+
*
270+
* @param first the first argument.
271+
* @param second the second argument.
272+
* @param function the function.
273+
* @param <F> the first's type.
274+
* @param <S> the second's type.
275+
* @param <T> the third's type.
276+
* @param <R> the result's type.
277+
* @return the result.
278+
* @since 9.9.0
279+
*/
280+
public static <F, S, T, R> @NotNull R uncheckedGet(
281+
@NotNull F first,
282+
@NotNull S second,
283+
@NotNull T third,
284+
@NotNull NotNullSafeTriFunction<F, S, T, R> function
285+
) {
286+
try {
287+
return function.apply(first, second, third);
288+
} catch (IOException e) {
289+
throw new UncheckedIOException(e);
290+
} catch (Exception e) {
291+
throw new RuntimeException(e);
292+
}
293+
}
294+
267295
/**
268296
* Try to execute a function with some result.
269297
*

‎rlib-logger-api/src/main/java/com/ss/rlib/logger/api/Logger.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package com.ss.rlib.logger.api;
22

33
import org.jetbrains.annotations.NotNull;
4+
import org.jetbrains.annotations.Nullable;
45

56
/**
67
* The interface to implement a logger.
@@ -33,6 +34,12 @@ interface BiFactory<F, S> {
3334
@NotNull String make(@NotNull F first, @NotNull S second);
3435
}
3536

37+
@FunctionalInterface
38+
interface NullableBiFactory<F, S> {
39+
40+
@NotNull String make(@Nullable F first, @Nullable S second);
41+
}
42+
3643
@FunctionalInterface
3744
interface ObjIntFactory<F> {
3845

@@ -98,6 +105,23 @@ default <F, S> void debug(
98105
print(LoggerLevel.DEBUG, first, second, messageFactory);
99106
}
100107

108+
/**
109+
* Print a build debug message.
110+
*
111+
* @param first the first arg for the message factory.
112+
* @param second the second arg for the message factory.
113+
* @param messageFactory the message factory.
114+
* @param <F> the first argument's type.
115+
* @param <S> the second argument's type.
116+
*/
117+
default <F, S> void debugNullable(
118+
@Nullable F first,
119+
@Nullable S second,
120+
@NotNull Logger.NullableBiFactory<F, S> messageFactory
121+
) {
122+
print(LoggerLevel.DEBUG, first, second, messageFactory);
123+
}
124+
101125
/**
102126
* Print a build debug message.
103127
*
@@ -366,6 +390,27 @@ default <F, S> void print(
366390
}
367391
}
368392

393+
/**
394+
* Print a build message.
395+
*
396+
* @param level the level of the message.
397+
* @param first the first arg for the message factory.
398+
* @param second the second arg for the message factory.
399+
* @param messageFactory the message factory.
400+
* @param <F> the first argument's type.
401+
* @param <S> the second argument's type.
402+
*/
403+
default <F, S> void print(
404+
@NotNull LoggerLevel level,
405+
@Nullable F first,
406+
@Nullable S second,
407+
@NotNull Logger.NullableBiFactory<F, S> messageFactory
408+
) {
409+
if (isEnabled(level)) {
410+
print(level, messageFactory.make(first, second));
411+
}
412+
}
413+
369414
/**
370415
* Print a build message.
371416
*

‎rlib-logger-impl/src/main/java/com/ss/rlib/logger/impl/DefaultLoggerFactory.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,11 @@ private void write(@NotNull LoggerLevel level, @NotNull String resultMessage) {
117117
listeners.forEachInReadLockR(resultMessage, LoggerListener::println);
118118
writers.forEachInReadLockR(resultMessage, DefaultLoggerFactory::append);
119119

120-
System.err.println(resultMessage);
120+
try {
121+
System.err.write((resultMessage + "\n").getBytes());
122+
} catch (IOException e) {
123+
throw new RuntimeException(e);
124+
}
121125

122126
if (!level.isForceFlush()) {
123127
return;

‎rlib-network/src/main/java/com/ss/rlib/network/Connection.java‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,13 @@
11
package com.ss.rlib.network;
22

3+
import com.ss.rlib.common.function.NotNullBiConsumer;
34
import com.ss.rlib.network.packet.ReadablePacket;
45
import com.ss.rlib.network.packet.WritablePacket;
56
import lombok.AllArgsConstructor;
67
import org.jetbrains.annotations.NotNull;
78
import reactor.core.publisher.Flux;
89

910
import java.util.concurrent.CompletableFuture;
10-
import java.util.function.BiConsumer;
1111

1212
/**
1313
* The interface to implement an async connection.
@@ -69,7 +69,7 @@ class ReceivedPacketEvent<C extends Connection<?, ?>, R extends ReadablePacket>
6969
*
7070
* @param consumer the consumer.
7171
*/
72-
void onReceive(@NotNull BiConsumer<? super Connection<R, W>, ? super R> consumer);
72+
void onReceive(@NotNull NotNullBiConsumer<? super Connection<R, W>, ? super R> consumer);
7373

7474
/**
7575
* Get a stream of received packet events.
@@ -84,11 +84,4 @@ class ReceivedPacketEvent<C extends Connection<?, ?>, R extends ReadablePacket>
8484
* @return the stream of received packets.
8585
*/
8686
@NotNull Flux<? extends R> receivedPackets();
87-
88-
/**
89-
* Get a used implementation of a data crypt.
90-
*
91-
* @return the used implementation of a data crypt.
92-
*/
93-
@NotNull NetworkCryptor getCrypt();
9487
}

‎rlib-network/src/main/java/com/ss/rlib/network/NetworkFactory.java‎

Lines changed: 27 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,14 @@
55
import com.ss.rlib.network.impl.DefaultBufferAllocator;
66
import com.ss.rlib.network.impl.DefaultConnection;
77
import com.ss.rlib.network.impl.StringDataConnection;
8+
import com.ss.rlib.network.impl.StringDataSSLConnection;
89
import com.ss.rlib.network.packet.impl.DefaultReadablePacket;
910
import com.ss.rlib.network.packet.registry.ReadablePacketRegistry;
1011
import com.ss.rlib.network.server.ServerNetwork;
1112
import com.ss.rlib.network.server.impl.DefaultServerNetwork;
1213
import org.jetbrains.annotations.NotNull;
1314

15+
import javax.net.ssl.SSLContext;
1416
import java.nio.channels.AsynchronousSocketChannel;
1517
import java.util.function.BiFunction;
1618

@@ -21,14 +23,14 @@
2123
*/
2224
public final class NetworkFactory {
2325

24-
public static <C extends Connection<?, ?>> @NotNull ClientNetwork<C> newClientNetwork(
26+
public static <C extends UnsafeConnection<?, ?>> @NotNull ClientNetwork<C> newClientNetwork(
2527
@NotNull NetworkConfig networkConfig,
2628
@NotNull BiFunction<Network<C>, AsynchronousSocketChannel, C> channelToConnection
2729
) {
2830
return new DefaultClientNetwork<>(networkConfig, channelToConnection);
2931
}
3032

31-
public static <C extends Connection<?, ?>> @NotNull ServerNetwork<C> newServerNetwork(
33+
public static <C extends UnsafeConnection<?, ?>> @NotNull ServerNetwork<C> newServerNetwork(
3234
@NotNull ServerNetworkConfig networkConfig,
3335
@NotNull BiFunction<Network<C>, AsynchronousSocketChannel, C> channelToConnection
3436
) {
@@ -107,9 +109,9 @@ public final class NetworkFactory {
107109
) {
108110
return newClientNetwork(
109111
networkConfig,
110-
(network, channel) -> new DefaultConnection(network,
112+
(network, channel) -> new DefaultConnection(
113+
network,
111114
channel,
112-
NetworkCryptor.NULL,
113115
bufferAllocator,
114116
packetRegistry
115117
)
@@ -157,6 +159,25 @@ public final class NetworkFactory {
157159
);
158160
}
159161

162+
/**
163+
* Create string packet based asynchronous secure server network.
164+
*
165+
* @param networkConfig the network config.
166+
* @param bufferAllocator the buffer allocator.
167+
* @param sslContext the ssl context.
168+
* @return the server network.
169+
*/
170+
public static @NotNull ServerNetwork<StringDataSSLConnection> newStringDataSSLServerNetwork(
171+
@NotNull ServerNetworkConfig networkConfig,
172+
@NotNull BufferAllocator bufferAllocator,
173+
@NotNull SSLContext sslContext
174+
) {
175+
return newServerNetwork(
176+
networkConfig,
177+
(network, channel) -> new StringDataSSLConnection(network, channel, bufferAllocator, sslContext, false)
178+
);
179+
}
180+
160181
/**
161182
* Create id based packet default asynchronous server network.
162183
*
@@ -188,9 +209,9 @@ public final class NetworkFactory {
188209
) {
189210
return newServerNetwork(
190211
networkConfig,
191-
(network, channel) -> new DefaultConnection(network,
212+
(network, channel) -> new DefaultConnection(
213+
network,
192214
channel,
193-
NetworkCryptor.NULL,
194215
bufferAllocator,
195216
packetRegistry
196217
)

‎rlib-network/src/main/java/com/ss/rlib/network/UnsafeConnection.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
package com.ss.rlib.network;
2+
3+
import com.ss.rlib.network.packet.ReadablePacket;
4+
import com.ss.rlib.network.packet.WritablePacket;
5+
6+
public interface UnsafeConnection<R extends ReadablePacket, W extends WritablePacket> extends Connection<R, W> {
7+
8+
void onConnected();
9+
10+
}

‎rlib-network/src/main/java/com/ss/rlib/network/impl/AbstractConnection.java‎

Lines changed: 19 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import static com.ss.rlib.common.util.Utils.unchecked;
44
import static com.ss.rlib.network.util.NetworkUtils.getSocketAddress;
5+
import com.ss.rlib.common.function.NotNullBiConsumer;
56
import com.ss.rlib.common.util.array.Array;
67
import com.ss.rlib.common.util.array.ArrayFactory;
78
import com.ss.rlib.common.util.linkedlist.LinkedList;
@@ -11,7 +12,7 @@
1112
import com.ss.rlib.network.BufferAllocator;
1213
import com.ss.rlib.network.Connection;
1314
import com.ss.rlib.network.Network;
14-
import com.ss.rlib.network.NetworkCryptor;
15+
import com.ss.rlib.network.UnsafeConnection;
1516
import com.ss.rlib.network.packet.PacketReader;
1617
import com.ss.rlib.network.packet.PacketWriter;
1718
import com.ss.rlib.network.packet.ReadablePacket;
@@ -37,7 +38,7 @@
3738
* @author JavaSaBr
3839
*/
3940
public abstract class AbstractConnection<R extends ReadablePacket, W extends WritablePacket> implements
40-
Connection<R, W> {
41+
UnsafeConnection<R, W> {
4142

4243
private static final Logger LOGGER = LoggerManager.getLogger(AbstractConnection.class);
4344

@@ -49,19 +50,18 @@ public WritablePacketWithFeedback(@NotNull CompletableFuture<Boolean> attachment
4950
}
5051
}
5152

52-
protected final @Getter NetworkCryptor crypt;
53-
protected final @Getter String remoteAddress;
53+
protected final @Getter @NotNull String remoteAddress;
5454

55-
protected final Network<? extends Connection<R, W>> network;
56-
protected final BufferAllocator bufferAllocator;
57-
protected final AsynchronousSocketChannel channel;
58-
protected final LinkedList<WritablePacket> pendingPackets;
59-
protected final StampedLock lock;
55+
protected final @NotNull Network<? extends Connection<R, W>> network;
56+
protected final @NotNull BufferAllocator bufferAllocator;
57+
protected final @NotNull AsynchronousSocketChannel channel;
58+
protected final @NotNull LinkedList<WritablePacket> pendingPackets;
59+
protected final @NotNull StampedLock lock;
6060

61-
protected final AtomicBoolean isWriting;
62-
protected final AtomicBoolean closed;
61+
protected final @NotNull AtomicBoolean isWriting;
62+
protected final @NotNull AtomicBoolean closed;
6363

64-
protected final Array<BiConsumer<? super Connection<R, W>, ? super R>> subscribers;
64+
protected final @NotNull Array<NotNullBiConsumer<? super Connection<R, W>, ? super R>> subscribers;
6565

6666
protected final int maxPacketsByRead;
6767

@@ -70,23 +70,24 @@ public WritablePacketWithFeedback(@NotNull CompletableFuture<Boolean> attachment
7070
public AbstractConnection(
7171
@NotNull Network<? extends Connection<R, W>> network,
7272
@NotNull AsynchronousSocketChannel channel,
73-
@NotNull NetworkCryptor crypt,
7473
@NotNull BufferAllocator bufferAllocator,
7574
int maxPacketsByRead
7675
) {
7776
this.bufferAllocator = bufferAllocator;
7877
this.maxPacketsByRead = maxPacketsByRead;
7978
this.lock = new StampedLock();
80-
this.crypt = crypt;
8179
this.channel = channel;
8280
this.pendingPackets = LinkedListFactory.newLinkedList(WritablePacket.class);
8381
this.network = network;
8482
this.isWriting = new AtomicBoolean(false);
8583
this.closed = new AtomicBoolean(false);
86-
this.subscribers = ArrayFactory.newCopyOnModifyArray(BiConsumer.class);
84+
this.subscribers = ArrayFactory.newCopyOnModifyArray(NotNullBiConsumer.class);
8785
this.remoteAddress = String.valueOf(NetworkUtils.getSocketAddress(channel));
8886
}
8987

88+
@Override
89+
public void onConnected() {}
90+
9091
protected abstract @NotNull PacketReader getPacketReader();
9192

9293
protected abstract @NotNull PacketWriter getPacketWriter();
@@ -97,7 +98,7 @@ protected void handleReadPacket(@NotNull R packet) {
9798
}
9899

99100
@Override
100-
public void onReceive(@NotNull BiConsumer<? super Connection<R, W>, ? super R> consumer) {
101+
public void onReceive(@NotNull NotNullBiConsumer<? super Connection<R, W>, ? super R> consumer) {
101102
subscribers.add(consumer);
102103
getPacketReader().startRead();
103104
}
@@ -116,7 +117,7 @@ protected void registerFluxOnReceivedEvents(
116117
@NotNull FluxSink<ReceivedPacketEvent<? extends Connection<R, W>, ? extends R>> sink
117118
) {
118119

119-
BiConsumer<Connection<R, W>, R> listener =
120+
NotNullBiConsumer<Connection<R, W>, R> listener =
120121
(connection, packet) -> sink.next(new ReceivedPacketEvent<>(connection, packet));
121122

122123
onReceive(listener);
@@ -126,7 +127,7 @@ protected void registerFluxOnReceivedEvents(
126127

127128
protected void registerFluxOnReceivedPackets(@NotNull FluxSink<? super R> sink) {
128129

129-
BiConsumer<Connection<R, W>, R> listener = (connection, packet) -> sink.next(packet);
130+
NotNullBiConsumer<Connection<R, W>, R> listener = (connection, packet) -> sink.next(packet);
130131

131132
onReceive(listener);
132133

0 commit comments

Comments
 (0)