Add UDP support to dart:io

tests/standalone/io/raw_datagram_socket_test.dart

+// Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import "dart:async";
+import "dart:io";
+import "dart:typed_data";
+
+import "package:async_helper/async_helper.dart";
+import "package:expect/expect.dart";
+
+class FutureExpect {
+  static Future check(Future result, check) =>
+      result.then((value) => check(value));
+  static Future throws(Future result) =>
+      result.then((value) {
+        throw new ExpectException(
+            "FutureExpect.throws received $value instead of an exception");
+        }, onError: (_) => null);
+}
+
+testDatagramSocketOptions() {
+  test(address) {
+    asyncStart();
+    RawDatagramSocket.bind(address, 0).then((socket) {
+      Expect.isTrue(socket.multicastLoopback);
+      Expect.equals(1, socket.multicastHops);
+      Expect.throws(() => socket.multicastInterface);
+      Expect.isFalse(socket.broadcastEnabled);
+
+      socket.multicastLoopback = false;
+      socket.multicastHops = 4;
+      socket.broadcastEnabled = true;
+      Expect.isFalse(socket.multicastLoopback);
+      Expect.equals(4, socket.multicastHops);
+      Expect.throws(() => socket.multicastInterface = null);
+      Expect.isTrue(socket.broadcastEnabled);
+
+      socket.multicastLoopback = true;
+      socket.multicastHops = 1;
+      socket.broadcastEnabled = false;
+      Expect.isTrue(socket.multicastLoopback);
+      Expect.equals(1, socket.multicastHops);
+      Expect.throws(() => socket.multicastInterface);
+      Expect.isFalse(socket.broadcastEnabled);
+      asyncEnd();
+    });
+  }
+
+  test(InternetAddress.LOOPBACK_IP_V4);
+  test(InternetAddress.ANY_IP_V4);
+  test(InternetAddress.LOOPBACK_IP_V6);
+  test(InternetAddress.ANY_IP_V6);
+}
+
+testDatagramSocketReuseAddress() {
+  test(address, reuseAddress) {
+    asyncStart();
+    RawDatagramSocket.bind(
+        address, 0, reuseAddress: reuseAddress).then((socket) {
+            if (reuseAddress) {
+              RawDatagramSocket.bind(address, socket.port)
+                  .then((s) => Expect.isTrue(s is RawDatagramSocket))
+                  .then(asyncSuccess);
+            } else {
+              FutureExpect.throws(RawDatagramSocket.bind(address, socket.port))
+                  .then(asyncSuccess);
+            }
+    });
+  }
+
+  test(InternetAddress.LOOPBACK_IP_V4, true);
+  test(new InternetAddress('224.0.0.0'), true);
+  test(InternetAddress.LOOPBACK_IP_V4, false);
+  test(new InternetAddress('224.0.0.0'), false);
+  test(InternetAddress.LOOPBACK_IP_V6, true);
+  test(new InternetAddress("ff11::0"), true);
+  test(InternetAddress.LOOPBACK_IP_V6, false);
+  test(new InternetAddress("ff11::0"), false);
+}
+
+testBroadcast() {
+  test(bindAddress, broadcastAddress, enabled) {
+    asyncStart();
+    Future.wait([RawDatagramSocket.bind(bindAddress, 0),
+                 RawDatagramSocket.bind(bindAddress, 0)]).then((values) {
+      var broadcastTimer;
+      var sender = values[0];
+      var receiver = values[1];
+      sender.broadcastEnabled = enabled;
+      receiver.listen((event) {
+        if (event == RawSocketEvent.READ) {
+          Expect.isTrue(enabled);
+          sender.close();
+          receiver.close();
+          broadcastTimer.cancel();
+          asyncEnd();
+        }
+      });
+
+      int sendCount = 0;
+      send(_) {
+        sender.send(new Uint8List(1), broadcastAddress, receiver.port);
+        sendCount++;
+        if (!enabled && sendCount == 50) {
+          sender.close();
+          receiver.close();
+          broadcastTimer.cancel();
+          asyncEnd();
+        }
+      }
+      broadcastTimer = new Timer.periodic(new Duration(milliseconds: 10), send);
+    });
+  }
+
+  var broadcast = new InternetAddress("255.255.255.255");
+  test(InternetAddress.ANY_IP_V4, broadcast, false);
+  test(InternetAddress.ANY_IP_V4, broadcast, true);
+}
+
+testLoopbackMulticast() {
+  test(bindAddress, multicastAddress, enabled) {
+    asyncStart();
+    Future.wait([RawDatagramSocket.bind(bindAddress, 0),
+                 RawDatagramSocket.bind(bindAddress, 0)]).then((values) {
+      var senderTimer;
+      var sender = values[0];
+      var receiver = values[1];
+
+      sender.join(multicastAddress);
+      receiver.join(multicastAddress);
+      receiver.multicastLoopback = false;
+      sender.multicastLoopback = enabled;
+
+      receiver.listen((event) {
+        if (event == RawSocketEvent.READ) {
+          Expect.isTrue(enabled);
+          sender.close();
+          receiver.close();
+          senderTimer.cancel();
+          asyncEnd();
+        }
+      });
+
+      int sendCount = 0;
+      send(_) {
+        sender.send(new Uint8List(1), multicastAddress, receiver.port);
+        sendCount++;
+        if (!enabled && sendCount == 50) {
+          sender.close();
+          receiver.close();
+          senderTimer.cancel();
+          asyncEnd();
+        }
+      }
+      senderTimer = new Timer.periodic(new Duration(milliseconds: 10), send);
+    });
+  }
+
+  test(InternetAddress.ANY_IP_V4, new InternetAddress("224.0.0.0"), true);
+  test(InternetAddress.ANY_IP_V4, new InternetAddress("224.0.0.0"), false);
+  test(InternetAddress.ANY_IP_V6, new InternetAddress("ff11::0"), true);
+  test(InternetAddress.ANY_IP_V6, new InternetAddress("ff11::0"), false);
+}
+
+testSendReceive(InternetAddress bindAddress) {
+  asyncStart();
+
+  var total = 10;
+
+  int receivedSeq = 0;
+  var ackSeq = 0;
+  Timer ackTimer;
+
+  Future.wait([RawDatagramSocket.bind(bindAddress, 0),
+               RawDatagramSocket.bind(bindAddress, 0)]).then((values) {
+  var sender = values[0];
+  var receiver = values[1];
+  if (bindAddress.isMulticast) {
+    sender.multicastLoopback = true;
+    receiver.multicastLoopback = true;
+    sender.join(bindAddress);
+    receiver.join(bindAddress);
+  }
+
+  UInt8List createDataPackage(int seq) {
+    var data = new Uint8List(1000);
+    (new ByteData.view(data.buffer, 0, 4)).setUint32(0, seq);
+    return data;
+  }
+
+  UInt8List createAckPackage(int seq) {
+    var data = new Uint8List(4);
+    new ByteData.view(data.buffer, 0, 4).setUint32(0, seq);
+    return data;
+  }
+
+  int packageSeq(Datagram datagram) =>
+      new ByteData.view(datagram.data.buffer).getUint32(0);
+
+  void sendData(int seq) {
+    // Send a datagram acknowledging the received sequence.
+    sender.send(createDataPackage(seq), bindAddress, receiver.port);

[Anders Johnsen, 2013/11/29 12:36:54]

Check send return value.

[Søren Gjesse, 2013/12/12 11:38:46]

Done.

+  }
+
+  void sendAck(address, port) {
+    // Send a datagram acknowledging the received sequence.
+    receiver.send(createAckPackage(receivedSeq), address, port);
+    // Start a "long" timer for more data.
+    if (ackTimer != null) ackTimer.cancel();
+    ackTimer = new Timer.periodic(
+        new Duration(milliseconds: 100), (_) => sendAck(address, port));
+  }
+
+  sender.listen((event) {
+    switch (event) {
+      case RawSocketEvent.READ:
+        var datagram = sender.receive();
+        if (datagram != null) {
+          Expect.equals(datagram.port, receiver.port);
+          if (!bindAddress.isMulticast) {
+            Expect.equals(receiver.address, datagram.address);
+          }
+          ackSeq = packageSeq(datagram);
+          if (ackSeq < total) {
+            sender.writeEventsEnabled = true;
+          } else {
+            sender.close();
+            receiver.close();
+            ackTimer.cancel();
+            asyncEnd();
+          }
+        }
+        break;
+      case RawSocketEvent.WRITE:
+        // Send the next package.
+        sendData(ackSeq + 1);
+        break;
+      case RawSocketEvent.CLOSED:
+        break;
+      default:
+        throw "Unexpected event $event";
+    }
+  });
+
+  receiver.writeEventsEnabled = false;
+  receiver.listen((event) {
+    switch (event) {
+      case RawSocketEvent.READ:
+        var datagram = receiver.receive();
+        if (datagram != null) {
+          Expect.equals(datagram.port, sender.port);
+          if (!bindAddress.isMulticast) {
+            Expect.equals(receiver.address, datagram.address);
+          }
+          var seq = packageSeq(datagram);
+          if (seq == receivedSeq + 1) {
+            receivedSeq = seq;
+            sendAck(bindAddress, sender.port);
+          }
+        }
+        break;
+      case RawSocketEvent.WRITE:
+        throw "Unexpected WRITE";
+        break;
+      case RawSocketEvent.CLOSED:
+        break;
+      default:
+        throw "Unexpected event $event";
+    }
+  });
+  });
+}
+
+
+main() {
+  testDatagramSocketOptions();
+  testDatagramSocketReuseAddress();
+  testBroadcast();
+  testLoopbackMulticast();
+  testSendReceive(InternetAddress.LOOPBACK_IP_V4);
+  testSendReceive(new InternetAddress("224.0.0.0"));
+  testSendReceive(InternetAddress.LOOPBACK_IP_V6);
+  testSendReceive(new InternetAddress("ff11::114"));
+}