Make DataPipe own an impl.

mojo/edk/system/local_data_pipe_impl_unittest.cc

 // Copyright 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 "mojo/edk/system/local_data_pipe_impl.h"
 
 #include <string.h>
 
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #include "mojo/edk/system/data_pipe.h"
 #include "mojo/edk/system/waiter.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace mojo {
 namespace system {
 namespace {
 
 const uint32_t kSizeOfOptions =
     static_cast<uint32_t>(sizeof(MojoCreateDataPipeOptions));
 
 // Validate options.
 TEST(LocalDataPipeImplTest, Creation) {
   // Create using default options.
   {
     // Get default options.
     MojoCreateDataPipeOptions default_options = {0};
     EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                   NullUserPointer(), &default_options));
-    scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(default_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(default_options));
     dp->ProducerClose();
     dp->ConsumerClose();
   }
 
   // Create using non-default options.
   {
     const MojoCreateDataPipeOptions options = {
         kSizeOfOptions,                           // |struct_size|.
         MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
         1,                                        // |element_num_bytes|.
         1000                                      // |capacity_num_bytes|.
     };
     MojoCreateDataPipeOptions validated_options = {0};
     EXPECT_EQ(MOJO_RESULT_OK,
               DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
                                               &validated_options));
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     dp->ProducerClose();
     dp->ConsumerClose();
   }
   {
     const MojoCreateDataPipeOptions options = {
         kSizeOfOptions,                           // |struct_size|.
         MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
         4,                                        // |element_num_bytes|.
         4000                                      // |capacity_num_bytes|.
     };
     MojoCreateDataPipeOptions validated_options = {0};
     EXPECT_EQ(MOJO_RESULT_OK,
               DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
                                               &validated_options));
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     dp->ProducerClose();
     dp->ConsumerClose();
   }
   {
     const MojoCreateDataPipeOptions options = {
         kSizeOfOptions,                                  // |struct_size|.
         MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
         7,                                               // |element_num_bytes|.
         7000000  // |capacity_num_bytes|.
     };
     MojoCreateDataPipeOptions validated_options = {0};
     EXPECT_EQ(MOJO_RESULT_OK,
               DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
                                               &validated_options));
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     dp->ProducerClose();
     dp->ConsumerClose();
   }
   // Default capacity.
   {
     const MojoCreateDataPipeOptions options = {
         kSizeOfOptions,                                  // |struct_size|.
         MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
         100,                                             // |element_num_bytes|.
         0  // |capacity_num_bytes|.
     };
     MojoCreateDataPipeOptions validated_options = {0};
     EXPECT_EQ(MOJO_RESULT_OK,
               DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
                                               &validated_options));
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     dp->ProducerClose();
     dp->ConsumerClose();
   }
 }
 
 TEST(LocalDataPipeImplTest, SimpleReadWrite) {
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       1000 * sizeof(int32_t)                    // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   int32_t elements[10] = {0};
   uint32_t num_bytes = 0;
 
   // Try reading; nothing there yet.
   num_bytes = static_cast<uint32_t>(arraysize(elements) * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
             dp->ConsumerReadData(UserPointer<void>(elements),
                                  MakeUserPointer(&num_bytes), false, false));
 
@@ skipping 99 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       2 * sizeof(int32_t)                       // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
   Waiter waiter;
   uint32_t context = 0;
   HandleSignalsState hss;
 
   // Never readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_FAILED_PRECONDITION,
       dp->ProducerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 12, &hss));
@@ skipping 160 matching lines @@
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
   Waiter waiter;
   HandleSignalsState hss;
 
   // Check MOJO_HANDLE_SIGNAL_PEER_CLOSED on producer.
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     // Add a waiter.
     waiter.Init();
     ASSERT_EQ(MOJO_RESULT_OK,
               dp->ProducerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                                       12, nullptr));
 
     // Close the consumer.
     dp->ConsumerClose();
 
     // It should be signaled.
     uint32_t context = 0;
     EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(1000, &context));
     EXPECT_EQ(12u, context);
     hss = HandleSignalsState();
     dp->ProducerRemoveAwakable(&waiter, &hss);
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
     dp->ProducerClose();
   }
 
   // Check MOJO_HANDLE_SIGNAL_PEER_CLOSED on consumer.
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     // Add a waiter.
     waiter.Init();
     ASSERT_EQ(MOJO_RESULT_OK,
               dp->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                                       12, nullptr));
 
     // Close the producer.
     dp->ProducerClose();
 
     // It should be signaled.
@@ skipping 14 matching lines @@
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       1000 * sizeof(int32_t)                    // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     Waiter waiter;
     uint32_t context = 0;
     HandleSignalsState hss;
 
     // Never writable.
     waiter.Init();
     hss = HandleSignalsState();
     EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
               dp->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 12,
                                       &hss));
@@ skipping 134 matching lines @@
                                       &hss));
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
     dp->ConsumerClose();
   }
 
   // Test with two-phase APIs and closing the producer with an active consumer
   // waiter.
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
     Waiter waiter;
     uint32_t context = 0;
     HandleSignalsState hss;
 
     // Write two elements.
     int32_t* elements = nullptr;
     void* buffer = nullptr;
     // Request room for three (but we'll only write two).
     uint32_t num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
     EXPECT_EQ(MOJO_RESULT_OK,
@@ skipping 81 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       1000 * sizeof(int32_t)                    // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
   Waiter waiter;
   HandleSignalsState hss;
 
   // It should be writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
@@ skipping 132 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                                  // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),          // |element_num_bytes|.
       1 * sizeof(int32_t)                              // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
   Waiter waiter;
   HandleSignalsState hss;
 
   // Writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
@@ skipping 111 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                                  // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),          // |element_num_bytes|.
       10 * sizeof(int32_t)                             // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   int32_t buffer[100] = {0};
   uint32_t num_bytes = 0;
 
   num_bytes = 20u * sizeof(int32_t);
   Seq(0, arraysize(buffer), buffer);
   // Try writing more than capacity. (This test relies on the implementation
   // enforcing the capacity strictly.)
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerWriteData(UserPointer<const void>(buffer),
@@ skipping 172 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       10 * sizeof(int32_t)                      // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   // Try writing way too much.
   uint32_t num_bytes = 20u * sizeof(int32_t);
   int32_t buffer[100];
   Seq(0, arraysize(buffer), buffer);
   EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
             dp->ProducerWriteData(UserPointer<const void>(buffer),
                                   MakeUserPointer(&num_bytes), true));
 
   // Should still be empty.
@@ skipping 139 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                                  // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),          // |element_num_bytes|.
       10 * sizeof(int32_t)                             // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   // Try writing way too much.
   uint32_t num_bytes = 20u * sizeof(int32_t);
   int32_t buffer[100];
   Seq(0, arraysize(buffer), buffer);
   EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
             dp->ProducerWriteData(UserPointer<const void>(buffer),
                                   MakeUserPointer(&num_bytes), true));
 
   // Write some stuff.
@@ skipping 78 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       10 * sizeof(int32_t)                      // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   // Try writing way too much (two-phase).
   uint32_t num_bytes = 20u * sizeof(int32_t);
   void* write_ptr = nullptr;
   EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
             dp->ProducerBeginWriteData(MakeUserPointer(&write_ptr),
                                        MakeUserPointer(&num_bytes), true));
 
   // Try writing an amount which isn't a multiple of the element size
   // (two-phase).
@@ skipping 121 matching lines @@
       1u,                                       // |element_num_bytes|.
       100u                                      // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
   // This test won't be valid if |ValidateCreateOptions()| decides to give the
   // pipe more space.
   ASSERT_EQ(100u, validated_options.capacity_num_bytes);
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   // Write 20 bytes.
   uint32_t num_bytes = 20u;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerWriteData(UserPointer<const void>(&test_data[0]),
                                   MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(20u, num_bytes);
 
   // Read 10 bytes.
   unsigned char read_buffer[1000] = {0};
@@ skipping 62 matching lines @@
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       1u,                                       // |element_num_bytes|.
       1000u                                     // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
   // Close producer first, then consumer.
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
     // Write some data, so we'll have something to read.
     uint32_t num_bytes = kTestDataSize;
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->ProducerWriteData(UserPointer<const void>(kTestData),
                                     MakeUserPointer(&num_bytes), false));
     EXPECT_EQ(kTestDataSize, num_bytes);
 
     // Write it again, so we'll have something left over.
     num_bytes = kTestDataSize;
@@ skipping 35 matching lines @@
                                         MakeUserPointer(&num_bytes), false));
     EXPECT_TRUE(read_buffer_ptr);
     EXPECT_EQ(kTestDataSize, num_bytes);
 
     // Close the consumer, which cancels the two-phase read.
     dp->ConsumerClose();
   }
 
   // Close consumer first, then producer.
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
     // Write some data, so we'll have something to read.
     uint32_t num_bytes = kTestDataSize;
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->ProducerWriteData(UserPointer<const void>(kTestData),
                                     MakeUserPointer(&num_bytes), false));
     EXPECT_EQ(kTestDataSize, num_bytes);
 
     // Start two-phase write.
     void* write_buffer_ptr = nullptr;
@@ skipping 35 matching lines @@
     EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
               dp->ProducerBeginWriteData(MakeUserPointer(&write_buffer_ptr),
                                          MakeUserPointer(&num_bytes), false));
 
     dp->ProducerClose();
   }
 
   // Test closing the consumer first, then the producer, with an active
   // two-phase write.
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
     // Start two-phase write.
     void* write_buffer_ptr = nullptr;
     uint32_t num_bytes = 0u;
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->ProducerBeginWriteData(MakeUserPointer(&write_buffer_ptr),
                                          MakeUserPointer(&num_bytes), false));
     EXPECT_TRUE(write_buffer_ptr);
     ASSERT_GT(num_bytes, kTestDataSize);
 
     dp->ConsumerClose();
     dp->ProducerClose();
   }
 
   // Test closing the producer and then trying to read (with no data).
   {
-    scoped_refptr<LocalDataPipeImpl> dp(
-        new LocalDataPipeImpl(validated_options));
+    scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
     // Write some data, so we'll have something to read.
     uint32_t num_bytes = kTestDataSize;
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->ProducerWriteData(UserPointer<const void>(kTestData),
                                     MakeUserPointer(&num_bytes), false));
     EXPECT_EQ(kTestDataSize, num_bytes);
 
     // Close the producer.
     dp->ProducerClose();
@@ skipping 42 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       10 * sizeof(int32_t)                      // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   // No data.
   uint32_t num_bytes = 1000u;
   EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerQueryData(MakeUserPointer(&num_bytes)));
   EXPECT_EQ(0u, num_bytes);
 
   // Try "ending" a two-phase write when one isn't active.
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             dp->ProducerEndWriteData(1u * sizeof(int32_t)));
 
@@ skipping 105 matching lines @@
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                                  // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
       1,                                               // |element_num_bytes|.
       2                                                // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
-  scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+  scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
 
   // Write some elements.
   char elements[2] = {'a', 'b'};
   uint32_t num_bytes = 2u;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerWriteData(UserPointer<const void>(elements),
                                   MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(2u, num_bytes);
 
   // Begin reading.
@@ skipping 44 matching lines @@
   // End reading.
   EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerEndReadData(2u));
 
   dp->ProducerClose();
   dp->ConsumerClose();
 }
 
 }  // namespace
 }  // namespace system
 }  // namespace mojo