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/public/cpp/application_connection.h"
   9 #include "mojo/application/public/cpp/application_impl.h"
  10 #include "mojo/application/public/cpp/application_test_base.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     mojo::URLRequestPtr request(mojo::URLRequest::New());
 326     request->url = mojo::String::From("mojo:network_service");
 327     scoped_ptr<ApplicationConnection> connection =
 328         application_impl()->ConnectToApplication(request.Pass());
 329     connection->ConnectToService(&network_service_);
 330
 331     network_service_->CreateUDPSocket(GetProxy(&socket_));
 332   }
 333
 334  protected:
 335   NetworkServicePtr network_service_;
 336   UDPSocketPtr socket_;
 337   UDPSocketReceiverImpl receiver_;
 338   Binding<UDPSocketReceiver> receiver_binding_;
 339
 340   DISALLOW_COPY_AND_ASSIGN(UDPSocketAppTest);
 341 };
 342
 343 }  // namespace
 344
 345 TEST_F(UDPSocketAppTest, Settings) {
 346   TestCallback callback1;
 347   socket_->AllowAddressReuse(callback1.callback());
 348   callback1.WaitForResult();
 349   EXPECT_EQ(net::OK, callback1.result()->code);
 350
 351   // Should fail because the socket hasn't been bound.
 352   TestCallback callback2;
 353   socket_->SetSendBufferSize(1024, callback2.callback());
 354   callback2.WaitForResult();
 355   EXPECT_NE(net::OK, callback2.result()->code);
 356
 357   // Should fail because the socket hasn't been bound.
 358   TestCallback callback3;
 359   socket_->SetReceiveBufferSize(2048, callback3.callback());
 360   callback3.WaitForResult();
 361   EXPECT_NE(net::OK, callback3.result()->code);
 362
 363   TestCallbackWithAddressAndReceiver callback4;
 364   socket_->Bind(GetLocalHostWithAnyPort(), callback4.callback());
 365   callback4.WaitForResult();
 366   EXPECT_EQ(net::OK, callback4.result()->code);
 367   EXPECT_NE(0u, callback4.net_address()->ipv4->port);
 368
 369   // Should fail because the socket has been bound.
 370   TestCallback callback5;
 371   socket_->AllowAddressReuse(callback5.callback());
 372   callback5.WaitForResult();
 373   EXPECT_NE(net::OK, callback5.result()->code);
 374
 375   TestCallback callback6;
 376   socket_->SetSendBufferSize(1024, callback6.callback());
 377   callback6.WaitForResult();
 378   EXPECT_EQ(net::OK, callback6.result()->code);
 379
 380   TestCallback callback7;
 381   socket_->SetReceiveBufferSize(2048, callback7.callback());
 382   callback7.WaitForResult();
 383   EXPECT_EQ(net::OK, callback7.result()->code);
 384
 385   TestCallbackWithUint32 callback8;
 386   socket_->NegotiateMaxPendingSendRequests(0, callback8.callback());
 387   callback8.WaitForResult();
 388   EXPECT_GT(callback8.result(), 0u);
 389
 390   TestCallbackWithUint32 callback9;
 391   socket_->NegotiateMaxPendingSendRequests(16, callback9.callback());
 392   callback9.WaitForResult();
 393   EXPECT_GT(callback9.result(), 0u);
 394 }
 395
 396 TEST_F(UDPSocketAppTest, TestReadWrite) {
 397   TestCallbackWithAddressAndReceiver callback1;
 398   socket_->Bind(GetLocalHostWithAnyPort(), callback1.callback());
 399   callback1.WaitForResult();
 400   ASSERT_EQ(net::OK, callback1.result()->code);
 401   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 402
 403   receiver_binding_.Bind(callback1.receiver().Pass());
 404
 405   NetAddressPtr server_addr = callback1.net_address().Clone();
 406
 407   UDPSocketPtr client_socket;
 408   network_service_->CreateUDPSocket(GetProxy(&client_socket));
 409
 410   TestCallbackWithAddressAndReceiver callback2;
 411   client_socket->Bind(GetLocalHostWithAnyPort(), callback2.callback());
 412   callback2.WaitForResult();
 413   ASSERT_EQ(net::OK, callback2.result()->code);
 414   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 415
 416   NetAddressPtr client_addr = callback2.net_address().Clone();
 417
 418   const size_t kDatagramCount = 6;
 419   const size_t kDatagramSize = 255;
 420   socket_->ReceiveMore(kDatagramCount);
 421
 422   for (size_t i = 0; i < kDatagramCount; ++i) {
 423     TestCallback callback;
 424     client_socket->SendTo(
 425         server_addr.Clone(),
 426         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize),
 427         callback.callback());
 428     callback.WaitForResult();
 429     EXPECT_EQ(255, callback.result()->code);
 430   }
 431
 432   receiver_.WaitForReceiveResults(kDatagramCount);
 433   for (size_t i = 0; i < kDatagramCount; ++i) {
 434     scoped_ptr<ReceiveResult> result(receiver_.results()->front());
 435     receiver_.results()->pop();
 436
 437     EXPECT_EQ(static_cast<int>(kDatagramSize), result->result->code);
 438     EXPECT_TRUE(result->addr.Equals(client_addr));
 439     EXPECT_TRUE(result->data.Equals(
 440         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize)));
 441   }
 442 }
 443
 444 TEST_F(UDPSocketAppTest, TestConnectedReadWrite) {
 445   TestCallbackWithAddressAndReceiver callback1;
 446   socket_->Bind(GetLocalHostWithAnyPort(), callback1.callback());
 447   callback1.WaitForResult();
 448   ASSERT_EQ(net::OK, callback1.result()->code);
 449   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 450
 451   receiver_binding_.Bind(callback1.receiver().Pass());
 452
 453   NetAddressPtr server_addr = callback1.net_address().Clone();
 454
 455   UDPSocketPtr client_socket;
 456   network_service_->CreateUDPSocket(GetProxy(&client_socket));
 457
 458   TestCallbackWithAddressAndReceiver callback2;
 459   client_socket->Connect(server_addr.Clone(), callback2.callback());
 460   callback2.WaitForResult();
 461   ASSERT_EQ(net::OK, callback2.result()->code);
 462   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 463
 464   UDPSocketReceiverImpl client_socket_receiver;
 465   Binding<UDPSocketReceiver> client_receiver_binding(
 466       &client_socket_receiver, callback2.receiver().Pass());
 467
 468   NetAddressPtr client_addr = callback2.net_address().Clone();
 469
 470   const size_t kDatagramCount = 6;
 471   const size_t kDatagramSize = 255;
 472
 473   // Test send using a connected socket.
 474   socket_->ReceiveMore(kDatagramCount);
 475
 476   for (size_t i = 0; i < kDatagramCount; ++i) {
 477     TestCallback callback;
 478     client_socket->SendTo(
 479         nullptr,
 480         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize),
 481         callback.callback());
 482     callback.WaitForResult();
 483     EXPECT_EQ(255, callback.result()->code);
 484   }
 485
 486   receiver_.WaitForReceiveResults(kDatagramCount);
 487   for (size_t i = 0; i < kDatagramCount; ++i) {
 488     scoped_ptr<ReceiveResult> result(receiver_.results()->front());
 489     receiver_.results()->pop();
 490
 491     EXPECT_EQ(static_cast<int>(kDatagramSize), result->result->code);
 492     EXPECT_TRUE(result->addr.Equals(client_addr));
 493     EXPECT_TRUE(result->data.Equals(
 494         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize)));
 495   }
 496
 497   // Test receive using a connected socket.
 498   client_socket->ReceiveMore(kDatagramCount);
 499
 500   for (size_t i = 0; i < kDatagramCount; ++i) {
 501     TestCallback callback;
 502     socket_->SendTo(
 503         client_addr.Clone(),
 504         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize),
 505         callback.callback());
 506     callback.WaitForResult();
 507     EXPECT_EQ(255, callback.result()->code);
 508   }
 509
 510   client_socket_receiver.WaitForReceiveResults(kDatagramCount);
 511   for (size_t i = 0; i < kDatagramCount; ++i) {
 512     scoped_ptr<ReceiveResult> result(client_socket_receiver.results()->front());
 513     client_socket_receiver.results()->pop();
 514
 515     EXPECT_EQ(static_cast<int>(kDatagramSize), result->result->code);
 516     EXPECT_FALSE(result->addr);
 517     EXPECT_TRUE(result->data.Equals(
 518         CreateTestMessage(static_cast<uint8_t>(i), kDatagramSize)));
 519   }
 520 }
 521
 522 TEST_F(UDPSocketAppTest, TestWrapperReadWrite) {
 523   UDPSocketWrapper socket(socket_.Pass(), 4, 4);
 524
 525   TestCallbackWithAddress callback1;
 526   socket.Bind(GetLocalHostWithAnyPort(), callback1.callback());
 527   callback1.WaitForResult();
 528   ASSERT_EQ(net::OK, callback1.result()->code);
 529   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 530
 531   NetAddressPtr server_addr = callback1.net_address().Clone();
 532
 533   UDPSocketPtr raw_client_socket;
 534   network_service_->CreateUDPSocket(GetProxy(&raw_client_socket));
 535   UDPSocketWrapper client_socket(raw_client_socket.Pass(), 4, 4);
 536
 537   TestCallbackWithAddress callback2;
 538   client_socket.Bind(GetLocalHostWithAnyPort(), callback2.callback());
 539   callback2.WaitForResult();
 540   ASSERT_EQ(net::OK, callback2.result()->code);
 541   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 542
 543   NetAddressPtr client_addr = callback2.net_address().Clone();
 544
 545   const size_t kDatagramCount = 16;
 546   const size_t kDatagramSize = 255;
 547
 548   for (size_t i = 1; i < kDatagramCount; ++i) {
 549     scoped_ptr<TestCallback[]> send_callbacks(new TestCallback[i]);
 550     scoped_ptr<TestReceiveCallback[]> receive_callbacks(
 551         new TestReceiveCallback[i]);
 552
 553     for (size_t j = 0; j < i; ++j) {
 554       client_socket.SendTo(
 555           server_addr.Clone(),
 556           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize),
 557           send_callbacks[j].callback());
 558
 559       socket.ReceiveFrom(receive_callbacks[j].callback());
 560     }
 561
 562     receive_callbacks[i - 1].WaitForResult();
 563
 564     for (size_t j = 0; j < i; ++j) {
 565       EXPECT_EQ(static_cast<int>(kDatagramSize),
 566                 receive_callbacks[j].result()->code);
 567       EXPECT_TRUE(receive_callbacks[j].src_addr().Equals(client_addr));
 568       EXPECT_TRUE(receive_callbacks[j].data().Equals(
 569           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize)));
 570     }
 571   }
 572 }
 573
 574 TEST_F(UDPSocketAppTest, TestWrapperConnectedReadWrite) {
 575   UDPSocketWrapper socket(socket_.Pass(), 4, 4);
 576
 577   TestCallbackWithAddress callback1;
 578   socket.Bind(GetLocalHostWithAnyPort(), callback1.callback());
 579   callback1.WaitForResult();
 580   ASSERT_EQ(net::OK, callback1.result()->code);
 581   ASSERT_NE(0u, callback1.net_address()->ipv4->port);
 582
 583   NetAddressPtr server_addr = callback1.net_address().Clone();
 584
 585   UDPSocketPtr raw_client_socket;
 586   network_service_->CreateUDPSocket(GetProxy(&raw_client_socket));
 587   UDPSocketWrapper client_socket(raw_client_socket.Pass(), 4, 4);
 588
 589   TestCallbackWithAddress callback2;
 590   client_socket.Connect(server_addr.Pass(), callback2.callback());
 591   callback2.WaitForResult();
 592   ASSERT_EQ(net::OK, callback2.result()->code);
 593   ASSERT_NE(0u, callback2.net_address()->ipv4->port);
 594
 595   NetAddressPtr client_addr = callback2.net_address().Clone();
 596
 597   const size_t kDatagramCount = 16;
 598   const size_t kDatagramSize = 255;
 599
 600   // Test send using a connected socket.
 601   for (size_t i = 1; i < kDatagramCount; ++i) {
 602     scoped_ptr<TestCallback[]> send_callbacks(new TestCallback[i]);
 603     scoped_ptr<TestReceiveCallback[]> receive_callbacks(
 604         new TestReceiveCallback[i]);
 605
 606     for (size_t j = 0; j < i; ++j) {
 607       client_socket.SendTo(
 608           nullptr,
 609           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize),
 610           send_callbacks[j].callback());
 611
 612       socket.ReceiveFrom(receive_callbacks[j].callback());
 613     }
 614
 615     receive_callbacks[i - 1].WaitForResult();
 616
 617     for (size_t j = 0; j < i; ++j) {
 618       EXPECT_EQ(static_cast<int>(kDatagramSize),
 619                 receive_callbacks[j].result()->code);
 620       EXPECT_TRUE(receive_callbacks[j].src_addr().Equals(client_addr));
 621       EXPECT_TRUE(receive_callbacks[j].data().Equals(
 622           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize)));
 623     }
 624   }
 625
 626   // Test receive using a connected socket.
 627   for (size_t i = 1; i < kDatagramCount; ++i) {
 628     scoped_ptr<TestCallback[]> send_callbacks(new TestCallback[i]);
 629     scoped_ptr<TestReceiveCallback[]> receive_callbacks(
 630         new TestReceiveCallback[i]);
 631
 632     for (size_t j = 0; j < i; ++j) {
 633       socket.SendTo(
 634           client_addr.Clone(),
 635           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize),
 636           send_callbacks[j].callback());
 637
 638       client_socket.ReceiveFrom(receive_callbacks[j].callback());
 639     }
 640
 641     receive_callbacks[i - 1].WaitForResult();
 642
 643     for (size_t j = 0; j < i; ++j) {
 644       EXPECT_EQ(static_cast<int>(kDatagramSize),
 645                 receive_callbacks[j].result()->code);
 646       EXPECT_FALSE(receive_callbacks[j].src_addr());
 647       EXPECT_TRUE(receive_callbacks[j].data().Equals(
 648           CreateTestMessage(static_cast<uint8_t>(j), kDatagramSize)));
 649     }
 650   }
 651 }
 652
 653 }  // namespace service
 654 }  // namespace mojo