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