Skip to content

Commit cdd239b

Browse files
robertroeserrobertroeser
committed
adding netty tcp support for rs
1 parent 32a526f commit cdd239b

12 files changed

Lines changed: 1011 additions & 2 deletions

File tree

‎reactivesocket-netty/src/main/java/io/reactivesocket/netty/tcp/client/ClientTcpDuplexConnection.java‎

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,148 @@
1+
/**
2+
* Copyright 2015 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.reactivesocket.netty.tcp.client;
17+
18+
import io.netty.bootstrap.Bootstrap;
19+
import io.netty.buffer.ByteBuf;
20+
import io.netty.buffer.Unpooled;
21+
import io.netty.channel.Channel;
22+
import io.netty.channel.ChannelFuture;
23+
import io.netty.channel.ChannelInitializer;
24+
import io.netty.channel.ChannelPipeline;
25+
import io.netty.channel.EventLoopGroup;
26+
import io.netty.channel.socket.SocketChannel;
27+
import io.netty.channel.socket.nio.NioSocketChannel;
28+
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
29+
import io.reactivesocket.DuplexConnection;
30+
import io.reactivesocket.Frame;
31+
import io.reactivesocket.rx.Completable;
32+
import io.reactivesocket.rx.Observable;
33+
import io.reactivesocket.rx.Observer;
34+
import org.agrona.BitUtil;
35+
import org.reactivestreams.Publisher;
36+
import org.reactivestreams.Subscriber;
37+
import org.reactivestreams.Subscription;
38+
39+
import java.io.IOException;
40+
import java.net.InetSocketAddress;
41+
import java.net.SocketAddress;
42+
import java.util.concurrent.CopyOnWriteArrayList;
43+
44+
public class ClientTcpDuplexConnection implements DuplexConnection {
45+
private Channel channel;
46+
47+
private Bootstrap bootstrap;
48+
49+
private final CopyOnWriteArrayList<Observer<Frame>> subjects;
50+
51+
private ClientTcpDuplexConnection(Channel channel, Bootstrap bootstrap, CopyOnWriteArrayList<Observer<Frame>> subjects) {
52+
this.subjects = subjects;
53+
this.channel = channel;
54+
this.bootstrap = bootstrap;
55+
}
56+
57+
public static Publisher<ClientTcpDuplexConnection> create(SocketAddress socketAddress, EventLoopGroup eventLoopGroup) {
58+
if (socketAddress instanceof InetSocketAddress) {
59+
try {
60+
return create(socketAddress, eventLoopGroup);
61+
} catch (Exception e) {
62+
throw new IllegalArgumentException(e.getMessage(), e);
63+
}
64+
} else {
65+
throw new IllegalArgumentException("unknown socket address type => " + socketAddress.getClass());
66+
}
67+
}
68+
69+
public static Publisher<ClientTcpDuplexConnection> create(InetSocketAddress address, EventLoopGroup eventLoopGroup) {
70+
return s -> {
71+
CopyOnWriteArrayList<Observer<Frame>> subjects = new CopyOnWriteArrayList<>();
72+
ReactiveSocketClientHandler clientHandler = new ReactiveSocketClientHandler(subjects);
73+
Bootstrap bootstrap = new Bootstrap();
74+
ChannelFuture connect = bootstrap
75+
.group(eventLoopGroup)
76+
.channel(NioSocketChannel.class)
77+
.handler(new ChannelInitializer<SocketChannel>() {
78+
@Override
79+
protected void initChannel(SocketChannel ch) throws Exception {
80+
ChannelPipeline p = ch.pipeline();
81+
p.addLast(
82+
new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE >> 1, 0, BitUtil.SIZE_OF_INT, -1 * BitUtil.SIZE_OF_INT, 0),
83+
clientHandler
84+
);
85+
}
86+
}).connect(address.getHostName(), address.getPort());
87+
88+
connect.addListener(connectFuture -> {
89+
if (connectFuture.isSuccess()) {
90+
final Channel ch = connect.channel();
91+
s.onNext(new ClientTcpDuplexConnection(ch, bootstrap, subjects));
92+
s.onComplete();
93+
} else {
94+
s.onError(connectFuture.cause());
95+
}
96+
});
97+
};
98+
}
99+
100+
@Override
101+
public final Observable<Frame> getInput() {
102+
return o -> {
103+
o.onSubscribe(() -> subjects.removeIf(s -> s == o));
104+
subjects.add(o);
105+
};
106+
}
107+
108+
@Override
109+
public void addOutput(Publisher<Frame> o, Completable callback) {
110+
o.subscribe(new Subscriber<Frame>() {
111+
@Override
112+
public void onSubscribe(Subscription s) {
113+
s.request(Long.MAX_VALUE);
114+
}
115+
116+
@Override
117+
public void onNext(Frame frame) {
118+
try {
119+
ByteBuf byteBuf = Unpooled.wrappedBuffer(frame.getByteBuffer());
120+
ChannelFuture channelFuture = channel.writeAndFlush(byteBuf);
121+
channelFuture.addListener(future -> {
122+
Throwable cause = future.cause();
123+
if (cause != null) {
124+
callback.error(cause);
125+
}
126+
});
127+
} catch (Throwable t) {
128+
onError(t);
129+
}
130+
}
131+
132+
@Override
133+
public void onError(Throwable t) {
134+
callback.error(t);
135+
}
136+
137+
@Override
138+
public void onComplete() {
139+
callback.success();
140+
}
141+
});
142+
}
143+
144+
@Override
145+
public void close() throws IOException {
146+
channel.close();
147+
}
148+
}

‎reactivesocket-netty/src/main/java/io/reactivesocket/netty/tcp/client/ReactiveSocketClientHandler.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
/**
2+
* Copyright 2015 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.reactivesocket.netty.tcp.client;
17+
18+
import io.netty.buffer.ByteBuf;
19+
import io.netty.channel.ChannelHandler;
20+
import io.netty.channel.ChannelHandlerContext;
21+
import io.netty.channel.ChannelInboundHandlerAdapter;
22+
import io.reactivesocket.Frame;
23+
import io.reactivesocket.netty.MutableDirectByteBuf;
24+
import io.reactivesocket.rx.Observer;
25+
26+
import java.util.concurrent.CopyOnWriteArrayList;
27+
28+
@ChannelHandler.Sharable
29+
public class ReactiveSocketClientHandler extends ChannelInboundHandlerAdapter {
30+
31+
private final CopyOnWriteArrayList<Observer<Frame>> subjects;
32+
33+
public ReactiveSocketClientHandler(CopyOnWriteArrayList<Observer<Frame>> subjects) {
34+
this.subjects = subjects;
35+
}
36+
37+
@Override
38+
public void channelRead(ChannelHandlerContext ctx, Object content) {
39+
ByteBuf byteBuf = (ByteBuf) content;
40+
try {
41+
MutableDirectByteBuf mutableDirectByteBuf = new MutableDirectByteBuf(byteBuf);
42+
final Frame from = Frame.from(mutableDirectByteBuf, 0, mutableDirectByteBuf.capacity());
43+
subjects.forEach(o -> o.onNext(from));
44+
} finally {
45+
byteBuf.release();
46+
}
47+
}
48+
49+
@Override
50+
public void channelReadComplete(ChannelHandlerContext ctx) {
51+
ctx.flush();
52+
}
53+
54+
@Override
55+
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
56+
// Close the connection when an exception is raised.
57+
cause.printStackTrace();
58+
ctx.close();
59+
}
60+
}

‎reactivesocket-netty/src/main/java/io/reactivesocket/netty/tcp/client/TcpReactiveSocketFactory.java‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
/**
2+
* Copyright 2016 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.reactivesocket.netty.tcp.client;
17+
18+
import io.netty.channel.EventLoopGroup;
19+
import io.reactivesocket.ConnectionSetupPayload;
20+
import io.reactivesocket.DefaultReactiveSocket;
21+
import io.reactivesocket.ReactiveSocket;
22+
import io.reactivesocket.ReactiveSocketFactory;
23+
import io.reactivesocket.ReactiveSocketSocketAddressFactory;
24+
import io.reactivesocket.rx.Completable;
25+
import org.reactivestreams.Publisher;
26+
import org.reactivestreams.Subscriber;
27+
import org.reactivestreams.Subscription;
28+
import org.slf4j.Logger;
29+
import org.slf4j.LoggerFactory;
30+
import rx.Observable;
31+
import rx.RxReactiveStreams;
32+
33+
import java.net.SocketAddress;
34+
import java.util.function.Consumer;
35+
36+
/**
37+
* An implementation of {@link ReactiveSocketFactory} that creates Netty WebSocket ReactiveSockets.
38+
*/
39+
public class TcpReactiveSocketFactory implements ReactiveSocketSocketAddressFactory<ReactiveSocket> {
40+
private static final Logger logger = LoggerFactory.getLogger(TcpReactiveSocketFactory.class);
41+
42+
private final ConnectionSetupPayload connectionSetupPayload;
43+
private final Consumer<Throwable> errorStream;
44+
private final EventLoopGroup eventLoopGroup;
45+
46+
public TcpReactiveSocketFactory(EventLoopGroup eventLoopGroup, ConnectionSetupPayload connectionSetupPayload, Consumer<Throwable> errorStream) {
47+
this.connectionSetupPayload = connectionSetupPayload;
48+
this.errorStream = errorStream;
49+
this.eventLoopGroup = eventLoopGroup;
50+
}
51+
52+
@Override
53+
public Publisher<ReactiveSocket> call(SocketAddress address) {
54+
Publisher<ClientTcpDuplexConnection> connection
55+
= ClientTcpDuplexConnection.create(address, eventLoopGroup);
56+
57+
Observable<ReactiveSocket> result = Observable.create(s ->
58+
connection.subscribe(new Subscriber<ClientTcpDuplexConnection>() {
59+
@Override
60+
public void onSubscribe(Subscription s) {
61+
s.request(1);
62+
}
63+
64+
@Override
65+
public void onNext(ClientTcpDuplexConnection connection) {
66+
ReactiveSocket reactiveSocket = DefaultReactiveSocket.fromClientConnection(connection, connectionSetupPayload, errorStream);
67+
reactiveSocket.start(new Completable() {
68+
@Override
69+
public void success() {
70+
s.onNext(reactiveSocket);
71+
s.onCompleted();
72+
}
73+
74+
@Override
75+
public void error(Throwable e) {
76+
s.onError(e);
77+
}
78+
});
79+
}
80+
81+
@Override
82+
public void onError(Throwable t) {
83+
s.onError(t);
84+
}
85+
86+
@Override
87+
public void onComplete() {
88+
}
89+
})
90+
);
91+
92+
return RxReactiveStreams.toPublisher(result);
93+
}
94+
}

0 commit comments

Comments
 (0)