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.socket.oio;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.channel.AddressedEnvelope;
20  import io.netty.channel.Channel;
21  import io.netty.channel.ChannelException;
22  import io.netty.channel.ChannelFuture;
23  import io.netty.channel.ChannelMetadata;
24  import io.netty.channel.ChannelOption;
25  import io.netty.channel.ChannelOutboundBuffer;
26  import io.netty.channel.ChannelPromise;
27  import io.netty.channel.RecvByteBufAllocator;
28  import io.netty.channel.oio.AbstractOioMessageChannel;
29  import io.netty.channel.socket.DatagramChannel;
30  import io.netty.channel.socket.DatagramChannelConfig;
31  import io.netty.channel.socket.DatagramPacket;
32  import io.netty.util.internal.EmptyArrays;
33  import io.netty.util.internal.PlatformDependent;
34  import io.netty.util.internal.StringUtil;
35  import io.netty.util.internal.logging.InternalLogger;
36  import io.netty.util.internal.logging.InternalLoggerFactory;
37  
38  import java.io.IOException;
39  import java.net.InetAddress;
40  import java.net.InetSocketAddress;
41  import java.net.MulticastSocket;
42  import java.net.NetworkInterface;
43  import java.net.SocketAddress;
44  import java.net.SocketException;
45  import java.net.SocketTimeoutException;
46  import java.nio.channels.NotYetConnectedException;
47  import java.util.List;
48  import java.util.Locale;
49  
50  /**
51   * An OIO datagram {@link Channel} that sends and receives an
52   * {@link AddressedEnvelope AddressedEnvelope<ByteBuf, SocketAddress>}.
53   *
54   * @see AddressedEnvelope
55   * @see DatagramPacket
56   */
57  public class OioDatagramChannel extends AbstractOioMessageChannel
58                                  implements DatagramChannel {
59  
60      private static final InternalLogger logger = InternalLoggerFactory.getInstance(OioDatagramChannel.class);
61  
62      private static final ChannelMetadata METADATA = new ChannelMetadata(true);
63      private static final String EXPECTED_TYPES =
64              " (expected: " + StringUtil.simpleClassName(DatagramPacket.class) + ", " +
65              StringUtil.simpleClassName(AddressedEnvelope.class) + '<' +
66              StringUtil.simpleClassName(ByteBuf.class) + ", " +
67              StringUtil.simpleClassName(SocketAddress.class) + ">, " +
68              StringUtil.simpleClassName(ByteBuf.class) + ')';
69  
70      private final MulticastSocket socket;
71      private final OioDatagramChannelConfig config;
72      private final java.net.DatagramPacket tmpPacket = new java.net.DatagramPacket(EmptyArrays.EMPTY_BYTES, 0);
73  
74      private static MulticastSocket newSocket() {
75          try {
76              return new MulticastSocket(null);
77          } catch (Exception e) {
78              throw new ChannelException("failed to create a new socket", e);
79          }
80      }
81  
82      /**
83       * Create a new instance with an new {@link MulticastSocket}.
84       */
85      public OioDatagramChannel() {
86          this(newSocket());
87      }
88  
89      /**
90       * Create a new instance from the given {@link MulticastSocket}.
91       *
92       * @param socket    the {@link MulticastSocket} which is used by this instance
93       */
94      public OioDatagramChannel(MulticastSocket socket) {
95          super(null);
96  
97          boolean success = false;
98          try {
99              socket.setSoTimeout(SO_TIMEOUT);
100             socket.setBroadcast(false);
101             success = true;
102         } catch (SocketException e) {
103             throw new ChannelException(
104                     "Failed to configure the datagram socket timeout.", e);
105         } finally {
106             if (!success) {
107                 socket.close();
108             }
109         }
110 
111         this.socket = socket;
112         config = new DefaultOioDatagramChannelConfig(this, socket);
113     }
114 
115     @Override
116     public ChannelMetadata metadata() {
117         return METADATA;
118     }
119 
120     /**
121      * {@inheritDoc}
122      *
123      * This can be safetly cast to {@link OioDatagramChannelConfig}.
124      */
125     @Override
126     // TODO: Change return type to OioDatagramChannelConfig in next major release
127     public DatagramChannelConfig config() {
128         return config;
129     }
130 
131     @Override
132     public boolean isOpen() {
133         return !socket.isClosed();
134     }
135 
136     @Override
137     @SuppressWarnings("deprecation")
138     public boolean isActive() {
139         return isOpen()
140             && (config.getOption(ChannelOption.DATAGRAM_CHANNEL_ACTIVE_ON_REGISTRATION) && isRegistered()
141                  || socket.isBound());
142     }
143 
144     @Override
145     public boolean isConnected() {
146         return socket.isConnected();
147     }
148 
149     @Override
150     protected SocketAddress localAddress0() {
151         return socket.getLocalSocketAddress();
152     }
153 
154     @Override
155     protected SocketAddress remoteAddress0() {
156         return socket.getRemoteSocketAddress();
157     }
158 
159     @Override
160     protected void doBind(SocketAddress localAddress) throws Exception {
161         socket.bind(localAddress);
162     }
163 
164     @Override
165     public InetSocketAddress localAddress() {
166         return (InetSocketAddress) super.localAddress();
167     }
168 
169     @Override
170     public InetSocketAddress remoteAddress() {
171         return (InetSocketAddress) super.remoteAddress();
172     }
173 
174     @Override
175     protected void doConnect(SocketAddress remoteAddress,
176             SocketAddress localAddress) throws Exception {
177         if (localAddress != null) {
178             socket.bind(localAddress);
179         }
180 
181         boolean success = false;
182         try {
183             socket.connect(remoteAddress);
184             success = true;
185         } finally {
186             if (!success) {
187                 try {
188                     socket.close();
189                 } catch (Throwable t) {
190                     logger.warn("Failed to close a socket.", t);
191                 }
192             }
193         }
194     }
195 
196     @Override
197     protected void doDisconnect() throws Exception {
198         socket.disconnect();
199     }
200 
201     @Override
202     protected void doClose() throws Exception {
203         socket.close();
204     }
205 
206     @Override
207     protected int doReadMessages(List<Object> buf) throws Exception {
208         DatagramChannelConfig config = config();
209         final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
210 
211         ByteBuf data = config.getAllocator().heapBuffer(allocHandle.guess());
212         boolean free = true;
213         try {
214             // Ensure we null out the address which may have been set before.
215             tmpPacket.setAddress(null);
216             tmpPacket.setData(data.array(), data.arrayOffset(), data.capacity());
217             socket.receive(tmpPacket);
218 
219             InetSocketAddress remoteAddr = (InetSocketAddress) tmpPacket.getSocketAddress();
220 
221             allocHandle.lastBytesRead(tmpPacket.getLength());
222             buf.add(new DatagramPacket(data.writerIndex(allocHandle.lastBytesRead()), localAddress(), remoteAddr));
223             free = false;
224             return 1;
225         } catch (SocketTimeoutException e) {
226             // Expected
227             return 0;
228         } catch (SocketException e) {
229             if (!e.getMessage().toLowerCase(Locale.US).contains("socket closed")) {
230                 throw e;
231             }
232             return -1;
233         } catch (Throwable cause) {
234             PlatformDependent.throwException(cause);
235             return -1;
236         } finally {
237             if (free) {
238                 data.release();
239             }
240         }
241     }
242 
243     @Override
244     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
245         for (;;) {
246             final Object o = in.current();
247             if (o == null) {
248                 break;
249             }
250 
251             final ByteBuf data;
252             final SocketAddress remoteAddress;
253             if (o instanceof AddressedEnvelope) {
254                 @SuppressWarnings("unchecked")
255                 AddressedEnvelope<ByteBuf, SocketAddress> envelope = (AddressedEnvelope<ByteBuf, SocketAddress>) o;
256                 remoteAddress = envelope.recipient();
257                 data = envelope.content();
258             } else {
259                 data = (ByteBuf) o;
260                 remoteAddress = null;
261             }
262 
263             final int length = data.readableBytes();
264             try {
265                 if (remoteAddress != null) {
266                     tmpPacket.setSocketAddress(remoteAddress);
267                 } else {
268                     if (!isConnected()) {
269                         // If not connected we should throw a NotYetConnectedException() to be consistent with
270                         // NioDatagramChannel
271                         throw new NotYetConnectedException();
272                     }
273                     // Ensure we null out the address which may have been set before.
274                     tmpPacket.setAddress(null);
275                 }
276                 if (data.hasArray()) {
277                     tmpPacket.setData(data.array(), data.arrayOffset() + data.readerIndex(), length);
278                 } else {
279                     byte[] tmp = new byte[length];
280                     data.getBytes(data.readerIndex(), tmp);
281                     tmpPacket.setData(tmp);
282                 }
283                 socket.send(tmpPacket);
284                 in.remove();
285             } catch (Exception e) {
286                 // Continue on write error as a DatagramChannel can write to multiple remote peers
287                 //
288                 // See https://github.com/netty/netty/issues/2665
289                 in.remove(e);
290             }
291         }
292     }
293 
294     @Override
295     protected Object filterOutboundMessage(Object msg) {
296         if (msg instanceof DatagramPacket || msg instanceof ByteBuf) {
297             return msg;
298         }
299 
300         if (msg instanceof AddressedEnvelope) {
301             @SuppressWarnings("unchecked")
302             AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
303             if (e.content() instanceof ByteBuf) {
304                 return msg;
305             }
306         }
307 
308         throw new UnsupportedOperationException(
309                 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
310     }
311 
312     @Override
313     public ChannelFuture joinGroup(InetAddress multicastAddress) {
314         return joinGroup(multicastAddress, newPromise());
315     }
316 
317     @Override
318     public ChannelFuture joinGroup(InetAddress multicastAddress, ChannelPromise promise) {
319         ensureBound();
320         try {
321             socket.joinGroup(multicastAddress);
322             promise.setSuccess();
323         } catch (IOException e) {
324             promise.setFailure(e);
325         }
326         return promise;
327     }
328 
329     @Override
330     public ChannelFuture joinGroup(InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
331         return joinGroup(multicastAddress, networkInterface, newPromise());
332     }
333 
334     @Override
335     public ChannelFuture joinGroup(
336             InetSocketAddress multicastAddress, NetworkInterface networkInterface,
337             ChannelPromise promise) {
338         ensureBound();
339         try {
340             socket.joinGroup(multicastAddress, networkInterface);
341             promise.setSuccess();
342         } catch (IOException e) {
343             promise.setFailure(e);
344         }
345         return promise;
346     }
347 
348     @Override
349     public ChannelFuture joinGroup(
350             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
351         return newFailedFuture(new UnsupportedOperationException());
352     }
353 
354     @Override
355     public ChannelFuture joinGroup(
356             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source,
357             ChannelPromise promise) {
358         promise.setFailure(new UnsupportedOperationException());
359         return promise;
360     }
361 
362     private void ensureBound() {
363         if (!isActive()) {
364             throw new IllegalStateException(
365                     DatagramChannel.class.getName() +
366                     " must be bound to join a group.");
367         }
368     }
369 
370     @Override
371     public ChannelFuture leaveGroup(InetAddress multicastAddress) {
372         return leaveGroup(multicastAddress, newPromise());
373     }
374 
375     @Override
376     public ChannelFuture leaveGroup(InetAddress multicastAddress, ChannelPromise promise) {
377         try {
378             socket.leaveGroup(multicastAddress);
379             promise.setSuccess();
380         } catch (IOException e) {
381             promise.setFailure(e);
382         }
383         return promise;
384     }
385 
386     @Override
387     public ChannelFuture leaveGroup(
388             InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
389         return leaveGroup(multicastAddress, networkInterface, newPromise());
390     }
391 
392     @Override
393     public ChannelFuture leaveGroup(
394             InetSocketAddress multicastAddress, NetworkInterface networkInterface,
395             ChannelPromise promise) {
396         try {
397             socket.leaveGroup(multicastAddress, networkInterface);
398             promise.setSuccess();
399         } catch (IOException e) {
400             promise.setFailure(e);
401         }
402         return promise;
403     }
404 
405     @Override
406     public ChannelFuture leaveGroup(
407             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
408         return newFailedFuture(new UnsupportedOperationException());
409     }
410 
411     @Override
412     public ChannelFuture leaveGroup(
413             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source,
414             ChannelPromise promise) {
415         promise.setFailure(new UnsupportedOperationException());
416         return promise;
417     }
418 
419     @Override
420     public ChannelFuture block(InetAddress multicastAddress,
421             NetworkInterface networkInterface, InetAddress sourceToBlock) {
422         return newFailedFuture(new UnsupportedOperationException());
423     }
424 
425     @Override
426     public ChannelFuture block(InetAddress multicastAddress,
427             NetworkInterface networkInterface, InetAddress sourceToBlock,
428             ChannelPromise promise) {
429         promise.setFailure(new UnsupportedOperationException());
430         return promise;
431     }
432 
433     @Override
434     public ChannelFuture block(InetAddress multicastAddress,
435             InetAddress sourceToBlock) {
436         return newFailedFuture(new UnsupportedOperationException());
437     }
438 
439     @Override
440     public ChannelFuture block(InetAddress multicastAddress,
441             InetAddress sourceToBlock, ChannelPromise promise) {
442         promise.setFailure(new UnsupportedOperationException());
443         return promise;
444     }
445 }