EDK: Add implementation of data pipe consumer read threshold stuff.

mojo/edk/system/remote_producer_data_pipe_impl.cc

Index: mojo/edk/system/remote_producer_data_pipe_impl.cc
diff --git a/mojo/edk/system/remote_producer_data_pipe_impl.cc b/mojo/edk/system/remote_producer_data_pipe_impl.cc
index 35de22a008b0d02d38fd18e45ea9e05fe52856a8..653c2b4fdc90cb8a018c4d18d754e47e5931e550 100644
--- a/mojo/edk/system/remote_producer_data_pipe_impl.cc
+++ b/mojo/edk/system/remote_producer_data_pipe_impl.cc
@@ -281,16 +281,27 @@ MojoResult RemoteProducerDataPipeImpl::ConsumerEndReadData(
 HandleSignalsState RemoteProducerDataPipeImpl::ConsumerGetHandleSignalsState()
     const {
   HandleSignalsState rv;
-  if (current_num_bytes_ > 0) {
+  // |consumer_read_threshold_num_bytes()| is always at least 1.
+  if (current_num_bytes_ >= consumer_read_threshold_num_bytes()) {
+    if (!consumer_in_two_phase_read()) {
+      rv.satisfied_signals |=
+          MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD;
+    }
+    rv.satisfiable_signals |=
+        MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD;
+  } else if (current_num_bytes_ > 0u) {
     if (!consumer_in_two_phase_read())
       rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
-  } else if (producer_open()) {
-    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
   }
-  if (!producer_open())
+  if (producer_open()) {
+    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE |
+                              MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                              MOJO_HANDLE_SIGNAL_READ_THRESHOLD;
+  } else {
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
-  rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+  }
   return rv;
 }