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