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.channel.nio.AbstractNioMessageChannel;
26  import io.netty.channel.udt.DefaultUdtChannelConfig;
27  import io.netty.channel.udt.UdtChannel;
28  import io.netty.channel.udt.UdtChannelConfig;
29  import io.netty.channel.udt.UdtMessage;
30  import io.netty.util.internal.logging.InternalLogger;
31  import io.netty.util.internal.logging.InternalLoggerFactory;
32  
33  import java.net.InetSocketAddress;
34  import java.net.SocketAddress;
35  import java.util.List;
36  
37  import static java.nio.channels.SelectionKey.*;
38  
39  /**
40   * Message Connector for UDT Datagrams.
41   * <p>
42   * Note: send/receive must use {@link UdtMessage} in the pipeline
43   */
44  public class NioUdtMessageConnectorChannel extends AbstractNioMessageChannel implements UdtChannel {
45  
46      private static final InternalLogger logger =
47              InternalLoggerFactory.getInstance(NioUdtMessageConnectorChannel.class);
48  
49      private static final ChannelMetadata METADATA = new ChannelMetadata(false);
50  
51      private final UdtChannelConfig config;
52  
53      public NioUdtMessageConnectorChannel() {
54          this(TypeUDT.DATAGRAM);
55      }
56  
57      public NioUdtMessageConnectorChannel(final Channel parent, final SocketChannelUDT channelUDT) {
58          super(parent, channelUDT, OP_READ);
59          try {
60              channelUDT.configureBlocking(false);
61              switch (channelUDT.socketUDT().status()) {
62              case INIT:
63              case OPENED:
64                  config = new DefaultUdtChannelConfig(this, channelUDT, true);
65                  break;
66              default:
67                  config = new DefaultUdtChannelConfig(this, channelUDT, false);
68                  break;
69              }
70          } catch (final Exception e) {
71              try {
72                  channelUDT.close();
73              } catch (final Exception e2) {
74                  if (logger.isWarnEnabled()) {
75                      logger.warn("Failed to close channel.", e2);
76                  }
77              }
78              throw new ChannelException("Failed to configure channel.", e);
79          }
80      }
81  
82      public NioUdtMessageConnectorChannel(final SocketChannelUDT channelUDT) {
83          this(null, channelUDT);
84      }
85  
86      public NioUdtMessageConnectorChannel(final TypeUDT type) {
87          this(NioUdtProvider.newConnectorChannelUDT(type));
88      }
89  
90      @Override
91      public UdtChannelConfig config() {
92          return config;
93      }
94  
95      @Override
96      protected void doBind(final SocketAddress localAddress) throws Exception {
97          javaChannel().bind(localAddress);
98      }
99  
100     @Override
101     protected void doClose() throws Exception {
102         javaChannel().close();
103     }
104 
105     @Override
106     protected boolean doConnect(final SocketAddress remoteAddress,
107             final SocketAddress localAddress) throws Exception {
108         doBind(localAddress != null? localAddress : new InetSocketAddress(0));
109         boolean success = false;
110         try {
111             final boolean connected = javaChannel().connect(remoteAddress);
112             if (!connected) {
113                 selectionKey().interestOps(
114                         selectionKey().interestOps() | OP_CONNECT);
115             }
116             success = true;
117             return connected;
118         } finally {
119             if (!success) {
120                 doClose();
121             }
122         }
123     }
124 
125     @Override
126     protected void doDisconnect() throws Exception {
127         doClose();
128     }
129 
130     @Override
131     protected void doFinishConnect() throws Exception {
132         if (javaChannel().finishConnect()) {
133             selectionKey().interestOps(
134                     selectionKey().interestOps() & ~OP_CONNECT);
135         } else {
136             throw new Error(
137                     "Provider error: failed to finish connect. Provider library should be upgraded.");
138         }
139     }
140 
141     @Override
142     protected int doReadMessages(List<Object> buf) throws Exception {
143 
144         final int maximumMessageSize = config.getReceiveBufferSize();
145 
146         final ByteBuf byteBuf = config.getAllocator().directBuffer(
147                 maximumMessageSize);
148 
149         final int receivedMessageSize = byteBuf.writeBytes(javaChannel(),
150                 maximumMessageSize);
151 
152         if (receivedMessageSize <= 0) {
153             byteBuf.release();
154             return 0;
155         }
156 
157         if (receivedMessageSize >= maximumMessageSize) {
158             javaChannel().close();
159             throw new ChannelException(
160                     "Invalid config : increase receive buffer size to avoid message truncation");
161         }
162 
163         // delivers a message
164         buf.add(new UdtMessage(byteBuf));
165 
166         return 1;
167     }
168 
169     @Override
170     protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
171         // expects a message
172         final UdtMessage message = (UdtMessage) msg;
173 
174         final ByteBuf byteBuf = message.content();
175 
176         final int messageSize = byteBuf.readableBytes();
177 
178         final long writtenBytes;
179         if (byteBuf.nioBufferCount() == 1) {
180             writtenBytes = javaChannel().write(byteBuf.nioBuffer());
181         } else {
182             writtenBytes = javaChannel().write(byteBuf.nioBuffers());
183         }
184 
185         // did not write the message
186         if (writtenBytes <= 0 && messageSize > 0) {
187             return false;
188         }
189 
190         // wrote message completely
191         if (writtenBytes != messageSize) {
192             throw new Error(
193                     "Provider error: failed to write message. Provider library should be upgraded.");
194         }
195 
196         return true;
197     }
198 
199     @Override
200     public boolean isActive() {
201         final SocketChannelUDT channelUDT = javaChannel();
202         return channelUDT.isOpen() && channelUDT.isConnectFinished();
203     }
204 
205     @Override
206     protected SocketChannelUDT javaChannel() {
207         return (SocketChannelUDT) super.javaChannel();
208     }
209 
210     @Override
211     protected SocketAddress localAddress0() {
212         return javaChannel().socket().getLocalSocketAddress();
213     }
214 
215     @Override
216     public ChannelMetadata metadata() {
217         return METADATA;
218     }
219 
220     @Override
221     protected SocketAddress remoteAddress0() {
222         return javaChannel().socket().getRemoteSocketAddress();
223     }
224 
225     @Override
226     public InetSocketAddress localAddress() {
227         return (InetSocketAddress) super.localAddress();
228     }
229 
230     @Override
231     public InetSocketAddress remoteAddress() {
232         return (InetSocketAddress) super.remoteAddress();
233     }
234 }