View Javadoc
1   /*
2    * Copyright 2012 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. 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, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  package io.netty.channel.udt.nio;
17  
18  import com.barchart.udt.TypeUDT;
19  import com.barchart.udt.nio.SocketChannelUDT;
20  import io.netty.buffer.ByteBuf;
21  import io.netty.channel.Channel;
22  import io.netty.channel.ChannelException;
23  import io.netty.channel.ChannelMetadata;
24  import io.netty.channel.ChannelOutboundBuffer;
25  import io.netty.util.internal.SocketUtils;
26  import io.netty.channel.nio.AbstractNioMessageChannel;
27  import io.netty.channel.udt.DefaultUdtChannelConfig;
28  import io.netty.channel.udt.UdtChannel;
29  import io.netty.channel.udt.UdtChannelConfig;
30  import io.netty.channel.udt.UdtMessage;
31  import io.netty.util.internal.logging.InternalLogger;
32  import io.netty.util.internal.logging.InternalLoggerFactory;
33  
34  import java.io.IOException;
35  import java.net.InetSocketAddress;
36  import java.net.SocketAddress;
37  import java.security.AccessController;
38  import java.security.PrivilegedActionException;
39  import java.security.PrivilegedExceptionAction;
40  import java.util.List;
41  
42  import static java.nio.channels.SelectionKey.*;
43  
44  /**
45   * Message Connector for UDT Datagrams.
46   * <p>
47   * Note: send/receive must use {@link UdtMessage} in the pipeline
48   *
49   * @deprecated The UDT transport is no longer maintained and will be removed.
50   */
51  @Deprecated
52  public class NioUdtMessageConnectorChannel extends AbstractNioMessageChannel implements UdtChannel {
53  
54      private static final InternalLogger logger =
55              InternalLoggerFactory.getInstance(NioUdtMessageConnectorChannel.class);
56  
57      private static final ChannelMetadata METADATA = new ChannelMetadata(false);
58  
59      private final UdtChannelConfig config;
60  
61      public NioUdtMessageConnectorChannel() {
62          this(TypeUDT.DATAGRAM);
63      }
64  
65      public NioUdtMessageConnectorChannel(final Channel parent, final SocketChannelUDT channelUDT) {
66          super(parent, channelUDT, OP_READ);
67          try {
68              channelUDT.configureBlocking(false);
69              switch (channelUDT.socketUDT().status()) {
70              case INIT:
71              case OPENED:
72                  config = new DefaultUdtChannelConfig(this, channelUDT, true);
73                  break;
74              default:
75                  config = new DefaultUdtChannelConfig(this, channelUDT, false);
76                  break;
77              }
78          } catch (final Exception e) {
79              try {
80                  channelUDT.close();
81              } catch (final Exception e2) {
82                  if (logger.isWarnEnabled()) {
83                      logger.warn("Failed to close channel.", e2);
84                  }
85              }
86              throw new ChannelException("Failed to configure channel.", e);
87          }
88      }
89  
90      public NioUdtMessageConnectorChannel(final SocketChannelUDT channelUDT) {
91          this(null, channelUDT);
92      }
93  
94      public NioUdtMessageConnectorChannel(final TypeUDT type) {
95          this(NioUdtProvider.newConnectorChannelUDT(type));
96      }
97  
98      @Override
99      public UdtChannelConfig config() {
100         return config;
101     }
102 
103     @Override
104     protected void doBind(final SocketAddress localAddress) throws Exception {
105         privilegedBind(javaChannel(), localAddress);
106     }
107 
108     @Override
109     protected void doClose() throws Exception {
110         javaChannel().close();
111     }
112 
113     @Override
114     protected boolean doConnect(final SocketAddress remoteAddress,
115             final SocketAddress localAddress) throws Exception {
116         doBind(localAddress != null? localAddress : new InetSocketAddress(0));
117         boolean success = false;
118         try {
119             final boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
120             if (!connected) {
121                 selectionKey().interestOps(
122                         selectionKey().interestOps() | OP_CONNECT);
123             }
124             success = true;
125             return connected;
126         } finally {
127             if (!success) {
128                 doClose();
129             }
130         }
131     }
132 
133     @Override
134     protected void doDisconnect() throws Exception {
135         doClose();
136     }
137 
138     @Override
139     protected void doFinishConnect() throws Exception {
140         if (javaChannel().finishConnect()) {
141             selectionKey().interestOps(
142                     selectionKey().interestOps() & ~OP_CONNECT);
143         } else {
144             throw new Error(
145                     "Provider error: failed to finish connect. Provider library should be upgraded.");
146         }
147     }
148 
149     @Override
150     protected int doReadMessages(List<Object> buf) throws Exception {
151 
152         final int maximumMessageSize = config.getReceiveBufferSize();
153 
154         final ByteBuf byteBuf = config.getAllocator().directBuffer(
155                 maximumMessageSize);
156 
157         final int receivedMessageSize = byteBuf.writeBytes(javaChannel(),
158                 maximumMessageSize);
159 
160         if (receivedMessageSize <= 0) {
161             byteBuf.release();
162             return 0;
163         }
164 
165         if (receivedMessageSize >= maximumMessageSize) {
166             javaChannel().close();
167             throw new ChannelException(
168                     "Invalid config : increase receive buffer size to avoid message truncation");
169         }
170 
171         // delivers a message
172         buf.add(new UdtMessage(byteBuf));
173 
174         return 1;
175     }
176 
177     @Override
178     protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
179         // expects a message
180         final UdtMessage message = (UdtMessage) msg;
181 
182         final ByteBuf byteBuf = message.content();
183 
184         final int messageSize = byteBuf.readableBytes();
185         if (messageSize == 0) {
186             return true;
187         }
188 
189         final long writtenBytes;
190         if (byteBuf.nioBufferCount() == 1) {
191             writtenBytes = javaChannel().write(byteBuf.nioBuffer());
192         } else {
193             writtenBytes = javaChannel().write(byteBuf.nioBuffers());
194         }
195 
196         // wrote message completely
197         if (writtenBytes > 0 && writtenBytes != messageSize) {
198             throw new Error(
199                     "Provider error: failed to write message. Provider library should be upgraded.");
200         }
201 
202         return writtenBytes > 0;
203     }
204 
205     @Override
206     public boolean isActive() {
207         final SocketChannelUDT channelUDT = javaChannel();
208         return channelUDT.isOpen() && channelUDT.isConnectFinished();
209     }
210 
211     @Override
212     protected SocketChannelUDT javaChannel() {
213         return (SocketChannelUDT) super.javaChannel();
214     }
215 
216     @Override
217     protected SocketAddress localAddress0() {
218         return javaChannel().socket().getLocalSocketAddress();
219     }
220 
221     @Override
222     public ChannelMetadata metadata() {
223         return METADATA;
224     }
225 
226     @Override
227     protected SocketAddress remoteAddress0() {
228         return javaChannel().socket().getRemoteSocketAddress();
229     }
230 
231     @Override
232     public InetSocketAddress localAddress() {
233         return (InetSocketAddress) super.localAddress();
234     }
235 
236     @Override
237     public InetSocketAddress remoteAddress() {
238         return (InetSocketAddress) super.remoteAddress();
239     }
240 
241     private static void privilegedBind(final SocketChannelUDT socketChannel, final SocketAddress localAddress)
242             throws IOException {
243         try {
244             AccessController.doPrivileged(new PrivilegedExceptionAction<Void>() {
245                 @Override
246                 public Void run() throws IOException {
247                     socketChannel.bind(localAddress);
248                     return null;
249                 }
250             });
251         } catch (PrivilegedActionException e) {
252             throw (IOException) e.getCause();
253         }
254     }
255 }