Add Source Specific Multicast (SSM) support

SSM can be enabled using "source" option in UDP options. Previously, "source" is considered the same as "interface", which is incorrect. 

Fixes #332.

docs/source/options/udp_file_options.rst

@@ -8,11 +8,16 @@ options:
 
     Allow or disallow reusing UDP sockets.
 
-:interface=<addr>, source=<addr>:
+:interface=<addr>:
 
-    Multicast group interface address. Only the packets sent to this address is
+    Multicast group interface address. Only the packets sent to this address are
     received. Default to "0.0.0.0" if not specified.
 
+:source=<addr>:
+
+    Multicast source ip address. Only the packets sent from this source address
+    are received. Defaults to "0.0.0.0" if not specified.
+
 :timeout=<microseconds>:
 
     UDP timeout in microseconds.

packager/file/udp_file.cc

@@ -204,7 +204,37 @@ bool UdpFile::Open() {
   }
 
   if (is_multicast) {
+    if (options->is_source_specific_multicast()) {
+      struct ip_mreq_source source_multicast_group;
+
+      source_multicast_group.imr_multiaddr = local_in_addr;
+      if (inet_pton(AF_INET,
+                    options->interface_address().c_str(),
+                    &source_multicast_group.imr_interface) != 1) {
+        LOG(ERROR) << "Malformed IPv4 interface address "
+                   << options->interface_address();
+        return false;
+      }
+      if (inet_pton(AF_INET,
+                    options->source_address().c_str(),
+                    &source_multicast_group.imr_sourceaddr) != 1) {
+        LOG(ERROR) << "Malformed IPv4 source specific multicast address "
+                   << options->source_address();
+        return false;
+      }
+
+      if (setsockopt(new_socket.get(),
+                     IPPROTO_IP,
+                     IP_ADD_SOURCE_MEMBERSHIP,
+                     reinterpret_cast<const char*>(&source_multicast_group),
+                     sizeof(source_multicast_group)) < 0) {
+          LOG(ERROR) << "Failed to join multicast group.";
+          return false;
+      }
+    } else {
+      // this is a v2 join without a specific source.
       struct ip_mreq multicast_group;
+
       multicast_group.imr_multiaddr = local_in_addr;
 
       if (inet_pton(AF_INET, options->interface_address().c_str(),
@@ -221,6 +251,8 @@ bool UdpFile::Open() {
         return false;
       }
 
+  }
+
 #if defined(__linux__)
     // Disable IP_MULTICAST_ALL to avoid interference caused when two sockets
     // are bound to the same port but joined to different multicast groups.

packager/file/udp_options.cc

@@ -24,6 +24,7 @@ enum FieldType {
   kUnknownField = 0,
   kReuseField,
   kInterfaceAddressField,
+  kMulticastSourceField,
   kTimeoutField,
 };
 
@@ -35,7 +36,7 @@ struct FieldNameToTypeMapping {
 const FieldNameToTypeMapping kFieldNameTypeMappings[] = {
     {"reuse", kReuseField},
     {"interface", kInterfaceAddressField},
-    {"source", kInterfaceAddressField},
+    {"source", kMulticastSourceField},
     {"timeout", kTimeoutField},
 };
 
@@ -105,6 +106,10 @@ std::unique_ptr<UdpOptions> UdpOptions::ParseFromString(
             return nullptr;
           }
           break;
+        case kMulticastSourceField:
+          options->source_address_ = pair.second;
+          options->is_source_specific_multicast_ = true;
+          break;
         default:
           LOG(ERROR) << "Unknown field in udp options (\"" << pair.first
                      << "\").";

packager/file/udp_options.h

@@ -26,6 +26,10 @@ class UdpOptions {
   bool reuse() const { return reuse_; }
   const std::string& interface_address() const { return interface_address_; }
   unsigned timeout_us() const { return timeout_us_; }
+  const std::string& source_address() const { return source_address_; }
+  bool is_source_specific_multicast() const {
+    return is_source_specific_multicast_;
+  }
 
  private:
   UdpOptions() = default;
@@ -39,6 +43,9 @@ class UdpOptions {
   std::string interface_address_ = "0.0.0.0";
   /// Timeout in microseconds. 0 to indicate unlimited timeout.
   unsigned timeout_us_ = 0;
+  // Source specific multicast source address
+  std::string source_address_ = "0.0.0.0";
+  bool is_source_specific_multicast_ = false;
 };
 
 }  // namespace shaka

packager/file/udp_options_unittest.cc

@@ -27,6 +27,8 @@ TEST_F(UdpOptionsTest, AddressAndPort) {
   EXPECT_FALSE(options->reuse());
   EXPECT_EQ("0.0.0.0", options->interface_address());
   EXPECT_EQ(0u, options->timeout_us());
+  EXPECT_FALSE(options->is_source_specific_multicast());
+  EXPECT_EQ("0.0.0.0", options->source_address());
 }
 
 TEST_F(UdpOptionsTest, MissingPort) {
@@ -55,6 +57,8 @@ TEST_F(UdpOptionsTest, UdpInterfaceAddressFlag) {
   EXPECT_FALSE(options->reuse());
   EXPECT_EQ("10.11.12.13", options->interface_address());
   EXPECT_EQ(0u, options->timeout_us());
+  EXPECT_FALSE(options->is_source_specific_multicast());
+  EXPECT_EQ("0.0.0.0", options->source_address());
 }
 
 TEST_F(UdpOptionsTest, Reuse) {
@@ -64,6 +68,8 @@ TEST_F(UdpOptionsTest, Reuse) {
   EXPECT_TRUE(options->reuse());
   EXPECT_EQ("0.0.0.0", options->interface_address());
   EXPECT_EQ(0u, options->timeout_us());
+  EXPECT_FALSE(options->is_source_specific_multicast());
+  EXPECT_EQ("0.0.0.0", options->source_address());
 }
 
 TEST_F(UdpOptionsTest, InvalidReuse) {
@@ -78,21 +84,37 @@ TEST_F(UdpOptionsTest, InterfaceAddress) {
   EXPECT_FALSE(options->reuse());
   EXPECT_EQ("10.11.12.13", options->interface_address());
   EXPECT_EQ(0u, options->timeout_us());
+  EXPECT_FALSE(options->is_source_specific_multicast());
+  EXPECT_EQ("0.0.0.0", options->source_address());
+}
+
+TEST_F(UdpOptionsTest, SourceAddress) {
+  auto options = UdpOptions::ParseFromString(
+      "224.1.2.30:88?interface=10.11.12.13&source=10.14.15.16");
+  EXPECT_EQ("224.1.2.30", options->address());
+  EXPECT_EQ(88u, options->port());
+  EXPECT_FALSE(options->reuse());
+  EXPECT_EQ("10.11.12.13", options->interface_address());
+  EXPECT_EQ(0u, options->timeout_us());
+  EXPECT_TRUE(options->is_source_specific_multicast());
+  EXPECT_EQ("10.14.15.16", options->source_address());
 }
 
 TEST_F(UdpOptionsTest, Timeout) {
   auto options = UdpOptions::ParseFromString(
-      "224.1.2.30:88?source=10.11.12.13&timeout=88888888");
+      "224.1.2.30:88?interface=10.11.12.13&timeout=88888888");
   EXPECT_EQ("224.1.2.30", options->address());
   EXPECT_EQ(88u, options->port());
   EXPECT_FALSE(options->reuse());
   EXPECT_EQ("10.11.12.13", options->interface_address());
   EXPECT_EQ(88888888u, options->timeout_us());
+  EXPECT_FALSE(options->is_source_specific_multicast());
+  EXPECT_EQ("0.0.0.0", options->source_address());
 }
 
 TEST_F(UdpOptionsTest, InvalidTimeout) {
   ASSERT_FALSE(UdpOptions::ParseFromString(
-      "224.1.2.30:88?source=10.11.12.13&timeout=1a9"));
+      "224.1.2.30:88?interface=10.11.12.13&timeout=1a9"));
 }
 
 }  // namespace shaka