1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.channel.kqueue;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.ByteBufAllocator;
20 import io.netty.channel.AddressedEnvelope;
21 import io.netty.channel.ChannelFuture;
22 import io.netty.channel.ChannelPipeline;
23 import io.netty.channel.ChannelPromise;
24 import io.netty.channel.DefaultAddressedEnvelope;
25 import io.netty.channel.socket.DatagramChannel;
26 import io.netty.channel.socket.DatagramChannelConfig;
27 import io.netty.channel.socket.DatagramPacket;
28 import io.netty.channel.socket.InternetProtocolFamily;
29 import io.netty.channel.socket.SocketProtocolFamily;
30 import io.netty.channel.unix.DatagramSocketAddress;
31 import io.netty.channel.unix.Errors;
32 import io.netty.channel.unix.IovArray;
33 import io.netty.channel.unix.UnixChannelUtil;
34 import io.netty.util.UncheckedBooleanSupplier;
35 import io.netty.util.internal.ObjectUtil;
36 import io.netty.util.internal.StringUtil;
37
38 import java.io.IOException;
39 import java.net.Inet4Address;
40 import java.net.InetAddress;
41 import java.net.InetSocketAddress;
42 import java.net.NetworkInterface;
43 import java.net.PortUnreachableException;
44 import java.net.SocketAddress;
45 import java.net.SocketException;
46 import java.nio.ByteBuffer;
47 import java.nio.channels.UnresolvedAddressException;
48
49 import static io.netty.channel.kqueue.BsdSocket.newSocketDgram;
50
51 public final class KQueueDatagramChannel extends AbstractKQueueDatagramChannel implements DatagramChannel {
52 private static final String EXPECTED_TYPES =
53 " (expected: " + StringUtil.simpleClassName(DatagramPacket.class) + ", " +
54 StringUtil.simpleClassName(AddressedEnvelope.class) + '<' +
55 StringUtil.simpleClassName(ByteBuf.class) + ", " +
56 StringUtil.simpleClassName(InetSocketAddress.class) + ">, " +
57 StringUtil.simpleClassName(ByteBuf.class) + ')';
58
59 private volatile boolean connected;
60 private final KQueueDatagramChannelConfig config;
61
62 public KQueueDatagramChannel() {
63 super(null, newSocketDgram(), false);
64 config = new KQueueDatagramChannelConfig(this);
65 }
66
67
68
69
70 @Deprecated
71 public KQueueDatagramChannel(InternetProtocolFamily protocol) {
72 super(null, newSocketDgram(protocol), false);
73 config = new KQueueDatagramChannelConfig(this);
74 }
75
76 public KQueueDatagramChannel(SocketProtocolFamily protocol) {
77 super(null, newSocketDgram(protocol), false);
78 config = new KQueueDatagramChannelConfig(this);
79 }
80
81 public KQueueDatagramChannel(int fd) {
82 this(new BsdSocket(fd), true);
83 }
84
85 KQueueDatagramChannel(BsdSocket socket, boolean active) {
86 super(null, socket, active);
87 config = new KQueueDatagramChannelConfig(this);
88 }
89
90 @Override
91 public InetSocketAddress remoteAddress() {
92 return (InetSocketAddress) super.remoteAddress();
93 }
94
95 @Override
96 public InetSocketAddress localAddress() {
97 return (InetSocketAddress) super.localAddress();
98 }
99
100 @Override
101 @SuppressWarnings("deprecation")
102 public boolean isActive() {
103 return socket.isOpen() && (config.getActiveOnOpen() && isRegistered() || active);
104 }
105
106 @Override
107 public boolean isConnected() {
108 return connected;
109 }
110
111 @Override
112 public ChannelFuture joinGroup(InetAddress multicastAddress) {
113 return joinGroup(multicastAddress, newPromise());
114 }
115
116 @Override
117 public ChannelFuture joinGroup(InetAddress multicastAddress, ChannelPromise promise) {
118 try {
119 NetworkInterface iface = config().getNetworkInterface();
120 if (iface == null) {
121 iface = NetworkInterface.getByInetAddress(localAddress().getAddress());
122 }
123 return joinGroup(multicastAddress, iface, null, promise);
124 } catch (SocketException e) {
125 promise.setFailure(e);
126 }
127 return promise;
128 }
129
130 @Override
131 public ChannelFuture joinGroup(
132 InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
133 return joinGroup(multicastAddress, networkInterface, newPromise());
134 }
135
136 @Override
137 public ChannelFuture joinGroup(
138 InetSocketAddress multicastAddress, NetworkInterface networkInterface,
139 ChannelPromise promise) {
140 return joinGroup(multicastAddress.getAddress(), networkInterface, null, promise);
141 }
142
143 @Override
144 public ChannelFuture joinGroup(
145 InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
146 return joinGroup(multicastAddress, networkInterface, source, newPromise());
147 }
148
149 @Override
150 public ChannelFuture joinGroup(
151 final InetAddress multicastAddress, final NetworkInterface networkInterface,
152 final InetAddress source, final ChannelPromise promise) {
153
154 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
155 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
156
157 promise.setFailure(new UnsupportedOperationException("Multicast not supported"));
158 return promise;
159 }
160
161 @Override
162 public ChannelFuture leaveGroup(InetAddress multicastAddress) {
163 return leaveGroup(multicastAddress, newPromise());
164 }
165
166 @Override
167 public ChannelFuture leaveGroup(InetAddress multicastAddress, ChannelPromise promise) {
168 try {
169 return leaveGroup(
170 multicastAddress, NetworkInterface.getByInetAddress(localAddress().getAddress()), null, promise);
171 } catch (SocketException e) {
172 promise.setFailure(e);
173 }
174 return promise;
175 }
176
177 @Override
178 public ChannelFuture leaveGroup(
179 InetSocketAddress multicastAddress, NetworkInterface networkInterface) {
180 return leaveGroup(multicastAddress, networkInterface, newPromise());
181 }
182
183 @Override
184 public ChannelFuture leaveGroup(
185 InetSocketAddress multicastAddress,
186 NetworkInterface networkInterface, ChannelPromise promise) {
187 return leaveGroup(multicastAddress.getAddress(), networkInterface, null, promise);
188 }
189
190 @Override
191 public ChannelFuture leaveGroup(
192 InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source) {
193 return leaveGroup(multicastAddress, networkInterface, source, newPromise());
194 }
195
196 @Override
197 public ChannelFuture leaveGroup(
198 final InetAddress multicastAddress, final NetworkInterface networkInterface, final InetAddress source,
199 final ChannelPromise promise) {
200 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
201 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
202
203 promise.setFailure(new UnsupportedOperationException("Multicast not supported"));
204
205 return promise;
206 }
207
208 @Override
209 public ChannelFuture block(
210 InetAddress multicastAddress, NetworkInterface networkInterface,
211 InetAddress sourceToBlock) {
212 return block(multicastAddress, networkInterface, sourceToBlock, newPromise());
213 }
214
215 @Override
216 public ChannelFuture block(
217 final InetAddress multicastAddress, final NetworkInterface networkInterface,
218 final InetAddress sourceToBlock, final ChannelPromise promise) {
219 ObjectUtil.checkNotNull(multicastAddress, "multicastAddress");
220 ObjectUtil.checkNotNull(sourceToBlock, "sourceToBlock");
221 ObjectUtil.checkNotNull(networkInterface, "networkInterface");
222 promise.setFailure(new UnsupportedOperationException("Multicast not supported"));
223 return promise;
224 }
225
226 @Override
227 public ChannelFuture block(InetAddress multicastAddress, InetAddress sourceToBlock) {
228 return block(multicastAddress, sourceToBlock, newPromise());
229 }
230
231 @Override
232 public ChannelFuture block(
233 InetAddress multicastAddress, InetAddress sourceToBlock, ChannelPromise promise) {
234 try {
235 return block(
236 multicastAddress,
237 NetworkInterface.getByInetAddress(localAddress().getAddress()),
238 sourceToBlock, promise);
239 } catch (Throwable e) {
240 promise.setFailure(e);
241 }
242 return promise;
243 }
244
245 @Override
246 protected AbstractKQueueUnsafe newUnsafe() {
247 return new KQueueDatagramChannelUnsafe();
248 }
249
250 @Override
251 protected void doBind(SocketAddress localAddress) throws Exception {
252 if (localAddress instanceof InetSocketAddress) {
253 InetSocketAddress socketAddress = (InetSocketAddress) localAddress;
254 if (socketAddress.getAddress().isAnyLocalAddress() &&
255 socketAddress.getAddress() instanceof Inet4Address) {
256 if (socket.family() == SocketProtocolFamily.INET6) {
257 localAddress = new InetSocketAddress(Native.INET6_ANY, socketAddress.getPort());
258 }
259 }
260 }
261 super.doBind(localAddress);
262 active = true;
263 }
264
265 @Override
266 protected boolean doWriteMessage(Object msg) throws Exception {
267 final ByteBuf data;
268 InetSocketAddress remoteAddress;
269 if (msg instanceof AddressedEnvelope) {
270 @SuppressWarnings("unchecked")
271 AddressedEnvelope<ByteBuf, InetSocketAddress> envelope =
272 (AddressedEnvelope<ByteBuf, InetSocketAddress>) msg;
273 data = envelope.content();
274 remoteAddress = envelope.recipient();
275 } else {
276 data = (ByteBuf) msg;
277 remoteAddress = null;
278 }
279
280 final int dataLen = data.readableBytes();
281 if (dataLen == 0) {
282 return true;
283 }
284
285 final long writtenBytes;
286 if (data.hasMemoryAddress()) {
287 long memoryAddress = data.memoryAddress();
288 if (remoteAddress == null) {
289 try {
290 writtenBytes = socket.writeAddress(memoryAddress, data.readerIndex(), data.writerIndex());
291 } catch (Errors.NativeIoException e) {
292 throw translateForConnected(e);
293 }
294 } else {
295 writtenBytes = socket.sendToAddress(memoryAddress, data.readerIndex(), data.writerIndex(),
296 remoteAddress.getAddress(), remoteAddress.getPort());
297 }
298 } else if (data.nioBufferCount() > 1) {
299 IovArray array = ((NativeArrays) registration().attachment()).cleanIovArray();
300 array.add(data, data.readerIndex(), data.readableBytes());
301 int cnt = array.count();
302 assert cnt != 0;
303
304 if (remoteAddress == null) {
305 writtenBytes = socket.writevAddresses(array.memoryAddress(0), cnt);
306 } else {
307 writtenBytes = socket.sendToAddresses(array.memoryAddress(0), cnt,
308 remoteAddress.getAddress(), remoteAddress.getPort());
309 }
310 } else {
311 ByteBuffer nioData = data.internalNioBuffer(data.readerIndex(), data.readableBytes());
312 if (remoteAddress == null) {
313 writtenBytes = socket.write(nioData, nioData.position(), nioData.limit());
314 } else {
315 writtenBytes = socket.sendTo(nioData, nioData.position(), nioData.limit(),
316 remoteAddress.getAddress(), remoteAddress.getPort());
317 }
318 }
319
320 return writtenBytes > 0;
321 }
322
323 private static IOException translateForConnected(Errors.NativeIoException e) {
324
325 if (e.expectedErr() == Errors.ERROR_ECONNREFUSED_NEGATIVE) {
326 PortUnreachableException error = new PortUnreachableException(e.getMessage());
327 error.initCause(e);
328 return error;
329 }
330 return e;
331 }
332
333 private static void checkUnresolved(AddressedEnvelope<?, ?> envelope) {
334 if (envelope.recipient() instanceof InetSocketAddress
335 && (((InetSocketAddress) envelope.recipient()).isUnresolved())) {
336 throw new UnresolvedAddressException();
337 }
338 }
339
340 @Override
341 protected Object filterOutboundMessage(Object msg) {
342 if (msg instanceof DatagramPacket) {
343 DatagramPacket packet = (DatagramPacket) msg;
344 checkUnresolved(packet);
345 ByteBuf content = packet.content();
346 return UnixChannelUtil.isBufferCopyNeededForWrite(content) ?
347 new DatagramPacket(newDirectBuffer(packet, content), packet.recipient()) : msg;
348 }
349
350 if (msg instanceof ByteBuf) {
351 ByteBuf buf = (ByteBuf) msg;
352 return UnixChannelUtil.isBufferCopyNeededForWrite(buf) ? newDirectBuffer(buf) : buf;
353 }
354
355 if (msg instanceof AddressedEnvelope) {
356 @SuppressWarnings("unchecked")
357 AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
358 checkUnresolved(e);
359
360 if (e.content() instanceof ByteBuf &&
361 (e.recipient() == null || e.recipient() instanceof InetSocketAddress)) {
362
363 ByteBuf content = (ByteBuf) e.content();
364 return UnixChannelUtil.isBufferCopyNeededForWrite(content) ?
365 new DefaultAddressedEnvelope<ByteBuf, InetSocketAddress>(
366 newDirectBuffer(e, content), (InetSocketAddress) e.recipient()) : e;
367 }
368 }
369
370 throw new UnsupportedOperationException(
371 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
372 }
373
374 @Override
375 public KQueueDatagramChannelConfig config() {
376 return config;
377 }
378
379 @Override
380 protected void doDisconnect() throws Exception {
381 socket.disconnect();
382 connected = active = false;
383 resetCachedAddresses();
384 }
385
386 @Override
387 protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
388 if (super.doConnect(remoteAddress, localAddress)) {
389 connected = true;
390 return true;
391 }
392 return false;
393 }
394
395 @Override
396 protected void doClose() throws Exception {
397 super.doClose();
398 connected = false;
399 }
400
401 final class KQueueDatagramChannelUnsafe extends AbstractKQueueUnsafe {
402
403 @Override
404 void readReady(KQueueRecvByteAllocatorHandle allocHandle) {
405 assert eventLoop().inEventLoop();
406 final DatagramChannelConfig config = config();
407 if (shouldBreakReadReady(config)) {
408 clearReadFilter0();
409 return;
410 }
411 final ChannelPipeline pipeline = pipeline();
412 final ByteBufAllocator allocator = config.getAllocator();
413 allocHandle.reset(config);
414
415 Throwable exception = null;
416 try {
417 ByteBuf byteBuf = null;
418 try {
419 boolean connected = isConnected();
420 do {
421 byteBuf = allocHandle.allocate(allocator);
422 allocHandle.attemptedBytesRead(byteBuf.writableBytes());
423
424 final DatagramPacket packet;
425 if (connected) {
426 try {
427 allocHandle.lastBytesRead(doReadBytes(byteBuf));
428 } catch (Errors.NativeIoException e) {
429
430 if (e.expectedErr() == Errors.ERROR_ECONNREFUSED_NEGATIVE) {
431 PortUnreachableException error = new PortUnreachableException(e.getMessage());
432 error.initCause(e);
433 throw error;
434 }
435 throw e;
436 }
437 if (allocHandle.lastBytesRead() <= 0) {
438
439 byteBuf.release();
440 byteBuf = null;
441 break;
442 }
443 packet = new DatagramPacket(byteBuf,
444 (InetSocketAddress) localAddress(), (InetSocketAddress) remoteAddress());
445 } else {
446 final DatagramSocketAddress remoteAddress;
447 if (byteBuf.hasMemoryAddress()) {
448
449 remoteAddress = socket.recvFromAddress(byteBuf.memoryAddress(), byteBuf.writerIndex(),
450 byteBuf.capacity());
451 } else {
452 ByteBuffer nioData = byteBuf.internalNioBuffer(
453 byteBuf.writerIndex(), byteBuf.writableBytes());
454 remoteAddress = socket.recvFrom(nioData, nioData.position(), nioData.limit());
455 }
456
457 if (remoteAddress == null) {
458 allocHandle.lastBytesRead(-1);
459 byteBuf.release();
460 byteBuf = null;
461 break;
462 }
463 InetSocketAddress localAddress = remoteAddress.localAddress();
464 if (localAddress == null) {
465 localAddress = (InetSocketAddress) localAddress();
466 }
467 allocHandle.lastBytesRead(remoteAddress.receivedAmount());
468 byteBuf.writerIndex(byteBuf.writerIndex() + allocHandle.lastBytesRead());
469
470 packet = new DatagramPacket(byteBuf, localAddress, remoteAddress);
471 }
472
473 allocHandle.incMessagesRead(1);
474
475 readPending = false;
476 pipeline.fireChannelRead(packet);
477
478 byteBuf = null;
479
480
481
482 } while (allocHandle.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER));
483 } catch (Throwable t) {
484 if (byteBuf != null) {
485 byteBuf.release();
486 }
487 exception = t;
488 }
489
490 allocHandle.readComplete();
491 pipeline.fireChannelReadComplete();
492
493 if (exception != null) {
494 pipeline.fireExceptionCaught(exception);
495 }
496 } finally {
497 if (shouldStopReading(config)) {
498 clearReadFilter0();
499 }
500 }
501 }
502 }
503 }