mojo/services/network/udp_socket_apptest.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "base/macros.h"
   6 #include "base/memory/scoped_ptr.h"
   7 #include "base/run_loop.h"
   8 #include "mojo/application/application_test_base_chromium.h"
   9 #include "mojo/public/cpp/application/application_connection.h"
  10 #include "mojo/public/cpp/application/application_impl.h"
  11 #include "mojo/public/cpp/bindings/callback.h"
  12 #include "mojo/services/network/public/cpp/udp_socket_wrapper.h"
  13 #include "mojo/services/network/public/interfaces/network_service.mojom.h"
  14 #include "mojo/services/network/public/interfaces/udp_socket.mojom.h"
  15 #include "net/base/net_errors.h"
  16 #include "testing/gtest/include/gtest/gtest.h"
  17
  18 namespace mojo {
  19 namespace service {
  20 namespace {
  21
  22 NetAddressPtr GetLocalHostWithAnyPort() {
  23   NetAddressPtr addr(NetAddress::New());
  24   addr->family = NET_ADDRESS_FAMILY_IPV4;
  25   addr->ipv4 = NetAddressIPv4::New();
  26   addr->ipv4->port = 0;
  27   addr->ipv4->addr.resize(4);
  28   addr->ipv4->addr[0] = 127;
  29   addr->ipv4->addr[1] = 0;
  30   addr->ipv4->addr[2] = 0;
  31   addr->ipv4->addr[3] = 1;
  32
  33   return addr.Pass();
  34 }
  35
  36 Array<uint8_t> CreateTestMessage(uint8_t initial, size_t size) {
  37   Array<uint8_t> array(size);
  38   for (size_t i = 0; i < size; ++i)
  39     array[i] = static_cast<uint8_t>((i + initial) % 256);
  40   return array.Pass();
  41 }
  42
  43 template <typename CallbackType>
  44 class TestCallbackBase {
  45  public:
  46   TestCallbackBase() : state_(nullptr), run_loop_(nullptr), ran_(false) {}
  47
  48   ~TestCallbackBase() {
  49     state_->set_test_callback(nullptr);
  50   }
  51
  52   CallbackType callback() const { return callback_; }
  53
  54   void WaitForResult() {
  55     if (ran_)
  56       return;
  57
  58     base::RunLoop run_loop;
  59     run_loop_ = &run_loop;
  60     run_loop.Run();
  61     run_loop_ = nullptr;
  62   }
  63
  64  protected:
  65   struct StateBase : public CallbackType::Runnable {
  66     StateBase() : test_callback_(nullptr) {}
  67     ~StateBase() override {}
  68
  69     void set_test_callback(TestCallbackBase* test_callback) {
  70       test_callback_ = test_callback;
  71     }
  72
  73    protected:
  74     void NotifyRun() const {
  75       if (test_callback_) {
  76         test_callback_->ran_ = true;
  77         if (test_callback_->run_loop_)
  78           test_callback_->run_loop_->Quit();
  79       }
  80     }
  81
  82     TestCallbackBase* test_callback_;
  83
  84    private:
  85     DISALLOW_COPY_AND_ASSIGN(StateBase);
  86   };
  87
  88   // Takes ownership of |state|, and guarantees that it lives at least as long
  89   // as this object.
  90   void Initialize(StateBase* state) {
  91     state_ = state;
  92     state_->set_test_callback(this);
  93     callback_ = CallbackType(
  94         static_cast<typename CallbackType::Runnable*>(state_));
  95   }
  96
  97  private:
  98   // The lifespan is managed by |callback_| (and its copies).
  99   StateBase* state_;
 100   CallbackType callback_;
 101   base::RunLoop* run_loop_;
 102   bool ran_;
 103
 104   DISALLOW_COPY_AND_ASSIGN(TestCallbackBase);
 105 };
 106
 107 class TestCallback : public TestCallbackBase<Callback<void(NetworkErrorPtr)>> {
 108  public:
 109   TestCallback() {
 110     Initialize(new State());
 111   }
 112   ~TestCallback() {}
 113
 114   const NetworkErrorPtr& result() const { return result_; }
 115
 116  private:
 117   struct State: public StateBase {
 118     ~State() override {}
 119
 120     void Run(NetworkErrorPtr result) const override {
 121       if (test_callback_) {
 122         TestCallback* callback = static_cast<TestCallback*>(test_callback_);
 123         callback->result_ = result.Pass();
 124       }
 125       NotifyRun();
 126     }
 127   };
 128
 129   NetworkErrorPtr result_;
 130 };
 131
 132 class TestCallbackWithAddressAndReceiver
 133     : public TestCallbackBase<
 134           Callback<void(NetworkErrorPtr,
 135                         NetAddressPtr,
 136                         InterfaceRequest<UDPSocketReceiver>)>> {
 137  public:
 138   TestCallbackWithAddressAndReceiver() { Initialize(new State()); }
 139   ~TestCallbackWithAddressAndReceiver() {}
 140
 141   const NetworkErrorPtr& result() const { return result_; }
 142   const NetAddressPtr& net_address() const { return net_address_; }
 143   InterfaceRequest<UDPSocketReceiver>& receiver() { return receiver_; }
 144
 145  private:
 146   struct State : public StateBase {
 147     ~State() override {}
 148
 149     void Run(NetworkErrorPtr result,
 150              NetAddressPtr net_address,
 151              InterfaceRequest<UDPSocketReceiver> receiver) const override {
 152       if (test_callback_) {
 153         TestCallbackWithAddressAndReceiver* callback =
 154             static_cast<TestCallbackWithAddressAndReceiver*>(test_callback_);
 155         callback->result_ = result.Pass();
 156         callback->net_address_ = net_address.Pass();
 157         callback->receiver_ = receiver.Pass();
 158       }
 159       NotifyRun();
 160     }
 161   };
 162
 163   NetworkErrorPtr result_;
 164   NetAddressPtr net_address_;
 165   InterfaceRequest<UDPSocketReceiver> receiver_;
 166 };
 167
 168 class TestCallbackWithAddress
 169     : public TestCallbackBase<Callback<void(NetworkErrorPtr, NetAddressPtr)>> {
 170  public:
 171   TestCallbackWithAddress() {
 172     Initialize(new State());
 173   }
 174   ~TestCallbackWithAddress() {}
 175
 176   const NetworkErrorPtr& result() const { return result_; }
 177   const NetAddressPtr& net_address() const { return net_address_; }
 178
 179  private:
 180   struct State : public StateBase {
 181     ~State() override {}
 182
 183     void Run(NetworkErrorPtr result, NetAddressPtr net_address) const override {
 184       if (test_callback_) {
 185         TestCallbackWithAddress* callback =
 186             static_cast<TestCallbackWithAddress*>(test_callback_);
 187         callback->result_ = result.Pass();
 188         callback->net_address_ = net_address.Pass();
 189       }
 190       NotifyRun();
 191     }
 192   };
 193
 194   NetworkErrorPtr result_;
 195   NetAddressPtr net_address_;
 196 };
 197
 198 class TestCallbackWithUint32
 199     : public TestCallbackBase<Callback<void(uint32_t)>> {
 200  public:
 201   TestCallbackWithUint32() : result_(0) {
 202     Initialize(new State());
 203   }
 204   ~TestCallbackWithUint32() {}
 205
 206   uint32_t result() const { return result_; }
 207
 208  private:
 209   struct State : public StateBase {
 210     ~State() override {}
 211
 212     void Run(uint32_t result) const override {
 213       if (test_callback_) {
 214         TestCallbackWithUint32* callback =
 215             static_cast<TestCallbackWithUint32*>(test_callback_);
 216         callback->result_ = result;
 217       }
 218       NotifyRun();
 219     }
 220   };
 221
 222   uint32_t result_;
 223 };
 224
 225 class TestReceiveCallback
 226     : public TestCallbackBase<
 227         Callback<void(NetworkErrorPtr, NetAddressPtr, Array<uint8_t>)>> {
 228  public:
 229   TestReceiveCallback() {
 230     Initialize(new State());
 231   }
 232   ~TestReceiveCallback() {}
 233
 234   const NetworkErrorPtr& result() const { return result_; }
 235   const NetAddressPtr& src_addr() const { return src_addr_; }
 236   const Array<uint8_t>& data() const { return data_; }
 237
 238  private:
 239   struct State : public StateBase {
 240     ~State() override {}
 241
 242     void Run(NetworkErrorPtr result,
 243              NetAddressPtr src_addr,
 244              Array<uint8_t> data) const override {
 245       if (test_callback_) {
 246         TestReceiveCallback* callback =
 247             static_cast<TestReceiveCallback*>(test_callback_);
 248         callback->result_ = result.Pass();
 249         callback->src_addr_ = src_addr.Pass();
 250         callback->data_ = data.Pass();
 251       }
 252       NotifyRun();
 253     }
 254   };
 255
 256   NetworkErrorPtr result_;
 257   NetAddressPtr src_addr_;
 258   Array<uint8_t> data_;
 259 };
 260
 261 struct ReceiveResult {
 262   NetworkErrorPtr result;
 263   NetAddressPtr addr;
 264   Array<uint8_t> data;
 265 };
 266
 267 class UDPSocketReceiverImpl : public UDPSocketReceiver {
 268  public:
 269   UDPSocketReceiverImpl() : run_loop_(nullptr), expected_receive_count_(0) {}
 270
 271   ~UDPSocketReceiverImpl() override {
 272     while (!results_.empty()) {
 273       delete results_.front();
 274       results_.pop();
 275     }
 276   }
 277
 278   std::queue<ReceiveResult*>* results() {
 279     return &results_;
 280   }
 281
 282   void WaitForReceiveResults(size_t count) {
 283     if (results_.size() == count)
 284       return;
 285
 286     expected_receive_count_ = count;
 287     base::RunLoop run_loop;
 288     run_loop_ = &run_loop;
 289     run_loop.Run();
 290     run_loop_ = nullptr;
 291   }
 292
 293  private:
 294   void OnReceived(NetworkErrorPtr result,
 295                   NetAddressPtr src_addr,
 296                   Array<uint8_t> data) override {
 297     ReceiveResult* entry = new ReceiveResult();
 298     entry->result = result.Pass();
 299     entry->addr = src_addr.Pass();
 300     entry->data = data.Pass();
 301
 302     results_.push(entry);
 303
 304     if (results_.size() == expected_receive_count_ && run_loop_) {
 305       expected_receive_count_ = 0;
 306       run_loop_->Quit();
 307     }
 308   }
 309
 310   base::RunLoop* run_loop_;
 311   std::queue<ReceiveResult*> results_;
 312   size_t expected_receive_count_;
 313
 314   DISALLOW_COPY_AND_ASSIGN(UDPSocketReceiverImpl);
 315 };
 316
 317 class UDPSocketAppTest : public test::ApplicationTestBase {
 318  public:
 319   UDPSocketAppTest() : receiver_binding_(&receiver_) {}
 320   ~UDPSocketAppTest() override {}
 321
 322   void SetUp() override {
 323     ApplicationTestBase::SetUp();
 324
 325     ApplicationConnection* connection =
 326         application_impl()->ConnectToApplication("mojo:network_service");
 327     connection->ConnectToService(&network_service_);
 328
 329     network_service_->CreateUDPSocket(GetProxy(&socket_));
 330   }
 331
 332  protected:
 333   NetworkServicePtr network_service_;
 334   UDPSocketPtr socket_;
 335   UDPSocketReceiverImpl receiver_;
 336   Binding<UDPSocketReceiver> receiver_binding_;
 337
 338   DISALLOW_COPY_AND_ASSIGN(UDPSocketAppTest);
 339 };
 340
 341 }  // namespace
 342
 343 TEST_F(UDPSocketAppTest, Settings) {
 344   TestCallback callback1;
 345   socket_->AllowAddressReuse(callback1.callback());
 346   callback1.WaitForResult();
 347   EXPECT_EQ(net::OK, callback1.result()->code);
 348
 349   // Should fail because the socket hasn't been bound.
 350   TestCallback callback2;
 351   socket_->SetSendBufferSize(1024, callback2.callback());
 352   callback2.WaitForResult();
 353   EXPECT_NE(net::OK, callback2.result()->code);
 354
 355   // Should fail because the socket hasn't been bound.
 356   TestCallback callback3;
 357   socket_->SetReceiveBufferSize(2048, callback3.callback());
 358   callback3.WaitForResult();
 359   EXPECT_NE(net::OK, callback3.result()->code);
 360
 361   TestCallbackWithAddressAndReceiver callback4;
 362   socket_->Bind(GetLocalHostWithAnyPort(), callback4.callback());
 363   callback4.WaitForResult();
 364   EXPECT_EQ(net::OK, callback4.result()->code);
 365   EXPECT_NE(0u, callback4.net_address()->ipv4->port);
 366
 367   // Should fail because the socket has been bound.
 368   TestCallback callback5;
 369   socket_->AllowAddressReuse(callback5.callback());
 370   callback5.WaitForResult();
 371   EXPECT_NE(net::OK, callback5.result()->code);
 372
 373   TestCallback callback6;
 374   socket_->SetSendBufferSize(1024, callback6.callback());
 375   callback6.WaitForResult();
 376   EXPECT_EQ(net::OK, callback6.result()->code);
 377
 378   TestCallback callback7;
 379   socket_->SetReceiveBufferSize(2048, callback7.callback());
 380   callback7.WaitForResult();
 381   EXPECT_EQ(net::OK, callback7.result()->code);
 382
 383   TestCallbackWithUint32 callback8;
 384   socket_->NegotiateMaxPendingSendRequests(0, callback8.callback());
 385   callback8.WaitForResult();
 386   EXPECT_GT(callback8.result(), 0u);
 387
 388   TestCallbackWithUint32 callback9;
 389   socket_->NegotiateMaxPendingSendRequests(16, callback9.callback());
 390   callback9.WaitForResult();
 391   EXPECT_GT(callback9.result(), 0u);
 392 }
 393
 394 TEST_F(UDPSocketAppTest, TestReadWrite) {
 395   TestCallbackWithAddressAndReceiver callback1;
 396   socket_->Bind(GetLocalHostWithAnyPort(), callback1.callback());
 397   callback1.WaitForResult();
 398   ASSERT_EQ(net::OK, callback1.result()->code);
 399   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 400
 401   receiver_binding_.Bind(callback1.receiver().Pass());
 402
 403   NetAddressPtr server_addr = callback1.net_address().Clone();
 404
 405   UDPSocketPtr client_socket;
 406   network_service_->CreateUDPSocket(GetProxy(&client_socket));
 407
 408   TestCallbackWithAddressAndReceiver callback2;
 409   client_socket->Bind(GetLocalHostWithAnyPort(), callback2.callback());
 410   callback2.WaitForResult();
 411   ASSERT_EQ(net::OK, callback2.result()->code);
 412   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 413
 414   NetAddressPtr client_addr = callback2.net_address().Clone();
 415
 416   const size_t kDatagramCount = 6;
 417   const size_t kDatagramSize = 255;
 418   socket_->ReceiveMore(kDatagramCount);
 419
 420   for (size_t i = 0; i < kDatagramCount; ++i) {
 421     TestCallback callback;
 422     client_socket->SendTo(
 423         server_addr.Clone(),
 424         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize),
 425         callback.callback());
 426     callback.WaitForResult();
 427     EXPECT_EQ(255, callback.result()->code);
 428   }
 429
 430   receiver_.WaitForReceiveResults(kDatagramCount);
 431   for (size_t i = 0; i < kDatagramCount; ++i) {
 432     scoped_ptr<ReceiveResult> result(receiver_.results()->front());
 433     receiver_.results()->pop();
 434
 435     EXPECT_EQ(static_cast<int>(kDatagramSize), result->result->code);
 436     EXPECT_TRUE(result->addr.Equals(client_addr));
 437     EXPECT_TRUE(result->data.Equals(
 438         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize)));
 439   }
 440 }
 441
 442 TEST_F(UDPSocketAppTest, TestConnectedReadWrite) {
 443   TestCallbackWithAddressAndReceiver callback1;
 444   socket_->Bind(GetLocalHostWithAnyPort(), callback1.callback());
 445   callback1.WaitForResult();
 446   ASSERT_EQ(net::OK, callback1.result()->code);
 447   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 448
 449   receiver_binding_.Bind(callback1.receiver().Pass());
 450
 451   NetAddressPtr server_addr = callback1.net_address().Clone();
 452
 453   UDPSocketPtr client_socket;
 454   network_service_->CreateUDPSocket(GetProxy(&client_socket));
 455
 456   TestCallbackWithAddressAndReceiver callback2;
 457   client_socket->Connect(server_addr.Clone(), callback2.callback());
 458   callback2.WaitForResult();
 459   ASSERT_EQ(net::OK, callback2.result()->code);
 460   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 461
 462   UDPSocketReceiverImpl client_socket_receiver;
 463   Binding<UDPSocketReceiver> client_receiver_binding(
 464       &client_socket_receiver, callback2.receiver().Pass());
 465
 466   NetAddressPtr client_addr = callback2.net_address().Clone();
 467
 468   const size_t kDatagramCount = 6;
 469   const size_t kDatagramSize = 255;
 470
 471   // Test send using a connected socket.
 472   socket_->ReceiveMore(kDatagramCount);
 473
 474   for (size_t i = 0; i < kDatagramCount; ++i) {
 475     TestCallback callback;
 476     client_socket->SendTo(
 477         nullptr,
 478         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize),
 479         callback.callback());
 480     callback.WaitForResult();
 481     EXPECT_EQ(255, callback.result()->code);
 482   }
 483
 484   receiver_.WaitForReceiveResults(kDatagramCount);
 485   for (size_t i = 0; i < kDatagramCount; ++i) {
 486     scoped_ptr<ReceiveResult> result(receiver_.results()->front());
 487     receiver_.results()->pop();
 488
 489     EXPECT_EQ(static_cast<int>(kDatagramSize), result->result->code);
 490     EXPECT_TRUE(result->addr.Equals(client_addr));
 491     EXPECT_TRUE(result->data.Equals(
 492         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize)));
 493   }
 494
 495   // Test receive using a connected socket.
 496   client_socket->ReceiveMore(kDatagramCount);
 497
 498   for (size_t i = 0; i < kDatagramCount; ++i) {
 499     TestCallback callback;
 500     socket_->SendTo(
 501         client_addr.Clone(),
 502         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize),
 503         callback.callback());
 504     callback.WaitForResult();
 505     EXPECT_EQ(255, callback.result()->code);
 506   }
 507
 508   client_socket_receiver.WaitForReceiveResults(kDatagramCount);
 509   for (size_t i = 0; i < kDatagramCount; ++i) {
 510     scoped_ptr<ReceiveResult> result(client_socket_receiver.results()->front());
 511     client_socket_receiver.results()->pop();
 512
 513     EXPECT_EQ(static_cast<int>(kDatagramSize), result->result->code);
 514     EXPECT_FALSE(result->addr);
 515     EXPECT_TRUE(result->data.Equals(
 516         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize)));
 517   }
 518 }
 519
 520 TEST_F(UDPSocketAppTest, TestWrapperReadWrite) {
 521   UDPSocketWrapper socket(socket_.Pass(), 4, 4);
 522
 523   TestCallbackWithAddress callback1;
 524   socket.Bind(GetLocalHostWithAnyPort(), callback1.callback());
 525   callback1.WaitForResult();
 526   ASSERT_EQ(net::OK, callback1.result()->code);
 527   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 528
 529   NetAddressPtr server_addr = callback1.net_address().Clone();
 530
 531   UDPSocketPtr raw_client_socket;
 532   network_service_->CreateUDPSocket(GetProxy(&raw_client_socket));
 533   UDPSocketWrapper client_socket(raw_client_socket.Pass(), 4, 4);
 534
 535   TestCallbackWithAddress callback2;
 536   client_socket.Bind(GetLocalHostWithAnyPort(), callback2.callback());
 537   callback2.WaitForResult();
 538   ASSERT_EQ(net::OK, callback2.result()->code);
 539   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 540
 541   NetAddressPtr client_addr = callback2.net_address().Clone();
 542
 543   const size_t kDatagramCount = 16;
 544   const size_t kDatagramSize = 255;
 545
 546   for (size_t i = 1; i < kDatagramCount; ++i) {
 547     scoped_ptr<TestCallback[]> send_callbacks(new TestCallback[i]);
 548     scoped_ptr<TestReceiveCallback[]> receive_callbacks(
 549         new TestReceiveCallback[i]);
 550
 551     for (size_t j = 0; j < i; ++j) {
 552       client_socket.SendTo(
 553           server_addr.Clone(),
 554           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize),
 555           send_callbacks[j].callback());
 556
 557       socket.ReceiveFrom(receive_callbacks[j].callback());
 558     }
 559
 560     receive_callbacks[i - 1].WaitForResult();
 561
 562     for (size_t j = 0; j < i; ++j) {
 563       EXPECT_EQ(static_cast<int>(kDatagramSize),
 564                 receive_callbacks[j].result()->code);
 565       EXPECT_TRUE(receive_callbacks[j].src_addr().Equals(client_addr));
 566       EXPECT_TRUE(receive_callbacks[j].data().Equals(
 567           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize)));
 568     }
 569   }
 570 }
 571
 572 TEST_F(UDPSocketAppTest, TestWrapperConnectedReadWrite) {
 573   UDPSocketWrapper socket(socket_.Pass(), 4, 4);
 574
 575   TestCallbackWithAddress callback1;
 576   socket.Bind(GetLocalHostWithAnyPort(), callback1.callback());
 577   callback1.WaitForResult();
 578   ASSERT_EQ(net::OK, callback1.result()->code);
 579   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 580
 581   NetAddressPtr server_addr = callback1.net_address().Clone();
 582
 583   UDPSocketPtr raw_client_socket;
 584   network_service_->CreateUDPSocket(GetProxy(&raw_client_socket));
 585   UDPSocketWrapper client_socket(raw_client_socket.Pass(), 4, 4);
 586
 587   TestCallbackWithAddress callback2;
 588   client_socket.Connect(server_addr.Pass(), callback2.callback());
 589   callback2.WaitForResult();
 590   ASSERT_EQ(net::OK, callback2.result()->code);
 591   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 592
 593   NetAddressPtr client_addr = callback2.net_address().Clone();
 594
 595   const size_t kDatagramCount = 16;
 596   const size_t kDatagramSize = 255;
 597
 598   // Test send using a connected socket.
 599   for (size_t i = 1; i < kDatagramCount; ++i) {
 600     scoped_ptr<TestCallback[]> send_callbacks(new TestCallback[i]);
 601     scoped_ptr<TestReceiveCallback[]> receive_callbacks(
 602         new TestReceiveCallback[i]);
 603
 604     for (size_t j = 0; j < i; ++j) {
 605       client_socket.SendTo(
 606           nullptr,
 607           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize),
 608           send_callbacks[j].callback());
 609
 610       socket.ReceiveFrom(receive_callbacks[j].callback());
 611     }
 612
 613     receive_callbacks[i - 1].WaitForResult();
 614
 615     for (size_t j = 0; j < i; ++j) {
 616       EXPECT_EQ(static_cast<int>(kDatagramSize),
 617                 receive_callbacks[j].result()->code);
 618       EXPECT_TRUE(receive_callbacks[j].src_addr().Equals(client_addr));
 619       EXPECT_TRUE(receive_callbacks[j].data().Equals(
 620           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize)));
 621     }
 622   }
 623
 624   // Test receive using a connected socket.
 625   for (size_t i = 1; i < kDatagramCount; ++i) {
 626     scoped_ptr<TestCallback[]> send_callbacks(new TestCallback[i]);
 627     scoped_ptr<TestReceiveCallback[]> receive_callbacks(
 628         new TestReceiveCallback[i]);
 629
 630     for (size_t j = 0; j < i; ++j) {
 631       socket.SendTo(
 632           client_addr.Clone(),
 633           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize),
 634           send_callbacks[j].callback());
 635
 636       client_socket.ReceiveFrom(receive_callbacks[j].callback());
 637     }
 638
 639     receive_callbacks[i - 1].WaitForResult();
 640
 641     for (size_t j = 0; j < i; ++j) {
 642       EXPECT_EQ(static_cast<int>(kDatagramSize),
 643                 receive_callbacks[j].result()->code);
 644       EXPECT_FALSE(receive_callbacks[j].src_addr());
 645       EXPECT_TRUE(receive_callbacks[j].data().Equals(
 646           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize)));
 647     }
 648   }
 649 }
 650
 651 }  // namespace service
 652 }  // namespace mojo