Multicast DNS implementation (initial)

net/dns/mdns_client_impl.cc

 // Copyright (c) 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/dns/mdns_client_impl.h"
 
 #include "base/bind.h"
 #include "base/message_loop_proxy.h"
 #include "base/stl_util.h"
 #include "base/time/default_clock.h"
 #include "net/base/dns_util.h"
 #include "net/base/net_errors.h"
 #include "net/base/net_log.h"
 #include "net/base/rand_callback.h"
 #include "net/dns/dns_protocol.h"
-#include "net/dns/mdns_query.h"
 #include "net/udp/datagram_socket.h"
 
 namespace net {
 
-static const char kMDNSMulticastGroupIPv4[] = "224.0.0.251";
-static const char kMDNSMulticastGroupIPv6[] = "FF02::FB";
+static const char kMDnsMulticastGroupIPv4[] = "224.0.0.251";
+static const char kMDnsMulticastGroupIPv6[] = "FF02::FB";
 
-static const unsigned kMDnsTransactionTimeoutSeconds = 3;
+static const unsigned kMDnsTransactionNoResultsSeconds = 3;
 
 MDnsClientImpl::Core::Core(MDnsClientImpl* client,
                            MDnsConnectionFactory* connection_factory)
     : client_(client), connection_(connection_factory->CreateConnection(this)) {
 }
 
 MDnsClientImpl::Core::~Core() {
   cleanup_callback_.Cancel();
   STLDeleteValues(&listeners_);
 }
 
 bool MDnsClientImpl::Core::Init() {
-  return connection_->Init();
+  return connection_->Init() == OK;
 }
 
 bool MDnsClientImpl::Core::SendQuery(uint16 rrtype, std::string name) {
   std::string name_dns;
   if (!DNSDomainFromDot(name, &name_dns))
     return false;
 
-  MDnsQuery query(name_dns, rrtype);
+  // TODO(noamsml): Remove RD flag from query and add known-answer suppression.
+  DnsQuery query(0, name_dns, rrtype);
 
-  connection_->Send(query.io_buffer(), query.size());
-
-  return true;
+  return connection_->Send(query.io_buffer(), query.io_buffer()->size()) == OK;
 }
 
 void MDnsClientImpl::Core::HandlePacket(DnsResponse* response,
                                         int bytes_read) {
   unsigned offset;
 
   if (!response->InitParseWithoutQuery(bytes_read)) {
     LOG(WARNING) << "Could not understand an mDNS packet.";
     return;  // Message is unreadable.
   }
@@ skipping 54 matching lines @@
       }
 
       AlertListeners(update_external,
                      ListenerKey(record->type(), record->name()), record);
       // Alert listeners listening only for rrtype and not for name.
       AlertListeners(update_external, ListenerKey(record->type(), ""), record);
     }
   }
 }
 
+void MDnsClientImpl::Core::OnConnectionError(int error) {
+  // TODO(noamsml): On connection error, recreate connection and flush cache.
+}
+
 void MDnsClientImpl::Core::AlertListeners(
     MDnsUpdateType update_type,
     const ListenerKey& key,
     const RecordParsed* record) {
   ListenerMap::iterator listener_map_iterator = listeners_.find(key);
   if (listener_map_iterator == listeners_.end()) return;
 
   FOR_EACH_OBSERVER(MDnsListenerImpl, *listener_map_iterator->second,
                     AlertDelegate(update_type, record));
 }
 
 void MDnsClientImpl::Core::AddListener(
-    MDnsListenerImpl* listener, bool alert_existing_records) {
+    MDnsListenerImpl* listener) {
   ListenerKey key(listener->GetType(), listener->GetName());
   std::pair<ListenerMap::iterator, bool> observer_insert_result =
       listeners_.insert(
           make_pair(key, static_cast<ObserverList<MDnsListenerImpl>*>(NULL)));
 
   // If an equivalent key does not exist, actually create the observer list.
   if (observer_insert_result.second) {
     observer_insert_result.first->second = new ObserverList<MDnsListenerImpl>();
   }
 
   ObserverList<MDnsListenerImpl>* observer_list =
       observer_insert_result.first->second;
 
   observer_list->AddObserver(listener);
-
-  if (alert_existing_records) {
-    std::vector<const RecordParsed*> records;
-
-    cache_.FindDnsRecords(listener->GetType(), listener->GetName(),
-                          &records, base::Time::Now());
-
-    base::WeakPtr<MDnsListenerImpl> weak_listener = listener->AsWeakPtr();
-
-    for (std::vector<const RecordParsed*>::iterator i = records.begin();
-         i != records.end(); i++) {
-      if (weak_listener) {
-        weak_listener->AlertDelegate(kMDnsRecordAdded, *i);
-      }
-    }
-  }
 }
 
 void MDnsClientImpl::Core::RemoveListener(MDnsListenerImpl* listener) {
   ListenerKey key(listener->GetType(), listener->GetName());
   ListenerMap::iterator observer_list_iterator = listeners_.find(key);
 
   DCHECK(observer_list_iterator != listeners_.end());
   DCHECK(observer_list_iterator->second->HasObserver(listener));
 
   observer_list_iterator->second->RemoveObserver(listener);
@@ skipping 85 matching lines @@
   }
 }
 
 bool MDnsClientImpl::IsListeningForTests() {
   return core_.get() != NULL;
 }
 
 scoped_ptr<MDnsListener> MDnsClientImpl::CreateListener(
     uint16 rrtype,
     const std::string& name,
-    bool active,
-    bool alert_existing_records,
     MDnsListener::Delegate* delegate) {
   return scoped_ptr<net::MDnsListener>(
-      new MDnsListenerImpl(rrtype, name, active, alert_existing_records,
-                           delegate, this));
+      new MDnsListenerImpl(rrtype, name, delegate, this));
 }
 
 scoped_ptr<MDnsTransaction> MDnsClientImpl::CreateTransaction(
     uint16 rrtype,
     const std::string& name,
+    int flags,
     const MDnsTransaction::ResultCallback& callback) {
   return scoped_ptr<MDnsTransaction>(
-      new MDnsTransactionImpl(rrtype, name, callback, this));
+      new MDnsTransactionImpl(rrtype, name, flags, callback, this));
 }
 
 MDnsListenerImpl::MDnsListenerImpl(
     uint16 rrtype,
     const std::string& name,
-    bool active,
-    bool alert_existing_records,
     MDnsListener::Delegate* delegate,
     MDnsClientImpl* client)
-    : rrtype_(rrtype), name_(name), active_(active),
-      alert_existing_records_(alert_existing_records), client_(client),
-      delegate_(delegate), started_(false) {
+    : rrtype_(rrtype), name_(name), client_(client), delegate_(delegate),
+      started_(false) {
 }
 
 bool MDnsListenerImpl::Start() {
   DCHECK(!started_);
 
   if (!client_->AddListenRef()) return false;
   started_ = true;
 
-  client_->core()->AddListener(this, alert_existing_records_);
+  client_->core()->AddListener(this);
 
-  if (active_) SendQuery(false);  // TODO(noamsml): Retry logic.
   return true;
 }
 
 MDnsListenerImpl::~MDnsListenerImpl() {
   if (started_) {
     client_->core()->RemoveListener(this);
     client_->SubtractListenRef();
   }
 }
 
 const std::string& MDnsListenerImpl::GetName() const {
   return name_;
 }
 
 uint16 MDnsListenerImpl::GetType() const {
   return rrtype_;
 }
 
-bool MDnsListenerImpl::IsActive() const {
-  return active_;
-}
-
-bool MDnsListenerImpl::SendQuery(bool force_refresh_cache) {
-  DCHECK(started_);
-  // TODO(noamsml): Logic related to force_refresh_cache
-  if (name_.size() == 0) return false;
-  return client_->core()->SendQuery(rrtype_, name_);
-}
-
-bool MDnsListenerImpl::QueryCache(
-    std::vector<const RecordParsed*>* records) const {
-  DCHECK(started_);
-  if (name_.size() == 0) return false;
-  client_->core()->QueryCache(rrtype_, name_, records);
-  return true;
-}
-
 void MDnsListenerImpl::AlertDelegate(MDnsUpdateType update_type,
                                      const RecordParsed* record) {
   DCHECK(started_);
   delegate_->OnRecordUpdate(update_type, record);
 }
 
 MDnsTransactionImpl::MDnsTransactionImpl(
     uint16 rrtype,
     const std::string& name,
+    int flags,
     const MDnsTransaction::ResultCallback& callback,
     MDnsClientImpl* client)
     : rrtype_(rrtype), name_(name), callback_(callback), client_(client),
-      started_(false) {
+      started_(false), flags_(flags) {
+  DCHECK((flags_ & kMDnsTransactionFlagMask) == flags_);
+  DCHECK(flags_ & kMDnsTransactionQueryCache ||
+         flags_ & kMDnsTransactionQueryNetwork);
 }
 
 MDnsTransactionImpl::~MDnsTransactionImpl() {
+  timeout_.Cancel();
 }
 
 bool MDnsTransactionImpl::Start() {
   DCHECK(!started_);
   started_ = true;
   std::vector<const RecordParsed*> records;
-  if (client_->core()) {
-    client_->core()->QueryCache(rrtype_, name_, &records);
-    if (!records.empty()) {
-      TriggerCallback(kMDnsTransactionSuccess, records.front());
-      return true;
+  base::WeakPtr<MDnsTransactionImpl> weak_this = AsWeakPtr();
+
+  if (flags_ & kMDnsTransactionQueryCache) {
+    if (client_->core()) {
+      client_->core()->QueryCache(rrtype_, name_, &records);
+      for (std::vector<const RecordParsed*>::iterator i = records.begin();
+           i != records.end() && weak_this; ++i) {
+        weak_this->TriggerCallback(kMDnsTransactionRecord, records.front());
+      }
     }
   }
 
-  listener_ = client_->CreateListener(rrtype_, name_, true /*active*/,
-                                      false /*alert existing*/, this);
-  listener_->Start();
+  if (!weak_this) return true;
 
-  timeout_.Reset(base::Bind(&MDnsTransactionImpl::OnTimedOut, AsWeakPtr()));
-  base::MessageLoop::current()->PostDelayedTask(
-      FROM_HERE,
-      timeout_.callback(),
-      base::TimeDelta::FromSeconds(kMDnsTransactionTimeoutSeconds));
+  if (is_active() && flags_ & kMDnsTransactionQueryNetwork) {
+    listener_ = client_->CreateListener(rrtype_, name_, this);
+    if (!listener_->Start()) return false;
 
-  return listener_.get() != NULL;
+    DCHECK(client_->core());
+    if (!client_->core()->SendQuery(rrtype_, name_)) {
+      return false;
+    }
+
+    timeout_.Reset(base::Bind(&MDnsTransactionImpl::SignalTransactionOver,
+                              weak_this));
+    base::MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        timeout_.callback(),
+        base::TimeDelta::FromSeconds(kMDnsTransactionNoResultsSeconds));
+
+    return listener_.get() != NULL;
+  } else {
+    // If this is a cache only query, signal that the transaction is over
+    // immediately.
+    SignalTransactionOver();
+  }
+
+  return true;
 }
 
 const std::string& MDnsTransactionImpl::GetName() const {
   return name_;
 }
 
 uint16 MDnsTransactionImpl::GetType() const {
   return rrtype_;
 }
 
 void MDnsTransactionImpl::CacheRecordFound(const RecordParsed* record) {
   DCHECK(started_);
   OnRecordUpdate(kMDnsRecordAdded, record);
 }
 
 void MDnsTransactionImpl::TriggerCallback(MDnsTransactionResult result,
                                           const RecordParsed* record) {
   DCHECK(started_);
-  if (callback_.is_null()) return;
+  if (!is_active()) return;
 
   // Ensure callback is run after touching all class state, so that
   // the callback can delete the transaction.
   MDnsTransaction::ResultCallback callback = callback_;
 
+  if (flags_ & kMDnsTransactionSingleResult) {
+    Reset();
+  }
+
+  callback.Run(result, record);
+}
+
+void MDnsTransactionImpl::Reset() {
   callback_.Reset();
   listener_.reset();
   timeout_.Cancel();
-
-  callback.Run(result, record);
 }
 
 void MDnsTransactionImpl::OnRecordUpdate(MDnsUpdateType update,
                                          const RecordParsed* record) {
   DCHECK(started_);
   if (update ==  kMDnsRecordAdded || update == kMDnsRecordChanged) {
-    TriggerCallback(kMDnsTransactionSuccess, record);
+    TriggerCallback(kMDnsTransactionRecord, record);
   }
 }
 
-void MDnsTransactionImpl::OnTimedOut() {
+void MDnsTransactionImpl::SignalTransactionOver() {
   DCHECK(started_);
-  TriggerCallback(kMDnsTransactionTimeout, NULL);
+  base::WeakPtr<MDnsTransactionImpl> weak_this = AsWeakPtr();
+
+  if (flags_ & kMDnsTransactionSingleResult) {
+    TriggerCallback(kMDnsTransactionNoResults, NULL);
+  } else {
+    TriggerCallback(kMDnsTransactionDone, NULL);
+  }
+
+  if (weak_this) {
+    weak_this->Reset();
+  }
 }
 
 void MDnsTransactionImpl::OnNsecRecord(const std::string& name, unsigned type) {
   // TODO(noamsml): NSEC records not yet implemented
 }
 
-MDnsConnectionImpl::RecvLoop::RecvLoop(
-    DatagramServerSocket* socket, MDnsConnectionImpl* connection)
-    : socket_(socket), connection_(connection),
-      response_(new DnsResponse(dns_protocol::kMaxMulticastSize)) {
+void MDnsTransactionImpl::OnCachePurged() {
+  // TODO(noamsml): Cache purge situations not yet implemented
 }
 
-MDnsConnectionImpl::RecvLoop::~RecvLoop() {
+MDnsConnectionImpl::SocketHandler::SocketHandler(
+    MDnsConnectionImpl* connection, const IPEndPoint& multicast_addr)
+    : socket_owned_(new UDPServerSocket(NULL, NetLog::Source())),
+      socket_(socket_owned_.get()), connection_(connection),
+      response_(new DnsResponse(dns_protocol::kMaxMulticastSize)),
+      multicast_addr_(multicast_addr) {
 }
 
-void MDnsConnectionImpl::RecvLoop::DoLoop(int rv) {
+MDnsConnectionImpl::SocketHandler::SocketHandler(
+    DatagramServerSocket* socket,
+    MDnsConnectionImpl* connection, const IPEndPoint& multicast_addr)
+    : socket_(socket), connection_(connection),
+      response_(new DnsResponse(dns_protocol::kMaxMulticastSize)),
+      multicast_addr_(multicast_addr) {
+}
+
+
+MDnsConnectionImpl::SocketHandler::~SocketHandler() {
+}
+
+int MDnsConnectionImpl::SocketHandler::Start() {
+  int rv = BindSocket();
+  if (rv != OK) {
+    return rv;
+  }
+
+  return DoLoop(0);
+}
+
+int MDnsConnectionImpl::SocketHandler::DoLoop(int rv) {
   do {
     if (rv > 0) {
       connection_->OnDatagramReceived(socket_,
                                       response_.get(), &recv_addr_, rv);
     }
 
     rv = socket_->RecvFrom(
         response_->io_buffer(),
         response_->io_buffer()->size(),
         &recv_addr_,
-        base::Bind(&MDnsConnectionImpl::RecvLoop::OnDatagramReceived,
+        base::Bind(&MDnsConnectionImpl::SocketHandler::OnDatagramReceived,
                    base::Unretained(this)));
   } while (rv > 0);
 
   if (rv != ERR_IO_PENDING) {
-    connection_->OnError(this, socket_, rv);
+    return rv;
+  }
+
+  return OK;
+}
+
+void MDnsConnectionImpl::SocketHandler::OnDatagramReceived(int rv) {
+  if (rv >= OK) {
+    rv = DoLoop(rv);
+  }
+
+  if (rv != OK) {
+    connection_->OnError(this, rv);
   }
 }
 
-void MDnsConnectionImpl::RecvLoop::OnDatagramReceived(int rv) {
-  DoLoop(rv);
+int MDnsConnectionImpl::SocketHandler::Send(IOBuffer* buffer, unsigned size) {
+  return socket_->SendTo(
+      buffer, size, multicast_addr_,
+      base::Bind(&MDnsConnectionImpl::SocketHandler::SendDone,
+                 base::Unretained(this) ));
 }
 
+void MDnsConnectionImpl::SocketHandler::SendDone(int sent) {
+  // TODO(noamsml): Retry logic.
+}
+
+int MDnsConnectionImpl::SocketHandler::BindSocket() {
+  IPAddressNumber address_any;
+  address_any.resize(multicast_addr_.address().size(), 0);
+
+  IPEndPoint bind_endpoint(address_any, multicast_addr_.port());
+
+  socket_->AllowAddressReuse();
+  int rv = socket_->Listen(bind_endpoint);
+
+  if (rv < OK) return rv;
+
+  socket_->SetMulticastLoopbackMode(false);
+
+  return socket_->JoinGroup(multicast_addr_.address());
+}
+
+
 MDnsConnectionImpl::MDnsConnectionImpl(MDnsConnection::Delegate* delegate)
-    : socket_ipv4_(new UDPServerSocket(NULL, NetLog::Source())),
-      socket_ipv6_(new UDPServerSocket(NULL, NetLog::Source())),
-      loop_ipv4_(socket_ipv4_.get(), this),
-      loop_ipv6_(socket_ipv6_.get(), this),
+    : socket_handler_ipv4_(this, GetIPEndPoint(kMDnsMulticastGroupIPv4,

[szym, 2013/06/10 21:58:28]

If you always use this with the same port, why not make it
GetMDnsIPEndPoint(const string& multicast_group)?

[Noam Samuel, 2013/06/11 20:35:03]

Done.

+                                     dns_protocol::kDefaultPortMulticast)),
+      socket_handler_ipv6_(this, GetIPEndPoint(kMDnsMulticastGroupIPv6,
+                                     dns_protocol::kDefaultPortMulticast)),
+  delegate_(delegate) {
+}
+
+MDnsConnectionImpl::MDnsConnectionImpl(DatagramServerSocket* socket_ipv4,
+                                       DatagramServerSocket* socket_ipv6,
+                                       MDnsConnection::Delegate* delegate)
+    : socket_handler_ipv4_(socket_ipv4, this,
+                           GetIPEndPoint(kMDnsMulticastGroupIPv4,
+                                         dns_protocol::kDefaultPortMulticast)),
+      socket_handler_ipv6_(socket_ipv6, this,
+                           GetIPEndPoint(kMDnsMulticastGroupIPv6,
+                                         dns_protocol::kDefaultPortMulticast)),
   delegate_(delegate) {
 }
 
 MDnsConnectionImpl::~MDnsConnectionImpl() {
-  socket_ipv4_->Close();
-  socket_ipv6_->Close();
 }
 
-bool MDnsConnectionImpl::Init() {
-  if (!BindSocket(socket_ipv4_.get(), kIPv4AddressSize,
-                  kMDNSMulticastGroupIPv4))
-    return false;
+int MDnsConnectionImpl::Init() {
+  int rv;
 
-  if (!BindSocket(socket_ipv6_.get(), kIPv6AddressSize,
-                  kMDNSMulticastGroupIPv6))
-    return false;
-
-  loop_ipv4_.DoLoop(0);
-  loop_ipv6_.DoLoop(0);
+  rv = socket_handler_ipv4_.Start();
+  if (rv != OK) return rv;
+  rv = socket_handler_ipv6_.Start();
+  if (rv != OK) return rv;
 
   return true;
 }
 
-bool MDnsConnectionImpl::Send(IOBuffer* buffer, unsigned size) {
-  int rv = socket_ipv4_->SendTo(
-      buffer,
-      size,
-      GetIPv4SendEndpoint(),
-      base::Bind(&MDnsConnectionImpl::SendDone,
-                 base::Unretained(this) ));
-  if (rv < OK && rv != ERR_IO_PENDING) return false;
+int MDnsConnectionImpl::Send(IOBuffer* buffer, unsigned size) {
+  int rv;
 
-  rv = socket_ipv6_->SendTo(
-      buffer,
-      size,
-      GetIPv6SendEndpoint(),
-      base::Bind(&MDnsConnectionImpl::SendDone,
-                 base::Unretained(this) ));
-  if (rv < OK && rv != ERR_IO_PENDING) return false;
+  rv = socket_handler_ipv4_.Send(buffer, size);
+  if (rv < OK && rv != ERR_IO_PENDING) return rv;
 
-  return true;
+  rv = socket_handler_ipv6_.Send(buffer, size);
+  if (rv < OK && rv != ERR_IO_PENDING) return rv;
+
+  return OK;
 }
 
-void MDnsConnectionImpl::SendDone(int sent) {
-  // TODO(noamsml): Queueing and retry logic.
+void MDnsConnectionImpl::OnError(SocketHandler* loop,
+                                 int error) {
+  // TODO(noamsml): Specific handling of intermittent errors that can be handled
+  // in the connection.
+  delegate_->OnConnectionError(error);
 }
 
-void MDnsConnectionImpl::OnError(RecvLoop* loop,
-                                 DatagramServerSocket* socket,
-                                 int error) {
-  // TODO(noamsml): Error handling.
-}
-
-bool MDnsConnectionImpl::BindSocket(
-    DatagramServerSocket* socket,
-    int addr_size,
-    const char* multicast_group) {
-  IPAddressNumber address_any;
-  address_any.resize(addr_size, 0);
-
+IPEndPoint MDnsConnectionImpl::GetIPEndPoint(const char* address, int port) {
   IPAddressNumber multicast_group_number;
-
-  IPEndPoint bind_endpoint(address_any, dns_protocol::kDefaultPortMulticast);
-
-  bool success = ParseIPLiteralToNumber(multicast_group,
+  bool success = ParseIPLiteralToNumber(address,
                                         &multicast_group_number);
   DCHECK(success);
-
-  socket->AllowAddressReuse();
-  int status = socket->Listen(bind_endpoint);
-
-  if (status < 0)
-    return false;
-
-  socket->SetMulticastLoopbackMode(false);
-
-  status = socket->JoinGroup(multicast_group_number);
-
-  if (status < 0)
-    return false;
-
-  return true;
-}
-
-IPEndPoint MDnsConnectionImpl::GetIPv4SendEndpoint() {
-  IPAddressNumber multicast_group_number;
-  bool success = ParseIPLiteralToNumber(kMDNSMulticastGroupIPv4,
-                                        &multicast_group_number);
-  DCHECK(success);
-  return IPEndPoint(multicast_group_number,
-                    dns_protocol::kDefaultPortMulticast);
-}
-
-IPEndPoint MDnsConnectionImpl::GetIPv6SendEndpoint() {
-  IPAddressNumber multicast_group_number;
-  bool success = ParseIPLiteralToNumber(kMDNSMulticastGroupIPv6,
-                                        &multicast_group_number);
-  DCHECK(success);
-  return IPEndPoint(multicast_group_number,
-                    dns_protocol::kDefaultPortMulticast);
+  return IPEndPoint(multicast_group_number, port);
 }
 
 void MDnsConnectionImpl::OnDatagramReceived(
     DatagramServerSocket* socket,
     DnsResponse* response,
     IPEndPoint* recv_addr,
     int bytes_read) {
   // TODO(noamsml): More sophisticated error handling.
   DCHECK_GT(bytes_read, 0);
   delegate_->HandlePacket(response, bytes_read);
 }
 
 MDnsConnectionImplFactory::MDnsConnectionImplFactory() {
 }
 
 MDnsConnectionImplFactory::~MDnsConnectionImplFactory() {
 }
 
 scoped_ptr<MDnsConnection> MDnsConnectionImplFactory::CreateConnection(
     MDnsConnection::Delegate* delegate) {
   return scoped_ptr<MDnsConnection>(new MDnsConnectionImpl(delegate));
 }
 
 }  // namespace net