Skip to content

Commit f03f832

Browse files
yschimkeyschimke
authored
Plugins refactor (#306)
Plugin registry per client/server builder, allowing connection client and server interceptors
1 parent 96110eb commit f03f832

13 files changed

Lines changed: 253 additions & 128 deletions

File tree

‎rsocket-core/src/main/java/io/rsocket/Plugins.java‎

Lines changed: 0 additions & 49 deletions
This file was deleted.

‎rsocket-core/src/main/java/io/rsocket/RSocketFactory.java‎

Lines changed: 65 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,10 @@
55
import io.rsocket.frame.SetupFrameFlyweight;
66
import io.rsocket.frame.VersionFlyweight;
77
import io.rsocket.internal.ClientServerInputMultiplexer;
8+
import io.rsocket.plugins.DuplexConnectionInterceptor;
9+
import io.rsocket.plugins.PluginRegistry;
10+
import io.rsocket.plugins.Plugins;
11+
import io.rsocket.plugins.RSocketInterceptor;
812
import io.rsocket.transport.ClientTransport;
913
import io.rsocket.transport.ServerTransport;
1014
import io.rsocket.util.PayloadImpl;
@@ -105,6 +109,7 @@ class ClientRSocketFactory
105109
private Supplier<io.rsocket.transport.ClientTransport> transportClient;
106110
private Consumer<Throwable> errorConsumer = Throwable::printStackTrace;
107111
private int mtu = 0;
112+
private PluginRegistry plugins = new PluginRegistry(Plugins.defaultPlugins());
108113
private int flags = SetupFrameFlyweight.FLAGS_STRICT_INTERPRETATION;
109114

110115
private Payload setupPayload = PayloadImpl.EMPTY;
@@ -116,6 +121,21 @@ class ClientRSocketFactory
116121
private String dataMineType = "application/binary";
117122
private String metadataMimeType = "application/binary";
118123

124+
public ClientRSocketFactory addConnectionPlugin(DuplexConnectionInterceptor interceptor) {
125+
plugins.addConnectionPlugin(interceptor);
126+
return this;
127+
}
128+
129+
public ClientRSocketFactory addClientPlugin(RSocketInterceptor interceptor) {
130+
plugins.addClientPlugin(interceptor);
131+
return this;
132+
}
133+
134+
public ClientRSocketFactory addServerPlugin(RSocketInterceptor interceptor) {
135+
plugins.addServerPlugin(interceptor);
136+
return this;
137+
}
138+
119139
@Override
120140
public ClientRSocketFactory keepAlive() {
121141
tickPeriod = Duration.ofSeconds(20);
@@ -225,15 +245,13 @@ public Mono<RSocket> start() {
225245
metadataMimeType,
226246
setupPayload);
227247

228-
ClientServerInputMultiplexer multiplexer;
229248
if (mtu > 0) {
230-
multiplexer =
231-
new ClientServerInputMultiplexer(
232-
new FragmentationDuplexConnection(connection, mtu));
233-
} else {
234-
multiplexer = new ClientServerInputMultiplexer(connection);
249+
connection = new FragmentationDuplexConnection(connection, mtu);
235250
}
236251

252+
ClientServerInputMultiplexer multiplexer =
253+
new ClientServerInputMultiplexer(connection, plugins);
254+
237255
RSocketClient rSocketClient =
238256
new RSocketClient(
239257
multiplexer.asClientConnection(),
@@ -243,23 +261,25 @@ public Mono<RSocket> start() {
243261
ackTimeout,
244262
missedAcks);
245263

246-
return Plugins.CLIENT_REACTIVE_SOCKET_INTERCEPTOR
247-
.apply(rSocketClient)
248-
.then(
249-
wrappedClientRSocket -> {
250-
RSocket unwrappedServerSocket =
251-
acceptor.get().apply(wrappedClientRSocket);
252-
return Plugins.SERVER_REACTIVE_SOCKET_INTERCEPTOR
253-
.apply(unwrappedServerSocket)
254-
.doOnNext(
255-
rSocket ->
256-
new RSocketServer(
257-
multiplexer.asServerConnection(),
258-
rSocket,
259-
errorConsumer))
260-
.then(connection.sendOne(setupFrame))
261-
.then(Mono.just(wrappedClientRSocket));
262-
});
264+
Mono<RSocket> wrappedRSocketClient =
265+
Mono.just(rSocketClient).map(plugins::applyClient);
266+
267+
DuplexConnection finalConnection = connection;
268+
return wrappedRSocketClient.then(
269+
wrappedClientRSocket -> {
270+
RSocket unwrappedServerSocket = acceptor.get().apply(wrappedClientRSocket);
271+
272+
Mono<RSocket> wrappedRSocketServer =
273+
Mono.just(unwrappedServerSocket).map(plugins::applyServer);
274+
275+
return wrappedRSocketServer
276+
.doOnNext(
277+
rSocket ->
278+
new RSocketServer(
279+
multiplexer.asServerConnection(), rSocket, errorConsumer))
280+
.then(finalConnection.sendOne(setupFrame))
281+
.then(Mono.just(wrappedClientRSocket));
282+
});
263283
});
264284
}
265285
}
@@ -274,9 +294,25 @@ class ServerRSocketFactory
274294
private Supplier<io.rsocket.transport.ServerTransport> transportServer;
275295
private Consumer<Throwable> errorConsumer = Throwable::printStackTrace;
276296
private int mtu = 0;
297+
private PluginRegistry plugins = new PluginRegistry(Plugins.defaultPlugins());
277298

278299
private ServerRSocketFactory() {}
279300

301+
public ServerRSocketFactory addConnectionPlugin(DuplexConnectionInterceptor interceptor) {
302+
plugins.addConnectionPlugin(interceptor);
303+
return this;
304+
}
305+
306+
public ServerRSocketFactory addClientPlugin(RSocketInterceptor interceptor) {
307+
plugins.addClientPlugin(interceptor);
308+
return this;
309+
}
310+
311+
public ServerRSocketFactory addServerPlugin(RSocketInterceptor interceptor) {
312+
plugins.addServerPlugin(interceptor);
313+
return this;
314+
}
315+
280316
@Override
281317
public Transport<io.rsocket.transport.ServerTransport, Closeable> acceptor(
282318
Supplier<SocketAcceptor> acceptor) {
@@ -312,15 +348,13 @@ public Mono<Closeable> start() {
312348
.get()
313349
.start(
314350
connection -> {
315-
ClientServerInputMultiplexer multiplexer;
316351
if (mtu > 0) {
317-
multiplexer =
318-
new ClientServerInputMultiplexer(
319-
new FragmentationDuplexConnection(connection, mtu));
320-
} else {
321-
multiplexer = new ClientServerInputMultiplexer(connection);
352+
connection = new FragmentationDuplexConnection(connection, mtu);
322353
}
323354

355+
ClientServerInputMultiplexer multiplexer =
356+
new ClientServerInputMultiplexer(connection, plugins);
357+
324358
return multiplexer
325359
.asStreamZeroConnection()
326360
.receive()
@@ -348,16 +382,10 @@ private Mono<? extends Void> processSetupFrame(
348382
new RSocketClient(
349383
multiplexer.asServerConnection(), errorConsumer, StreamIdSupplier.serverSupplier());
350384

351-
Mono<RSocket> wrappedRSocketClient =
352-
Plugins.CLIENT_REACTIVE_SOCKET_INTERCEPTOR.apply(rSocketClient);
385+
Mono<RSocket> wrappedRSocketClient = Mono.just(rSocketClient).map(plugins::applyClient);
353386

354387
return wrappedRSocketClient
355-
.then(
356-
sender ->
357-
acceptor
358-
.get()
359-
.accept(setupPayload, sender)
360-
.then(Plugins.SERVER_REACTIVE_SOCKET_INTERCEPTOR::apply))
388+
.then(sender -> acceptor.get().accept(setupPayload, sender).map(plugins::applyServer))
361389
.map(
362390
handler ->
363391
new RSocketServer(multiplexer.asClientConnection(), handler, errorConsumer))

‎rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,14 @@ public FragmentationDuplexConnection(DuplexConnection source, int mtu) {
2020
this.frameFragmenter = new FrameFragmenter(mtu);
2121
}
2222

23+
public static int getDefaultMTU() {
24+
if (Boolean.getBoolean("io.rsocket.fragmentation.enable")) {
25+
return Integer.getInteger("io.rsocket.fragmentation.mtu", 1024);
26+
}
27+
28+
return 0;
29+
}
30+
2331
@Override
2432
public double availability() {
2533
return source.availability();

‎rsocket-core/src/main/java/io/rsocket/internal/ClientServerInputMultiplexer.java‎

Lines changed: 7 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919
import io.rsocket.DuplexConnection;
2020
import io.rsocket.Frame;
2121
import io.rsocket.FrameType;
22-
import io.rsocket.Plugins;
23-
import io.rsocket.Plugins.DuplexConnectionInterceptor.Type;
22+
import io.rsocket.plugins.DuplexConnectionInterceptor.Type;
23+
import io.rsocket.plugins.PluginRegistry;
2424
import org.reactivestreams.Publisher;
2525
import org.slf4j.Logger;
2626
import org.slf4j.LoggerFactory;
@@ -51,22 +51,19 @@ public class ClientServerInputMultiplexer {
5151
private final DuplexConnection clientConnection;
5252
private final DuplexConnection source;
5353

54-
public ClientServerInputMultiplexer(DuplexConnection source) {
54+
public ClientServerInputMultiplexer(DuplexConnection source, PluginRegistry plugins) {
5555
this.source = source;
5656
final MonoProcessor<Flux<Frame>> streamZero = MonoProcessor.create();
5757
final MonoProcessor<Flux<Frame>> server = MonoProcessor.create();
5858
final MonoProcessor<Flux<Frame>> client = MonoProcessor.create();
5959

60-
source = Plugins.DUPLEX_CONNECTION_INTERCEPTOR.apply(Type.SOURCE, source);
60+
source = plugins.applyConnection(Type.SOURCE, source);
6161
streamZeroConnection =
62-
Plugins.DUPLEX_CONNECTION_INTERCEPTOR.apply(
63-
Type.STREAM_ZERO, new InternalDuplexConnection(source, streamZero));
62+
plugins.applyConnection(Type.STREAM_ZERO, new InternalDuplexConnection(source, streamZero));
6463
serverConnection =
65-
Plugins.DUPLEX_CONNECTION_INTERCEPTOR.apply(
66-
Type.SERVER, new InternalDuplexConnection(source, server));
64+
plugins.applyConnection(Type.SERVER, new InternalDuplexConnection(source, server));
6765
clientConnection =
68-
Plugins.DUPLEX_CONNECTION_INTERCEPTOR.apply(
69-
Type.CLIENT, new InternalDuplexConnection(source, client));
66+
plugins.applyConnection(Type.CLIENT, new InternalDuplexConnection(source, client));
7067

7168
source
7269
.receive()

‎rsocket-core/src/main/java/io/rsocket/plugins/DuplexConnectionInterceptor.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package io.rsocket.plugins;
2+
3+
import io.rsocket.DuplexConnection;
4+
import java.util.function.BiFunction;
5+
6+
/** */
7+
public @FunctionalInterface interface DuplexConnectionInterceptor
8+
extends BiFunction<DuplexConnectionInterceptor.Type, DuplexConnection, DuplexConnection> {
9+
enum Type {
10+
STREAM_ZERO,
11+
CLIENT,
12+
SERVER,
13+
SOURCE
14+
}
15+
}

‎rsocket-core/src/main/java/io/rsocket/plugins/PluginRegistry.java‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
package io.rsocket.plugins;
2+
3+
import io.rsocket.DuplexConnection;
4+
import io.rsocket.RSocket;
5+
import java.util.ArrayList;
6+
import java.util.List;
7+
8+
public class PluginRegistry {
9+
private List<DuplexConnectionInterceptor> connections = new ArrayList<>();
10+
private List<RSocketInterceptor> clients = new ArrayList<>();
11+
private List<RSocketInterceptor> servers = new ArrayList<>();
12+
13+
public PluginRegistry() {}
14+
15+
public PluginRegistry(PluginRegistry defaults) {
16+
this.connections.addAll(defaults.connections);
17+
this.clients.addAll(defaults.clients);
18+
this.servers.addAll(defaults.servers);
19+
}
20+
21+
public void addConnectionPlugin(DuplexConnectionInterceptor interceptor) {
22+
connections.add(interceptor);
23+
}
24+
25+
public void addClientPlugin(RSocketInterceptor interceptor) {
26+
clients.add(interceptor);
27+
}
28+
29+
public void addServerPlugin(RSocketInterceptor interceptor) {
30+
servers.add(interceptor);
31+
}
32+
33+
public RSocket applyClient(RSocket rSocket) {
34+
for (RSocketInterceptor i : clients) {
35+
rSocket = i.apply(rSocket);
36+
}
37+
38+
return rSocket;
39+
}
40+
41+
public RSocket applyServer(RSocket rSocket) {
42+
for (RSocketInterceptor i : servers) {
43+
rSocket = i.apply(rSocket);
44+
}
45+
46+
return rSocket;
47+
}
48+
49+
public DuplexConnection applyConnection(
50+
DuplexConnectionInterceptor.Type type, DuplexConnection connection) {
51+
for (DuplexConnectionInterceptor i : connections) {
52+
connection = i.apply(type, connection);
53+
}
54+
55+
return connection;
56+
}
57+
}

‎rsocket-core/src/main/java/io/rsocket/plugins/Plugins.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
package io.rsocket.plugins;
2+
3+
/** JVM wide plugins for RSocket */
4+
public class Plugins {
5+
private static PluginRegistry DEFAULT = new PluginRegistry();
6+
7+
private Plugins() {}
8+
9+
public static void interceptConnection(DuplexConnectionInterceptor interceptor) {
10+
DEFAULT.addConnectionPlugin(interceptor);
11+
}
12+
13+
public static void interceptClient(RSocketInterceptor interceptor) {
14+
DEFAULT.addClientPlugin(interceptor);
15+
}
16+
17+
public static void interceptServer(RSocketInterceptor interceptor) {
18+
DEFAULT.addServerPlugin(interceptor);
19+
}
20+
21+
public static PluginRegistry defaultPlugins() {
22+
return DEFAULT;
23+
}
24+
}

‎rsocket-core/src/main/java/io/rsocket/plugins/RSocketInterceptor.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package io.rsocket.plugins;
2+
3+
import io.rsocket.RSocket;
4+
import java.util.function.Function;
5+
6+
/** */
7+
public @FunctionalInterface interface RSocketInterceptor extends Function<RSocket, RSocket> {}

0 commit comments

Comments
 (0)