1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.channel.socket.nio;
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.DefaultAddressedEnvelope;
28 import io.netty.channel.RecvByteBufAllocator;
29 import io.netty.channel.nio.AbstractNioMessageChannel;
30 import io.netty.channel.socket.DatagramChannelConfig;
31 import io.netty.channel.socket.DatagramPacket;
32 import io.netty.channel.socket.InternetProtocolFamily;
33 import io.netty.util.UncheckedBooleanSupplier;
34 import io.netty.util.internal.ObjectUtil;
35 import io.netty.util.internal.SocketUtils;
36 import io.netty.util.internal.PlatformDependent;
37 import io.netty.util.internal.StringUtil;
38 import io.netty.util.internal.SuppressJava6Requirement;
39
40 import java.io.IOException;
41 import java.net.InetAddress;
42 import java.net.InetSocketAddress;
43 import java.net.NetworkInterface;
44 import java.net.SocketAddress;
45 import java.net.SocketException;
46 import java.nio.ByteBuffer;
47 import java.nio.channels.DatagramChannel;
48 import java.nio.channels.MembershipKey;
49 import java.nio.channels.SelectionKey;
50 import java.nio.channels.UnresolvedAddressException;
51 import java.nio.channels.spi.SelectorProvider;
52 import java.util.ArrayList;
53 import java.util.HashMap;
54 import java.util.Iterator;
55 import java.util.List;
56 import java.util.Map;
57
58
59
60
61
62
63
64
65 public final class NioDatagramChannel
66 extends AbstractNioMessageChannel implements io.netty.channel.socket.DatagramChannel {
67
68 private static final ChannelMetadata METADATA = new ChannelMetadata(true, 16);
69 private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
70 private static final String EXPECTED_TYPES =
71 " (expected: " + StringUtil.simpleClassName(DatagramPacket.class) + ", " +
72 StringUtil.simpleClassName(AddressedEnvelope.class) + '<' +
73 StringUtil.simpleClassName(ByteBuf.class) + ", " +
74 StringUtil.simpleClassName(SocketAddress.class) + ">, " +
75 StringUtil.simpleClassName(ByteBuf.class) + ')';
76
77 private final DatagramChannelConfig config;
78
79 private Map<InetAddress, List<MembershipKey>> memberships;
80
81 private static DatagramChannel newSocket(SelectorProvider provider) {
82 try {
83
84
85
86
87
88
89 return provider.openDatagramChannel();
90 } catch (IOException e) {
91 throw new ChannelException("Failed to open a socket.", e);
92 }
93 }
94
95 @SuppressJava6Requirement(reason = "Usage guarded by java version check")
96 private static DatagramChannel newSocket(SelectorProvider provider, InternetProtocolFamily ipFamily) {
97 if (ipFamily == null) {
98 return newSocket(provider);
99 }
100
101 checkJavaVersion();
102
103 try {
104 return provider.openDatagramChannel(ProtocolFamilyConverter.convert(ipFamily));
105 } catch (IOException e) {
106 throw new ChannelException("Failed to open a socket.", e);
107 }
108 }
109
110 private static void checkJavaVersion() {
111 if (PlatformDependent.javaVersion() < 7) {
112 throw new UnsupportedOperationException("Only supported on java 7+.");
113 }
114 }
115
116
117
118
119 public NioDatagramChannel() {
120 this(newSocket(DEFAULT_SELECTOR_PROVIDER));
121 }
122
123
124
125
126
127 public NioDatagramChannel(SelectorProvider provider) {
128 this(newSocket(provider));
129 }
130
131
132
133
134
135 public NioDatagramChannel(InternetProtocolFamily ipFamily) {
136 this(newSocket(DEFAULT_SELECTOR_PROVIDER, ipFamily));
137 }
138
139
140
141
142
143
144 public NioDatagramChannel(SelectorProvider provider, InternetProtocolFamily ipFamily) {
145 this(newSocket(provider, ipFamily));
146 }
147
148
149
150
151 public NioDatagramChannel(DatagramChannel socket) {
152 super(null, socket, SelectionKey.OP_READ);
153 config = new NioDatagramChannelConfig(this, socket);
154 }
155
156 @Override
157 public ChannelMetadata metadata() {
158 return METADATA;
159 }
160
161 @Override
162 public DatagramChannelConfig config() {
163 return config;
164 }
165
166 @Override
167 @SuppressWarnings("deprecation")
168 public boolean isActive() {
169 DatagramChannel ch = javaChannel();
170 return ch.isOpen() && (
171 config.getOption(ChannelOption.DATAGRAM_CHANNEL_ACTIVE_ON_REGISTRATION) && isRegistered()
172 || ch.socket().isBound());
173 }
174
175 @Override
176 public boolean isConnected() {
177 return javaChannel().isConnected();
178 }
179
180 @Override
181 protected DatagramChannel javaChannel() {
182 return (DatagramChannel) super.javaChannel();
183 }
184
185 @Override
186 protected SocketAddress localAddress0() {
187 return javaChannel().socket().getLocalSocketAddress();
188 }
189
190 @Override
191 protected SocketAddress remoteAddress0() {
192 return javaChannel().socket().getRemoteSocketAddress();
193 }
194
195 @Override
196 protected void doBind(SocketAddress localAddress) throws Exception {
197 doBind0(localAddress);
198 }
199
200 private void doBind0(SocketAddress localAddress) throws Exception {
201 if (PlatformDependent.javaVersion() >= 7) {
202 SocketUtils.bind(javaChannel(), localAddress);
203 } else {
204 javaChannel().socket().bind(localAddress);
205 }
206 }
207
208 @Override
209 protected boolean doConnect(SocketAddress remoteAddress,
210 SocketAddress localAddress) throws Exception {
211 if (localAddress != null) {
212 doBind0(localAddress);
213 }
214
215 boolean success = false;
216 try {
217 javaChannel().connect(remoteAddress);
218 success = true;
219 return true;
220 } finally {
221 if (!success) {
222 doClose();
223 }
224 }
225 }
226
227 @Override
228 protected void doFinishConnect() throws Exception {
229 throw new Error();
230 }
231
232 @Override
233 protected void doDisconnect() throws Exception {
234 javaChannel().disconnect();
235 }
236
237 @Override
238 protected void doClose() throws Exception {
239 javaChannel().close();
240 }
241
242 @Override
243 protected int doReadMessages(List<Object> buf) throws Exception {
244 DatagramChannel ch = javaChannel();
245 DatagramChannelConfig config = config();
246 RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
247
248 ByteBuf data = allocHandle.allocate(config.getAllocator());
249 allocHandle.attemptedBytesRead(data.writableBytes());
250 boolean free = true;
251 try {
252 ByteBuffer nioData = data.internalNioBuffer(data.writerIndex(), data.writableBytes());
253 int pos = nioData.position();
254 InetSocketAddress remoteAddress = (InetSocketAddress) ch.receive(nioData);
255 if (remoteAddress == null) {
256 return 0;
257 }
258
259 allocHandle.lastBytesRead(nioData.position() - pos);
260 buf.add(new DatagramPacket(data.writerIndex(data.writerIndex() + allocHandle.lastBytesRead()),
261 localAddress(), remoteAddress));
262 free = false;
263 return 1;
264 } catch (Throwable cause) {
265 PlatformDependent.throwException(cause);
266 return -1;
267 } finally {
268 if (free) {
269 data.release();
270 }
271 }
272 }
273
274 @Override
275 protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
276 final SocketAddress remoteAddress;
277 final ByteBuf data;
278 if (msg instanceof AddressedEnvelope) {
279 @SuppressWarnings("unchecked")
280 AddressedEnvelope<ByteBuf, SocketAddress> envelope = (AddressedEnvelope<ByteBuf, SocketAddress>) msg;
281 remoteAddress = envelope.recipient();
282 data = envelope.content();
283 } else {
284 data = (ByteBuf) msg;
285 remoteAddress = null;
286 }
287
288 final int dataLen = data.readableBytes();
289 final ByteBuffer nioData = data.nioBufferCount() == 1 ? data.internalNioBuffer(data.readerIndex(), dataLen)
290 : data.nioBuffer(data.readerIndex(), dataLen);
291 final int writtenBytes;
292 if (remoteAddress != null) {
293 writtenBytes = javaChannel().send(nioData, remoteAddress);
294 } else {
295 writtenBytes = javaChannel().write(nioData);
296 }
297
298 return writtenBytes > 0 || dataLen == 0;
299 }
300
301 private static void checkUnresolved(AddressedEnvelope<?, ?> envelope) {
302 if (envelope.recipient() instanceof InetSocketAddress
303 && (((InetSocketAddress) envelope.recipient()).isUnresolved())) {
304 throw new UnresolvedAddressException();
305 }
306 }
307
308 @Override
309 protected Object filterOutboundMessage(Object msg) {
310 if (msg instanceof DatagramPacket) {
311 DatagramPacket p = (DatagramPacket) msg;
312 checkUnresolved(p);
313 ByteBuf content = p.content();
314 if (isSingleDirectBuffer(content)) {
315 return p;
316 }
317 return new DatagramPacket(newDirectBuffer(p, content), p.recipient());
318 }
319
320 if (msg instanceof ByteBuf) {
321 ByteBuf buf = (ByteBuf) msg;
322 if (isSingleDirectBuffer(buf)) {
323 return buf;
324 }
325 return newDirectBuffer(buf);
326 }
327
328 if (msg instanceof AddressedEnvelope) {
329 @SuppressWarnings("unchecked")
330 AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
331 checkUnresolved(e);
332 if (e.content() instanceof ByteBuf) {
333 ByteBuf content = (ByteBuf) e.content();
334 if (isSingleDirectBuffer(content)) {
335 return e;
336 }
337 return new DefaultAddressedEnvelope<ByteBuf, SocketAddress>(newDirectBuffer(e, content), e.recipient());
338 }
339 }
340
341 throw new UnsupportedOperationException(
342 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
343 }
344
345
346
347
348
349 private static boolean isSingleDirectBuffer(ByteBuf buf) {
350 return buf.isDirect() && buf.nioBufferCount() == 1;
351 }
352
353 @Override
354 protected boolean continueOnWriteError() {
355
356
357
358 return true;
359 }
360
361 @Override
362 public InetSocketAddress localAddress() {
363 return (InetSocketAddress) super.localAddress();
364 }
365
366 @Override
367 public InetSocketAddress remoteAddress() {
368 return (InetSocketAddress) super.remoteAddress();
369 }
370
371 @Override
372 public ChannelFuture joinGroup(InetAddress multicastAddress) {
373 return joinGroup(multicastAddress, newPromise());
374 }
375
376 @Override
377 public ChannelFuture joinGroup(InetAddress multicastAddress, ChannelPromise promise) {
378 try {
379 return joinGroup(multicastAddress, defaultMulticastInterface(), null, promise);
380 } catch (SocketException e) {
381 promise.setFailure(e);
382 }
383 return promise;
384 }
385
386
387
388
389
390 private NetworkInterface defaultMulticastInterface() throws SocketException {
391 NetworkInterface iface = config.getNetworkInterface();
392 if (iface == null) {
393 iface = NetworkInterface.getByInetAddress(localAddress().getAddress());
394 }
395 return iface;
396 }
397
398 @Override
399 public ChannelFuture joinGroup(
400 InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
401 return joinGroup(multicastAddress, networkInterface, newPromise());
402 }
403
404 @Override
405 public ChannelFuture joinGroup(
406 InetSocketAddress multicastAddress, NetworkInterface networkInterface,
407 ChannelPromise promise) {
408 return joinGroup(multicastAddress.getAddress(), networkInterface, null, promise);
409 }
410
411 @Override
412 public ChannelFuture joinGroup(
413 InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
414 return joinGroup(multicastAddress, networkInterface, source, newPromise());
415 }
416
417 @SuppressJava6Requirement(reason = "Usage guarded by java version check")
418 @Override
419 public ChannelFuture joinGroup(
420 InetAddress multicastAddress, NetworkInterface networkInterface,
421 InetAddress source, ChannelPromise promise) {
422
423 checkJavaVersion();
424
425 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
426 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
427
428 try {
429 MembershipKey key;
430 if (source == null) {
431 key = javaChannel().join(multicastAddress, networkInterface);
432 } else {
433 key = javaChannel().join(multicastAddress, networkInterface, source);
434 }
435
436 synchronized (this) {
437 List<MembershipKey> keys = null;
438 if (memberships == null) {
439 memberships = new HashMap<InetAddress, List<MembershipKey>>();
440 } else {
441 keys = memberships.get(multicastAddress);
442 }
443 if (keys == null) {
444 keys = new ArrayList<MembershipKey>();
445 memberships.put(multicastAddress, keys);
446 }
447 keys.add(key);
448 }
449
450 promise.setSuccess();
451 } catch (Throwable e) {
452 promise.setFailure(e);
453 }
454
455 return promise;
456 }
457
458 @Override
459 public ChannelFuture leaveGroup(InetAddress multicastAddress) {
460 return leaveGroup(multicastAddress, newPromise());
461 }
462
463 @Override
464 public ChannelFuture leaveGroup(InetAddress multicastAddress, ChannelPromise promise) {
465 try {
466 return leaveGroup(multicastAddress, defaultMulticastInterface(), null, promise);
467 } catch (SocketException e) {
468 promise.setFailure(e);
469 }
470 return promise;
471 }
472
473 @Override
474 public ChannelFuture leaveGroup(
475 InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
476 return leaveGroup(multicastAddress, networkInterface, newPromise());
477 }
478
479 @Override
480 public ChannelFuture leaveGroup(
481 InetSocketAddress multicastAddress,
482 NetworkInterface networkInterface, ChannelPromise promise) {
483 return leaveGroup(multicastAddress.getAddress(), networkInterface, null, promise);
484 }
485
486 @Override
487 public ChannelFuture leaveGroup(
488 InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
489 return leaveGroup(multicastAddress, networkInterface, source, newPromise());
490 }
491
492 @SuppressJava6Requirement(reason = "Usage guarded by java version check")
493 @Override
494 public ChannelFuture leaveGroup(
495 InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source,
496 ChannelPromise promise) {
497 checkJavaVersion();
498
499 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
500 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
501
502 synchronized (this) {
503 if (memberships != null) {
504 List<MembershipKey> keys = memberships.get(multicastAddress);
505 if (keys != null) {
506 Iterator<MembershipKey> keyIt = keys.iterator();
507
508 while (keyIt.hasNext()) {
509 MembershipKey key = keyIt.next();
510 if (networkInterface.equals(key.networkInterface())) {
511 if (source == null && key.sourceAddress() == null ||
512 source != null && source.equals(key.sourceAddress())) {
513 key.drop();
514 keyIt.remove();
515 }
516 }
517 }
518 if (keys.isEmpty()) {
519 memberships.remove(multicastAddress);
520 }
521 }
522 }
523 }
524
525 promise.setSuccess();
526 return promise;
527 }
528
529
530
531
532 @Override
533 public ChannelFuture block(
534 InetAddress multicastAddress, NetworkInterface networkInterface,
535 InetAddress sourceToBlock) {
536 return block(multicastAddress, networkInterface, sourceToBlock, newPromise());
537 }
538
539
540
541
542 @SuppressJava6Requirement(reason = "Usage guarded by java version check")
543 @Override
544 public ChannelFuture block(
545 InetAddress multicastAddress, NetworkInterface networkInterface,
546 InetAddress sourceToBlock, ChannelPromise promise) {
547 checkJavaVersion();
548
549 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
550 ObjectUtil.checkNotNull(sourceToBlock, "sourceToBlock");
551 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
552
553 synchronized (this) {
554 if (memberships != null) {
555 List<MembershipKey> keys = memberships.get(multicastAddress);
556 for (MembershipKey key: keys) {
557 if (networkInterface.equals(key.networkInterface())) {
558 try {
559 key.block(sourceToBlock);
560 } catch (IOException e) {
561 return promise.setFailure(e);
562 }
563 }
564 }
565 }
566 }
567 promise.setSuccess();
568 return promise;
569 }
570
571
572
573
574
575 @Override
576 public ChannelFuture block(InetAddress multicastAddress, InetAddress sourceToBlock) {
577 return block(multicastAddress, sourceToBlock, newPromise());
578 }
579
580
581
582
583
584 @Override
585 public ChannelFuture block(
586 InetAddress multicastAddress, InetAddress sourceToBlock, ChannelPromise promise) {
587 try {
588 return block(multicastAddress, defaultMulticastInterface(), sourceToBlock, promise);
589 } catch (SocketException e) {
590 promise.setFailure(e);
591 }
592 return promise;
593 }
594
595 @Override
596 @Deprecated
597 protected void setReadPending(boolean readPending) {
598 super.setReadPending(readPending);
599 }
600
601 void clearReadPending0() {
602 clearReadPending();
603 }
604
605 @Override
606 protected boolean closeOnReadError(Throwable cause) {
607
608
609 if (cause instanceof SocketException) {
610 return false;
611 }
612 return super.closeOnReadError(cause);
613 }
614
615 @Override
616 protected boolean continueReading(RecvByteBufAllocator.Handle allocHandle) {
617 if (allocHandle instanceof RecvByteBufAllocator.ExtendedHandle) {
618
619
620 return ((RecvByteBufAllocator.ExtendedHandle) allocHandle)
621 .continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER);
622 }
623 return allocHandle.continueReading();
624 }
625 }