andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Added SCTP OIO Transport

Browse Source 
- Sctp OioChannel/ServeChannel implementation.
- Changed interrupt handling in OioEventLoop.
- SctpOioEchoClient/Server Example.
This commit is contained in:
Jestan Nirojan 2012-09-21 11:56:13 +08:00
parent 22cbaa489a
commit 47eafa386c
11 changed files with 650 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
example/src/main/java/io/netty/example/sctp/SctpEchoClient.java → example/src/main/java/io/netty/example/sctp/NioSctpEchoClient.java
View File

@ -20,10 +20,8 @@ import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SctpChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSctpChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
@ -37,13 +35,13 @@ import java.net.InetSocketAddress;
* traffic between the echo client and server by sending the first message to
* the server.
*/
public class SctpEchoClient {
public class NioSctpEchoClient {
private final String host;
private final int port;
private final int firstMessageSize;
public SctpEchoClient(String host, int port, int firstMessageSize) {
public NioSctpEchoClient(String host, int port, int firstMessageSize) {
this.host = host;
this.port = port;
this.firstMessageSize = firstMessageSize;
@ -81,7 +79,7 @@ public class SctpEchoClient {
// Print usage if no argument is specified.
if (args.length < 2 || args.length > 3) {
System.err.println(
"Usage: " + SctpEchoClient.class.getSimpleName() +
"Usage: " + NioSctpEchoClient.class.getSimpleName() +
" <host> <port> [<first message size>]");
return;
}
@ -96,6 +94,6 @@ public class SctpEchoClient {
firstMessageSize = 256;
}
new SctpEchoClient(host, port, firstMessageSize).run();
new NioSctpEchoClient(host, port, firstMessageSize).run();
}
}

8
example/src/main/java/io/netty/example/sctp/SctpEchoServer.java → example/src/main/java/io/netty/example/sctp/NioSctpEchoServer.java
View File

@ -20,10 +20,8 @@ import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SctpChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSctpServerChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
@ -32,11 +30,11 @@ import java.net.InetSocketAddress;
/**
* Echoes back any received data from a SCTP client.
*/
public class SctpEchoServer {
public class NioSctpEchoServer {
private final int port;
public SctpEchoServer(int port) {
public NioSctpEchoServer(int port) {
this.port = port;
}
@ -77,6 +75,6 @@ public class SctpEchoServer {
} else {
port = 2556;
}
new SctpEchoServer(port).run();
new NioSctpEchoServer(port).run();
}
}

99
example/src/main/java/io/netty/example/sctp/OioSctpEchoClient.java Normal file
View File

@ -0,0 +1,99 @@
/*
* Copyright 2012 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.example.sctp;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SctpChannel;
import io.netty.channel.socket.oio.OioEventLoopGroup;
import io.netty.channel.socket.oio.OioSctpChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.net.InetSocketAddress;
/**
* Sends one message when a connection is open and echoes back any received
* data to the server over SCTP connection.
*
* Simply put, the echo client initiates the ping-pong
* traffic between the echo client and server by sending the first message to
* the server.
*/
public class OioSctpEchoClient {
private final String host;
private final int port;
private final int firstMessageSize;
public OioSctpEchoClient(String host, int port, int firstMessageSize) {
this.host = host;
this.port = port;
this.firstMessageSize = firstMessageSize;
}
public void run() throws Exception {
// Configure the client.
Bootstrap b = new Bootstrap();
try {
b.group(new OioEventLoopGroup())
.channel(new OioSctpChannel())
.option(ChannelOption.SCTP_NODELAY, true)
.remoteAddress(new InetSocketAddress(host, port))
.handler(new ChannelInitializer<SctpChannel>() {
@Override
public void initChannel(SctpChannel ch) throws Exception {
ch.pipeline().addLast(
new LoggingHandler(LogLevel.INFO),
new SctpEchoClientHandler(firstMessageSize));
}
});
// Start the client.
ChannelFuture f = b.connect().sync();
// Wait until the connection is closed.
f.channel().closeFuture().sync();
} finally {
// Shut down the event loop to terminate all threads.
b.shutdown();
}
}
public static void main(String[] args) throws Exception {
// Print usage if no argument is specified.
if (args.length < 2 || args.length > 3) {
System.err.println(
"Usage: " + OioSctpEchoClient.class.getSimpleName() +
" <host> <port> [<first message size>]");
return;
}
// Parse options.
final String host = "localhost";
final int port = Integer.parseInt(args[1]);
final int firstMessageSize;
if (args.length == 3) {
firstMessageSize = Integer.parseInt(args[2]);
} else {
firstMessageSize = 256;
}
new OioSctpEchoClient(host, port, firstMessageSize).run();
}
}

80
example/src/main/java/io/netty/example/sctp/OioSctpEchoServer.java Normal file
View File

@ -0,0 +1,80 @@
/*
* Copyright 2012 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.example.sctp;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SctpChannel;
import io.netty.channel.socket.oio.OioEventLoopGroup;
import io.netty.channel.socket.oio.OioSctpServerChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.net.InetSocketAddress;
/**
* Echoes back any received data from a SCTP client.
*/
public class OioSctpEchoServer {
private final int port;
public OioSctpEchoServer(int port) {
this.port = port;
}
public void run() throws Exception {
// Configure the server.
ServerBootstrap b = new ServerBootstrap();
try {
b.group(new OioEventLoopGroup(), new OioEventLoopGroup())
.channel(new OioSctpServerChannel())
.option(ChannelOption.SO_BACKLOG, 100)
.localAddress(new InetSocketAddress(port))
.childOption(ChannelOption.SCTP_NODELAY, true)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SctpChannel>() {
@Override
public void initChannel(SctpChannel ch) throws Exception {
ch.pipeline().addLast(
new LoggingHandler(LogLevel.INFO),
new SctpEchoServerHandler());
}
});
// Start the server.
ChannelFuture f = b.bind().sync();
// Wait until the server socket is closed.
f.channel().closeFuture().sync();
} finally {
// Shut down all event loops to terminate all threads.
b.shutdown();
}
}
public static void main(String[] args) throws Exception {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 2556;
}
new OioSctpEchoServer(port).run();
}
}

2
example/src/main/java/io/netty/example/sctp/SctpEchoClientHandler.java
View File

@ -62,8 +62,6 @@ public class SctpEchoClientHandler extends ChannelInboundMessageHandlerAdapter<S
MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
out.add(msg);
ctx.flush();
} else {
logger.log(Level.INFO, "Received SCTP Notification", msg);
}
}

2
example/src/main/java/io/netty/example/sctp/SctpEchoServerHandler.java
View File

@ -47,8 +47,6 @@ public class SctpEchoServerHandler extends ChannelInboundMessageHandlerAdapter<S
MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
out.add(msg);
ctx.flush();
} else {
logger.log(Level.INFO, "Received SCTP Notification", msg);
}
}
}

5
transport/src/main/java/io/netty/channel/socket/SctpNotificationHandler.java
View File

@ -27,11 +27,9 @@ import io.netty.channel.ChannelPipeline;
public class SctpNotificationHandler extends AbstractNotificationHandler<Object> {
private final SctpChannel sctpChannel;
private final ChannelPipeline pipeline;
public SctpNotificationHandler(SctpChannel sctpChannel) {
this.sctpChannel = sctpChannel;
pipeline = sctpChannel.pipeline();
}
@Override
@ -54,12 +52,13 @@ public class SctpNotificationHandler extends AbstractNotificationHandler<Object>
@Override
public HandlerResult handleNotification(ShutdownNotification notification, Object o) {
updateInboundBuffer(notification, o);
sctpChannel.close();
return HandlerResult.RETURN;
}
private void updateInboundBuffer(Notification notification, Object o) {
pipeline.inboundMessageBuffer().add(new SctpNotification(notification, o));
sctpChannel.pipeline().inboundMessageBuffer().add(new SctpNotification(notification, o));
}
}

13
transport/src/main/java/io/netty/channel/socket/nio/NioSctpChannel.java
View File

@ -70,6 +70,8 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
super(parent, id, sctpChannel, SelectionKey.OP_READ);
try {
sctpChannel.configureBlocking(false);
config = new DefaultSctpChannelConfig(sctpChannel);
notificationHandler = new SctpNotificationHandler(this);
} catch (IOException e) {
try {
sctpChannel.close();
@ -83,9 +85,6 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
throw new ChannelException("Failed to enter non-blocking mode.", e);
}
config = new DefaultSctpChannelConfig(sctpChannel);
notificationHandler = new SctpNotificationHandler(this);
}
@Override
@ -245,11 +244,11 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
}
final MessageInfo messageInfo = MessageInfo.createOutgoing(association(), null, packet.getStreamIdentifier());
messageInfo.payloadProtocolID(packet.getProtocolIdentifier());
messageInfo.streamNumber(packet.getStreamIdentifier());
final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.getStreamIdentifier());
mi.payloadProtocolID(packet.getProtocolIdentifier());
mi.streamNumber(packet.getStreamIdentifier());
final int writtenBytes = javaChannel().send(nioData, messageInfo);
final int writtenBytes = javaChannel().send(nioData, mi);
final SelectionKey key = selectionKey();
final int interestOps = key.interestOps();

4
transport/src/main/java/io/netty/channel/socket/oio/OioEventLoop.java
View File

@ -100,7 +100,9 @@ class OioEventLoop extends SingleThreadEventLoop {
@Override
protected void wakeup(boolean inEventLoop) {
interruptThread();
if (!inEventLoop && isShutdown()) {
interruptThread();
}
}
private void deregister() {

251
transport/src/main/java/io/netty/channel/socket/oio/OioSctpChannel.java Executable file
View File

@ -0,0 +1,251 @@
/*
* Copyright 2012 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.channel.socket.oio;
import com.sun.nio.sctp.Association;
import com.sun.nio.sctp.MessageInfo;
import com.sun.nio.sctp.NotificationHandler;
import com.sun.nio.sctp.SctpChannel;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ChannelBufType;
import io.netty.buffer.MessageBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelMetadata;
import io.netty.channel.socket.DefaultSctpChannelConfig;
import io.netty.channel.socket.SctpChannelConfig;
import io.netty.channel.socket.SctpData;
import io.netty.channel.socket.SctpNotificationHandler;
import io.netty.logging.InternalLogger;
import io.netty.logging.InternalLoggerFactory;
import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
public class OioSctpChannel extends AbstractOioMessageChannel
implements io.netty.channel.socket.SctpChannel {
private static final InternalLogger logger =
InternalLoggerFactory.getInstance(OioSctpChannel.class);
private static final ChannelMetadata METADATA = new ChannelMetadata(ChannelBufType.MESSAGE, false);
private final SctpChannel ch;
private final SctpChannelConfig config;
private final NotificationHandler notificationHandler;
private static SctpChannel openChannel() {
try {
return SctpChannel.open();
} catch (IOException e) {
throw new ChannelException("Failed to open a sctp channel.", e);
}
}
public OioSctpChannel() {
this(openChannel());
}
public OioSctpChannel(SctpChannel ch) {
this(null, null, ch);
}
public OioSctpChannel(Channel parent, Integer id, SctpChannel ch) {
super(parent, id);
this.ch = ch;
boolean success = false;
try {
ch.configureBlocking(true);
config = new DefaultSctpChannelConfig(ch);
notificationHandler = new SctpNotificationHandler(this);
success = true;
} catch (Exception e) {
throw new ChannelException("failed to initialize a sctp channel", e);
} finally {
if (!success) {
try {
ch.close();
} catch (IOException e) {
logger.warn("Failed to close a sctp channel.", e);
}
}
}
}
@Override
public ChannelMetadata metadata() {
return METADATA;
}
@Override
public SctpChannelConfig config() {
return config;
}
@Override
public boolean isOpen() {
return ch.isOpen();
}
@Override
protected int doReadMessages(MessageBuf<Object> buf) throws Exception {
if (readSuspended) {
return 0;
}
ByteBuffer data = ByteBuffer.allocate(config().getReceiveBufferSize());
MessageInfo messageInfo = ch.receive(data, null, notificationHandler);
if (messageInfo == null) {
return 0;
}
data.flip();
buf.add(new SctpData(messageInfo, Unpooled.wrappedBuffer(data)));
if (readSuspended) {
return 0;
} else {
return 1;
}
}
@Override
protected void doWriteMessages(MessageBuf<Object> buf) throws Exception {
SctpData packet = (SctpData) buf.poll();
ByteBuf data = packet.getPayloadBuffer();
int dataLen = data.readableBytes();
ByteBuffer nioData;
if (data.hasNioBuffer()) {
nioData = data.nioBuffer();
} else {
nioData = ByteBuffer.allocate(dataLen);
data.getBytes(data.readerIndex(), nioData);
nioData.flip();
}
final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.getStreamIdentifier());
mi.payloadProtocolID(packet.getProtocolIdentifier());
mi.streamNumber(packet.getStreamIdentifier());
ch.send(nioData, mi);
}
@Override
public Association association() {
try {
return ch.association();
} catch (IOException e) {
return null;
}
}
@Override
public boolean isActive() {
return isOpen() && association() != null;
}
@Override
protected SocketAddress localAddress0() {
try {
for (SocketAddress address : ch.getAllLocalAddresses()) {
return address;
}
} catch (IOException e) {
// ignore
}
return null;
}
@Override
public Set<SocketAddress> allLocalAddresses() {
try {
final Set<SocketAddress> allLocalAddresses = ch.getAllLocalAddresses();
final Set<SocketAddress> addresses = new HashSet<SocketAddress>(allLocalAddresses.size());
for (SocketAddress socketAddress : allLocalAddresses) {
addresses.add(socketAddress);
}
return addresses;
} catch (Throwable t) {
return Collections.emptySet();
}
}
@Override
protected SocketAddress remoteAddress0() {
try {
for (SocketAddress address : ch.getRemoteAddresses()) {
return address;
}
} catch (IOException e) {
// ignore
}
return null;
}
@Override
public Set<SocketAddress> allRemoteAddresses() {
try {
final Set<SocketAddress> allLocalAddresses = ch.getRemoteAddresses();
final Set<SocketAddress> addresses = new HashSet<SocketAddress>(allLocalAddresses.size());
for (SocketAddress socketAddress : allLocalAddresses) {
addresses.add(socketAddress);
}
return addresses;
} catch (Throwable t) {
return Collections.emptySet();
}
}
@Override
protected void doBind(SocketAddress localAddress) throws Exception {
ch.bind(localAddress);
}
@Override
protected void doConnect(SocketAddress remoteAddress,
SocketAddress localAddress) throws Exception {
if (localAddress != null) {
ch.bind(localAddress);
}
boolean success = false;
try {
ch.connect(remoteAddress);
success = true;
} finally {
if (!success) {
doClose();
}
}
}
@Override
protected void doDisconnect() throws Exception {
doClose();
}
@Override
protected void doClose() throws Exception {
ch.close();
}
}

202
transport/src/main/java/io/netty/channel/socket/oio/OioSctpServerChannel.java Executable file
View File

@ -0,0 +1,202 @@
/*
* Copyright 2012 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.channel.socket.oio;
import com.sun.nio.sctp.SctpChannel;
import com.sun.nio.sctp.SctpServerChannel;
import io.netty.buffer.ChannelBufType;
import io.netty.buffer.MessageBuf;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelMetadata;
import io.netty.channel.socket.DefaultSctpChannelConfig;
import io.netty.channel.socket.DefaultSctpServerChannelConfig;
import io.netty.channel.socket.SctpNotificationHandler;
import io.netty.channel.socket.SctpServerChannelConfig;
import io.netty.logging.InternalLogger;
import io.netty.logging.InternalLoggerFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
public class OioSctpServerChannel extends AbstractOioMessageChannel
implements io.netty.channel.socket.SctpServerChannel {
private static final InternalLogger logger =
InternalLoggerFactory.getInstance(OioSctpServerChannel.class);
private static final ChannelMetadata METADATA = new ChannelMetadata(ChannelBufType.MESSAGE, false);
private static SctpServerChannel newServerSocket() {
try {
return SctpServerChannel.open();
} catch (IOException e) {
throw new ChannelException("failed to create a sctp server channel", e);
}
}
final SctpServerChannel sch;
private final SctpServerChannelConfig config;
public OioSctpServerChannel() {
this(newServerSocket());
}
public OioSctpServerChannel(SctpServerChannel sch) {
this(null, sch);
}
public OioSctpServerChannel(Integer id, SctpServerChannel sch) {
super(null, id);
if (sch == null) {
throw new NullPointerException("sctp server channel");
}
this.sch = sch;
boolean success = false;
try {
sch.configureBlocking(true);
config = new DefaultSctpServerChannelConfig(sch);
success = true;
} catch (Exception e) {
throw new ChannelException("failed to initialize a sctp server channel", e);
} finally {
if (!success) {
try {
sch.close();
} catch (IOException e) {
logger.warn("Failed to close a sctp server channel.", e);
}
}
}
}
@Override
public ChannelMetadata metadata() {
return METADATA;
}
@Override
public SctpServerChannelConfig config() {
return config;
}
@Override
public InetSocketAddress remoteAddress() {
return null;
}
@Override
public boolean isOpen() {
return sch.isOpen();
}
@Override
protected SocketAddress localAddress0() {
try {
for (SocketAddress address : sch.getAllLocalAddresses()) {
return address;
}
} catch (IOException e) {
// ignore
}
return null;
}
@Override
public Set<SocketAddress> allLocalAddresses() {
try {
final Set<SocketAddress> allLocalAddresses = sch.getAllLocalAddresses();
final Set<SocketAddress> addresses = new HashSet<SocketAddress>(allLocalAddresses.size());
for (SocketAddress socketAddress : allLocalAddresses) {
addresses.add(socketAddress);
}
return addresses;
} catch (Throwable t) {
return Collections.emptySet();
}
}
@Override
public boolean isActive() {
return isOpen() && localAddress0() != null;
}
@Override
protected void doBind(SocketAddress localAddress) throws Exception {
sch.bind(localAddress, config.getBacklog());
}
@Override
protected void doClose() throws Exception {
sch.close();
}
@Override
protected int doReadMessages(MessageBuf<Object> buf) throws Exception {
if (!isActive()) {
return -1;
}
if (readSuspended) {
return 0;
}
SctpChannel s = null;
try {
s = sch.accept();
if (s != null) {
buf.add(new OioSctpChannel(this, null, s));
return 1;
}
} catch (Throwable t) {
logger.warn("Failed to create a new channel from an accepted sctp channel.", t);
if (s != null) {
try {
s.close();
} catch (Throwable t2) {
logger.warn("Failed to close a sctp channel.", t2);
}
}
}
return 0;
}
@Override
protected void doConnect(
SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
throw new UnsupportedOperationException();
}
@Override
protected SocketAddress remoteAddress0() {
return null;
}
@Override
protected void doDisconnect() throws Exception {
throw new UnsupportedOperationException();
}
@Override
protected void doWriteMessages(MessageBuf<Object> buf) throws Exception {
throw new UnsupportedOperationException();
}
}