Separate StunProber::Start into Prepare and Run so we could create multiple of them and send out ST…

webrtc/p2p/stunprober/stunprober.cc

 /*
  *  Copyright 2015 The WebRTC Project Authors. All rights reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
@@ skipping 10 matching lines @@
 #include "webrtc/base/timeutils.h"
 #include "webrtc/base/thread.h"
 #include "webrtc/p2p/base/packetsocketfactory.h"
 #include "webrtc/p2p/base/stun.h"
 #include "webrtc/p2p/stunprober/stunprober.h"
 
 namespace stunprober {
 
 namespace {
 
-const int thread_wake_up_interval_ms = 5;
+const int THREAD_WAKE_UP_INTERVAL_MS = 5;
 
 template <typename T>
 void IncrementCounterByAddress(std::map<T, int>* counter_per_ip, const T& ip) {
   counter_per_ip->insert(std::make_pair(ip, 0)).first->second++;
 }
 
 }  // namespace
 
 // A requester tracks the requests and responses from a single socket to many
 // STUN servers
@@ skipping 94 matching lines @@
   cricket::StunMessage message;
 
   // Random transaction ID, STUN_BINDING_REQUEST
   message.SetTransactionID(
       rtc::CreateRandomString(cricket::kStunTransactionIdLength));
   message.SetType(cricket::STUN_BINDING_REQUEST);
 
   rtc::scoped_ptr<rtc::ByteBuffer> request_packet(
       new rtc::ByteBuffer(nullptr, kMaxUdpBufferSize));
   if (!message.Write(request_packet.get())) {
-    prober_->End(WRITE_FAILED);
+    prober_->ReportOnFinished(WRITE_FAILED);
     return;
   }
 
   auto addr = server_ips_[num_request_sent_];
   request.server_addr = addr.ipaddr();
 
   // The write must succeed immediately. Otherwise, the calculating of the STUN
   // request timing could become too complicated. Callback is ignored by passing
   // empty AsyncCallback.
   rtc::PacketOptions options;
   int rv = socket_->SendTo(const_cast<char*>(request_packet->Data()),
                            request_packet->Length(), addr, options);
   if (rv < 0) {
-    prober_->End(WRITE_FAILED);
+    prober_->ReportOnFinished(WRITE_FAILED);
     return;
   }
 
   request.sent_time_ms = rtc::Time();
 
   num_request_sent_++;
   RTC_DCHECK(static_cast<size_t>(num_request_sent_) <= server_ips_.size());
 }
 
 void StunProber::Requester::Request::ProcessResponse(const char* buf,
@@ skipping 29 matching lines @@
     rtc::AsyncPacketSocket* socket,
     const char* buf,
     size_t size,
     const rtc::SocketAddress& addr,
     const rtc::PacketTime& time) {
   RTC_DCHECK(thread_checker_.CalledOnValidThread());
   RTC_DCHECK(socket_);
   Request* request = GetRequestByAddress(addr.ipaddr());
   if (!request) {
     // Something is wrong, finish the test.
-    prober_->End(GENERIC_FAILURE);
+    prober_->ReportOnFinished(GENERIC_FAILURE);
     return;
   }
 
   num_response_received_++;
   request->ProcessResponse(buf, size);
 }
 
 StunProber::Requester::Request* StunProber::Requester::GetRequestByAddress(
     const rtc::IPAddress& ipaddr) {
   RTC_DCHECK(thread_checker_.CalledOnValidThread());
@@ skipping 27 matching lines @@
     }
   }
 }
 
 bool StunProber::Start(const std::vector<rtc::SocketAddress>& servers,
                        bool shared_socket_mode,
                        int interval_ms,
                        int num_request_per_ip,
                        int timeout_ms,
                        const AsyncCallback callback) {
+  observer_adapter_.set_callback(callback);
+  return Prepare(servers, shared_socket_mode, interval_ms, num_request_per_ip,
+                 timeout_ms, &observer_adapter_);
+}
+
+bool StunProber::Prepare(const std::vector<rtc::SocketAddress>& servers,
+                         bool shared_socket_mode,
+                         int interval_ms,
+                         int num_request_per_ip,
+                         int timeout_ms,
+                         StunProber::Observer* observer) {
   RTC_DCHECK(thread_checker_.CalledOnValidThread());
   interval_ms_ = interval_ms;
   shared_socket_mode_ = shared_socket_mode;
 
   requests_per_ip_ = num_request_per_ip;
   if (requests_per_ip_ == 0 || servers.size() == 0) {
     return false;
   }
 
   timeout_ms_ = timeout_ms;
   servers_ = servers;
-  finished_callback_ = callback;
+  observer_ = observer;
   return ResolveServerName(servers_.back());
 }
 
+bool StunProber::Run(StunProber::Observer* observer) {
+  observer_ = observer;
+  if (total_ready_sockets_ != total_socket_required()) {
+    return false;
+  }
+  MaybeScheduleStunRequests();
+  return true;
+}
+
 bool StunProber::ResolveServerName(const rtc::SocketAddress& addr) {
   rtc::AsyncResolverInterface* resolver =
       socket_factory_->CreateAsyncResolver();
   if (!resolver) {
     return false;
   }
   resolver->SignalDone.connect(this, &StunProber::OnServerResolved);
   resolver->Start(addr);
   return true;
 }
 
 void StunProber::OnSocketReady(rtc::AsyncPacketSocket* socket,
                                const rtc::SocketAddress& addr) {
   total_ready_sockets_++;
   if (total_ready_sockets_ == total_socket_required()) {
-    MaybeScheduleStunRequests();
+    ReportOnPrepared(SUCCESS);
   }
 }
 
 void StunProber::OnServerResolved(rtc::AsyncResolverInterface* resolver) {
   RTC_DCHECK(thread_checker_.CalledOnValidThread());
 
   if (resolver->GetError() == 0) {
     rtc::SocketAddress addr(resolver->address().ipaddr(),
                             resolver->address().port());
     all_servers_addrs_.push_back(addr);
   }
 
   // Deletion of AsyncResolverInterface can't be done in OnResolveResult which
   // handles SignalDone.
   invoker_.AsyncInvoke<void>(
       thread_,
       rtc::Bind(&rtc::AsyncResolverInterface::Destroy, resolver, false));
   servers_.pop_back();
 
   if (servers_.size()) {
     if (!ResolveServerName(servers_.back())) {
-      End(RESOLVE_FAILED);
+      ReportOnPrepared(RESOLVE_FAILED);
     }
     return;
   }
 
   if (all_servers_addrs_.size() == 0) {
-    End(RESOLVE_FAILED);
+    ReportOnPrepared(RESOLVE_FAILED);
     return;
   }
 
   // Dedupe.
   std::set<rtc::SocketAddress> addrs(all_servers_addrs_.begin(),
                                      all_servers_addrs_.end());
   all_servers_addrs_.assign(addrs.begin(), addrs.end());
 
   // Prepare all the sockets beforehand. All of them will bind to "any" address.
   while (sockets_.size() < total_socket_required()) {
     rtc::scoped_ptr<rtc::AsyncPacketSocket> socket(
         socket_factory_->CreateUdpSocket(rtc::SocketAddress(INADDR_ANY, 0), 0,
                                          0));
     if (!socket) {
-      End(GENERIC_FAILURE);
+      ReportOnPrepared(GENERIC_FAILURE);
       return;
     }
     // Chrome and WebRTC behave differently in terms of the state of a socket
     // once returned from PacketSocketFactory::CreateUdpSocket.
     if (socket->GetState() == rtc::AsyncPacketSocket::STATE_BINDING) {
       socket->SignalAddressReady.connect(this, &StunProber::OnSocketReady);
     } else {
       OnSocketReady(socket.get(), rtc::SocketAddress(INADDR_ANY, 0));
     }
     sockets_.push_back(socket.release());
@@ skipping 25 matching lines @@
     requesters_.push_back(current_requester_);
   }
   if (!current_requester_) {
     return false;
   }
   current_requester_->SendStunRequest();
   num_request_sent_++;
   return true;
 }
 
+bool StunProber::should_send_next_request(uint32_t now) {
+  if (interval_ms_ < THREAD_WAKE_UP_INTERVAL_MS) {
+    return now >= next_request_time_ms_;
+  } else {
+    return (now + (THREAD_WAKE_UP_INTERVAL_MS / 2)) >= next_request_time_ms_;
+  }
+}
+
+int StunProber::get_wake_up_interval_ms() {
+  if (interval_ms_ < THREAD_WAKE_UP_INTERVAL_MS) {
+    return 1;
+  } else {
+    return THREAD_WAKE_UP_INTERVAL_MS;
+  }
+}
+
 void StunProber::MaybeScheduleStunRequests() {
   RTC_DCHECK(thread_checker_.CalledOnValidThread());
   uint32_t now = rtc::Time();
 
   if (Done()) {
     invoker_.AsyncInvokeDelayed<void>(
-        thread_, rtc::Bind(&StunProber::End, this, SUCCESS), timeout_ms_);
+        thread_, rtc::Bind(&StunProber::ReportOnFinished, this, SUCCESS),
+        timeout_ms_);
     return;
   }
-  if ((now + (thread_wake_up_interval_ms / 2)) >= next_request_time_ms_) {
+  if (should_send_next_request(now)) {
     if (!SendNextRequest()) {
-      End(GENERIC_FAILURE);
+      ReportOnFinished(GENERIC_FAILURE);
       return;
     }
     next_request_time_ms_ = now + interval_ms_;
   }
   invoker_.AsyncInvokeDelayed<void>(
       thread_, rtc::Bind(&StunProber::MaybeScheduleStunRequests, this),
-      thread_wake_up_interval_ms /* ms */);
+      get_wake_up_interval_ms());
 }
 
 bool StunProber::GetStats(StunProber::Stats* prob_stats) const {
   // No need to be on the same thread.
   if (!prob_stats) {
     return false;
   }
 
   StunProber::Stats stats;
 
@@ skipping 107 matching lines @@
   }
 
   if (num_received) {
     stats.average_rtt_ms = static_cast<int>((rtt_sum / num_received));
   }
 
   *prob_stats = stats;
   return true;
 }
 
-void StunProber::End(StunProber::Status status) {
-  RTC_DCHECK(thread_checker_.CalledOnValidThread());
-  if (!finished_callback_.empty()) {
-    AsyncCallback callback = finished_callback_;
-    finished_callback_ = AsyncCallback();
-
-    // Callback at the last since the prober might be deleted in the callback.
-    callback(this, status);
+void StunProber::ReportOnPrepared(StunProber::Status status) {
+  if (observer_) {
+    observer_->OnPrepared(this, status);
   }
 }
 
+void StunProber::ReportOnFinished(StunProber::Status status) {
+  if (observer_) {
+    observer_->OnFinished(this, status);
+  }
+}
+
 }  // namespace stunprober