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.channel.socket.DefaultDatagramChannelConfig;
33  import io.netty.util.internal.EmptyArrays;
34  import io.netty.util.internal.PlatformDependent;
35  import io.netty.util.internal.StringUtil;
36  import io.netty.util.internal.logging.InternalLogger;
37  import io.netty.util.internal.logging.InternalLoggerFactory;
38  
39  import java.io.IOException;
40  import java.net.InetAddress;
41  import java.net.InetSocketAddress;
42  import java.net.MulticastSocket;
43  import java.net.NetworkInterface;
44  import java.net.SocketAddress;
45  import java.net.SocketException;
46  import java.net.SocketTimeoutException;
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 DatagramChannelConfig 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 DefaultDatagramChannelConfig(this, socket);
113     }
114 
115     @Override
116     public ChannelMetadata metadata() {
117         return METADATA;
118     }
119 
120     @Override
121     public DatagramChannelConfig config() {
122         return config;
123     }
124 
125     @Override
126     public boolean isOpen() {
127         return !socket.isClosed();
128     }
129 
130     @Override
131     @SuppressWarnings("deprecation")
132     public boolean isActive() {
133         return isOpen()
134             && (config.getOption(ChannelOption.DATAGRAM_CHANNEL_ACTIVE_ON_REGISTRATION) && isRegistered()
135                  || socket.isBound());
136     }
137 
138     @Override
139     public boolean isConnected() {
140         return socket.isConnected();
141     }
142 
143     @Override
144     protected SocketAddress localAddress0() {
145         return socket.getLocalSocketAddress();
146     }
147 
148     @Override
149     protected SocketAddress remoteAddress0() {
150         return socket.getRemoteSocketAddress();
151     }
152 
153     @Override
154     protected void doBind(SocketAddress localAddress) throws Exception {
155         socket.bind(localAddress);
156     }
157 
158     @Override
159     public InetSocketAddress localAddress() {
160         return (InetSocketAddress) super.localAddress();
161     }
162 
163     @Override
164     public InetSocketAddress remoteAddress() {
165         return (InetSocketAddress) super.remoteAddress();
166     }
167 
168     @Override
169     protected void doConnect(SocketAddress remoteAddress,
170             SocketAddress localAddress) throws Exception {
171         if (localAddress != null) {
172             socket.bind(localAddress);
173         }
174 
175         boolean success = false;
176         try {
177             socket.connect(remoteAddress);
178             success = true;
179         } finally {
180             if (!success) {
181                 try {
182                     socket.close();
183                 } catch (Throwable t) {
184                     logger.warn("Failed to close a socket.", t);
185                 }
186             }
187         }
188     }
189 
190     @Override
191     protected void doDisconnect() throws Exception {
192         socket.disconnect();
193     }
194 
195     @Override
196     protected void doClose() throws Exception {
197         socket.close();
198     }
199 
200     @Override
201     protected int doReadMessages(List<Object> buf) throws Exception {
202         DatagramChannelConfig config = config();
203         RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
204 
205         ByteBuf data = config.getAllocator().heapBuffer(allocHandle.guess());
206         boolean free = true;
207         try {
208             tmpPacket.setData(data.array(), data.arrayOffset(), data.capacity());
209             socket.receive(tmpPacket);
210 
211             InetSocketAddress remoteAddr = (InetSocketAddress) tmpPacket.getSocketAddress();
212 
213             int readBytes = tmpPacket.getLength();
214             allocHandle.record(readBytes);
215             buf.add(new DatagramPacket(data.writerIndex(readBytes), localAddress(), remoteAddr));
216             free = false;
217             return 1;
218         } catch (SocketTimeoutException e) {
219             // Expected
220             return 0;
221         } catch (SocketException e) {
222             if (!e.getMessage().toLowerCase(Locale.US).contains("socket closed")) {
223                 throw e;
224             }
225             return -1;
226         } catch (Throwable cause) {
227             PlatformDependent.throwException(cause);
228             return -1;
229         } finally {
230             if (free) {
231                 data.release();
232             }
233         }
234     }
235 
236     @Override
237     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
238         for (;;) {
239             final Object o = in.current();
240             if (o == null) {
241                 break;
242             }
243 
244             final ByteBuf data;
245             final SocketAddress remoteAddress;
246             if (o instanceof AddressedEnvelope) {
247                 @SuppressWarnings("unchecked")
248                 AddressedEnvelope<ByteBuf, SocketAddress> envelope = (AddressedEnvelope<ByteBuf, SocketAddress>) o;
249                 remoteAddress = envelope.recipient();
250                 data = envelope.content();
251             } else {
252                 data = (ByteBuf) o;
253                 remoteAddress = null;
254             }
255 
256             final int length = data.readableBytes();
257             if (remoteAddress != null) {
258                 tmpPacket.setSocketAddress(remoteAddress);
259             }
260             if (data.hasArray()) {
261                 tmpPacket.setData(data.array(), data.arrayOffset() + data.readerIndex(), length);
262             } else {
263                 byte[] tmp = new byte[length];
264                 data.getBytes(data.readerIndex(), tmp);
265                 tmpPacket.setData(tmp);
266             }
267             try {
268                 socket.send(tmpPacket);
269                 in.remove();
270             } catch (IOException e) {
271                 // Continue on write error as a DatagramChannel can write to multiple remote peers
272                 //
273                 // See https://github.com/netty/netty/issues/2665
274                 in.remove(e);
275             }
276         }
277     }
278 
279     @Override
280     protected Object filterOutboundMessage(Object msg) {
281         if (msg instanceof DatagramPacket || msg instanceof ByteBuf) {
282             return msg;
283         }
284 
285         if (msg instanceof AddressedEnvelope) {
286             @SuppressWarnings("unchecked")
287             AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
288             if (e.content() instanceof ByteBuf) {
289                 return msg;
290             }
291         }
292 
293         throw new UnsupportedOperationException(
294                 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
295     }
296 
297     @Override
298     public ChannelFuture joinGroup(InetAddress multicastAddress) {
299         return joinGroup(multicastAddress, newPromise());
300     }
301 
302     @Override
303     public ChannelFuture joinGroup(InetAddress multicastAddress, ChannelPromise promise) {
304         ensureBound();
305         try {
306             socket.joinGroup(multicastAddress);
307             promise.setSuccess();
308         } catch (IOException e) {
309             promise.setFailure(e);
310         }
311         return promise;
312     }
313 
314     @Override
315     public ChannelFuture joinGroup(InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
316         return joinGroup(multicastAddress, networkInterface, newPromise());
317     }
318 
319     @Override
320     public ChannelFuture joinGroup(
321             InetSocketAddress multicastAddress, NetworkInterface networkInterface,
322             ChannelPromise promise) {
323         ensureBound();
324         try {
325             socket.joinGroup(multicastAddress, networkInterface);
326             promise.setSuccess();
327         } catch (IOException e) {
328             promise.setFailure(e);
329         }
330         return promise;
331     }
332 
333     @Override
334     public ChannelFuture joinGroup(
335             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
336         return newFailedFuture(new UnsupportedOperationException());
337     }
338 
339     @Override
340     public ChannelFuture joinGroup(
341             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source,
342             ChannelPromise promise) {
343         promise.setFailure(new UnsupportedOperationException());
344         return promise;
345     }
346 
347     private void ensureBound() {
348         if (!isActive()) {
349             throw new IllegalStateException(
350                     DatagramChannel.class.getName() +
351                     " must be bound to join a group.");
352         }
353     }
354 
355     @Override
356     public ChannelFuture leaveGroup(InetAddress multicastAddress) {
357         return leaveGroup(multicastAddress, newPromise());
358     }
359 
360     @Override
361     public ChannelFuture leaveGroup(InetAddress multicastAddress, ChannelPromise promise) {
362         try {
363             socket.leaveGroup(multicastAddress);
364             promise.setSuccess();
365         } catch (IOException e) {
366             promise.setFailure(e);
367         }
368         return promise;
369     }
370 
371     @Override
372     public ChannelFuture leaveGroup(
373             InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
374         return leaveGroup(multicastAddress, networkInterface, newPromise());
375     }
376 
377     @Override
378     public ChannelFuture leaveGroup(
379             InetSocketAddress multicastAddress, NetworkInterface networkInterface,
380             ChannelPromise promise) {
381         try {
382             socket.leaveGroup(multicastAddress, networkInterface);
383             promise.setSuccess();
384         } catch (IOException e) {
385             promise.setFailure(e);
386         }
387         return promise;
388     }
389 
390     @Override
391     public ChannelFuture leaveGroup(
392             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
393         return newFailedFuture(new UnsupportedOperationException());
394     }
395 
396     @Override
397     public ChannelFuture leaveGroup(
398             InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source,
399             ChannelPromise promise) {
400         promise.setFailure(new UnsupportedOperationException());
401         return promise;
402     }
403 
404     @Override
405     public ChannelFuture block(InetAddress multicastAddress,
406             NetworkInterface networkInterface, InetAddress sourceToBlock) {
407         return newFailedFuture(new UnsupportedOperationException());
408     }
409 
410     @Override
411     public ChannelFuture block(InetAddress multicastAddress,
412             NetworkInterface networkInterface, InetAddress sourceToBlock,
413             ChannelPromise promise) {
414         promise.setFailure(new UnsupportedOperationException());
415         return promise;
416     }
417 
418     @Override
419     public ChannelFuture block(InetAddress multicastAddress,
420             InetAddress sourceToBlock) {
421         return newFailedFuture(new UnsupportedOperationException());
422     }
423 
424     @Override
425     public ChannelFuture block(InetAddress multicastAddress,
426             InetAddress sourceToBlock, ChannelPromise promise) {
427         promise.setFailure(new UnsupportedOperationException());
428         return promise;
429     }
430 }