Pepper WebSocket API: Implement new design Chrome IPC

ppapi/proxy/websocket_resource.cc

Index: ppapi/proxy/websocket_resource.cc
diff --git a/ppapi/proxy/websocket_resource.cc b/ppapi/proxy/websocket_resource.cc
new file mode 100644
index 0000000000000000000000000000000000000000..d7429a9f930513fccde686bc4477b24b0c85eff0
--- /dev/null
+++ b/ppapi/proxy/websocket_resource.cc
@@ -0,0 +1,544 @@
+// Copyright (c) 2012 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 "ppapi/proxy/websocket_resource.h"
+
+#include <set>
+#include <vector>
+
+#include "ppapi/c/pp_errors.h"
+#include "ppapi/proxy/ppapi_messages.h"
+#include "ppapi/shared_impl/ppapi_globals.h"
+#include "ppapi/shared_impl/var.h"
+#include "ppapi/shared_impl/var_tracker.h"
+#include "third_party/WebKit/Source/WebKit/chromium/public/WebSocket.h"
+
+namespace {
+
+const uint32_t kMaxReasonSizeInBytes = 123;
+const size_t kBaseFramingOverhead = 2;
+const size_t kMaskingKeyLength = 4;
+const size_t kMinimumPayloadSizeWithTwoByteExtendedPayloadLength = 126;
+const size_t kMinimumPayloadSizeWithEightByteExtendedPayloadLength = 0x10000;
+
+uint64_t SaturateAdd(uint64_t a, uint64_t b) {
+  if (kuint64max - a < b)
+    return kuint64max;
+  return a + b;
+}
+
+uint64_t GetFrameSize(uint64_t payload_size) {
+  uint64_t overhead = kBaseFramingOverhead + kMaskingKeyLength;
+  if (payload_size > kMinimumPayloadSizeWithEightByteExtendedPayloadLength)
+    overhead += 8;
+  else if (payload_size > kMinimumPayloadSizeWithTwoByteExtendedPayloadLength)
+    overhead += 2;
+  return SaturateAdd(payload_size, overhead);
+}
+
+bool InValidStateToReceive(PP_WebSocketReadyState state) {
+  return state == PP_WEBSOCKETREADYSTATE_OPEN ||
+         state == PP_WEBSOCKETREADYSTATE_CLOSING;
+}
+
+}  // namespace
+
+
+namespace ppapi {
+namespace proxy {
+
+WebSocketResource::WebSocketResource(Connection connection,
+                                     PP_Instance instance)
+    : PluginResource(connection, instance),
+      connect_request_id_(0),
+      close_request_id_(0),
+      state_(PP_WEBSOCKETREADYSTATE_INVALID),
+      error_was_received_(false),
+      receive_callback_var_(NULL),
+      empty_string_(new StringVar("", 0)),
+      close_code_(0),
+      close_reason_(NULL),
+      close_was_clean_(PP_FALSE),
+      extensions_(NULL),
+      protocol_(NULL),
+      url_(NULL),
+      buffered_amount_(0),
+      buffered_amount_after_close_(0) {
+}
+
+WebSocketResource::~WebSocketResource() {
+}
+
+thunk::PPB_WebSocket_API* WebSocketResource::AsPPB_WebSocket_API() {
+  return this;
+}
+
+int32_t WebSocketResource::Connect(
+    PP_Var url,

[brettw, 2012/10/03 04:54:02]

Can you make this (in the _api as well) a const ref? The same in the other
functions in the abstract interface.

[Takashi Toyoshima, 2012/10/03 11:42:20]

I think we have not used const ref for PP_Var usually in NaCl code base. But, I
agreed that we should use const ref because it is recommended as Google/Chromium
coding styles.
OK, I fix it in the next CL.

[brettw, 2012/10/03 20:30:24]

We shouldn't change the C bindings, which are the binary layer which must be C.
But this layer of stuff, along with the thunk code, is internal functions that
can do whatever we want. I usually use const refs for the _API interfaces as per
Chrome style. It should be easy to change in this patch since you're already
touching all of the places you need to change anyway.

[Takashi Toyoshima, 2012/10/05 07:35:01]

Thank you.
Now I understand the reason of PP_Var correctly.

+    const PP_Var protocols[],
+    uint32_t protocol_count,
+    scoped_refptr<TrackedCallback> callback) {
+  if (TrackedCallback::IsPending(connect_callback_))
+    return PP_ERROR_INPROGRESS;
+
+  // Connect() can be called at most once.
+  if (state_ != PP_WEBSOCKETREADYSTATE_INVALID)
+    return PP_ERROR_INPROGRESS;
+  state_ = PP_WEBSOCKETREADYSTATE_CLOSED;
+
+  // Get the URL.
+  scoped_refptr<StringVar> url_string = StringVar::FromPPVar(url);
+  if (!url_string)
+    return PP_ERROR_BADARGUMENT;
+
+  // Validate protocols.
+  std::string protocol_string;
+  std::set<std::string> protocol_set;
+  for (uint32_t i = 0; i < protocol_count; i++) {
+    scoped_refptr<StringVar> protocol(StringVar::FromPPVar(protocols[i]));
+
+    // Check invalid and empty entries.
+    if (!protocol || !protocol->value().length())
+      return PP_ERROR_BADARGUMENT;
+
+    // Check duplicated protocol entries.
+    if (protocol_set.find(protocol->value()) != protocol_set.end())
+      return PP_ERROR_BADARGUMENT;
+    protocol_set.insert(protocol->value());
+
+    // Check containing characters.
+    for (std::string::const_iterator it = protocol->value().begin();
+        it != protocol->value().end();
+        ++it) {
+      uint8_t character = *it;
+      // WebSocket specification says "(Subprotocol string must consist of)
+      // characters in the range U+0021 to U+007E not including separator
+      // characters as defined in [RFC2616]."
+      const uint8_t minimumProtocolCharacter = '!';  // U+0021.

[brettw, 2012/10/03 04:54:02]

To what extent is this checking security sensitive? If it's not sensitive and
just a nice thing to check, you should mention that in the comment. But if its
important that this check is correct (which I suspect it is) we should be doing
it on the host side instead, since the plugin code isn't trusted.

I think it's OK to have an asynchronous failure if you pass a bad protocol.
Although less convenient for the plugin, it prevents duplicate code and
presumably the protocols are hardcoded, so it you screw it up you just fix your
code and then you don't have any more problems.

[Takashi Toyoshima, 2012/10/03 11:46:28]

Firstly, I think this check is safe to run in NaCl.
But, now I notice that injecting <cr><lf> might cause security problem because
it means that user can inject arbitrary http request headers. Anyway, it must be
safe to implement this in renderer against unknown security issues.
I removed this check into renderer side.

+      const uint8_t maximumProtocolCharacter = '~';  // U+007E.
+      if (character < minimumProtocolCharacter ||
+          character > maximumProtocolCharacter ||
+          character == '"' || character == '(' || character == ')' ||
+          character == ',' || character == '/' ||
+          (character >= ':' && character <= '@') ||  // U+003A - U+0040
+          (character >= '[' && character <= ']') ||  // U+005B - u+005D
+          character == '{' || character == '}')
+        return PP_ERROR_BADARGUMENT;
+    }
+    // Join protocols with the comma separator.
+    if (i != 0)
+      protocol_string.append(",");
+    protocol_string.append(protocol->value());
+  }
+
+  // Install callback.
+  connect_callback_ = callback;
+
+  // Create remote host in the renderer, then request to check the URL and
+  // establish the connection.
+  state_ = PP_WEBSOCKETREADYSTATE_CONNECTING;
+  SendCreateToRenderer(PpapiHostMsg_WebSocket_Create());
+  connect_request_id_ =
+      CallRenderer(PpapiHostMsg_WebSocket_Connect(url_string->value(),
+                                                  protocol_string));
+
+  return PP_OK_COMPLETIONPENDING;
+}
+
+int32_t WebSocketResource::Close(uint16_t code,
+                                 PP_Var reason,
+                                 scoped_refptr<TrackedCallback> callback) {
+  if (TrackedCallback::IsPending(close_callback_))
+    return PP_ERROR_INPROGRESS;
+  if (state_ == PP_WEBSOCKETREADYSTATE_INVALID)
+    return PP_ERROR_FAILED;
+
+  // Validate |code| and |reason|.
+  scoped_refptr<StringVar> reason_string_var;
+  std::string reason_string;
+  WebKit::WebSocket::CloseEventCode event_code =
+      static_cast<WebKit::WebSocket::CloseEventCode>(code);
+  if (code == PP_WEBSOCKETSTATUSCODE_NOT_SPECIFIED) {
+    // PP_WEBSOCKETSTATUSCODE_NOT_SPECIFIED and CloseEventCodeNotSpecified are
+    // assigned to different values. A conversion is needed if
+    // PP_WEBSOCKETSTATUSCODE_NOT_SPECIFIED is specified.
+    event_code = WebKit::WebSocket::CloseEventCodeNotSpecified;
+  } else {
+    if (!(code == PP_WEBSOCKETSTATUSCODE_NORMAL_CLOSURE ||
+        (PP_WEBSOCKETSTATUSCODE_USER_REGISTERED_MIN <= code &&
+        code <= PP_WEBSOCKETSTATUSCODE_USER_PRIVATE_MAX)))
+      // RFC 6455 limits applications to use reserved connection close code in
+      // section 7.4.2.. The WebSocket API (http://www.w3.org/TR/websockets/)
+      // defines this out of range error as InvalidAccessError in JavaScript.
+      return PP_ERROR_NOACCESS;
+
+    // |reason| must be ignored if it is PP_VARTYPE_UNDEFINED or |code| is
+    // PP_WEBSOCKETSTATUSCODE_NOT_SPECIFIED.
+    if (reason.type != PP_VARTYPE_UNDEFINED) {
+      // Validate |reason|.
+      reason_string_var = StringVar::FromPPVar(reason);
+      if (!reason_string_var ||
+          reason_string_var->value().size() > kMaxReasonSizeInBytes)
+        return PP_ERROR_BADARGUMENT;
+      reason_string = reason_string_var->value();
+    }
+  }
+
+  // Check state.
+  if (state_ == PP_WEBSOCKETREADYSTATE_CLOSING)
+    return PP_ERROR_INPROGRESS;
+  if (state_ == PP_WEBSOCKETREADYSTATE_CLOSED)
+    return PP_OK;
+
+  // Install |callback|.
+  close_callback_ = callback;
+
+  // Abort ongoing connect.
+  if (connect_callback_) {
+    state_ = PP_WEBSOCKETREADYSTATE_CLOSING;
+    // Need to do a "Post" to avoid reentering the plugin.
+    connect_callback_->PostAbort();
+    connect_callback_ = NULL;
+    CallRenderer(PpapiHostMsg_WebSocket_Fail(
+        "WebSocket was closed before the connection was established."));
+    return PP_OK_COMPLETIONPENDING;
+  }
+
+  // Abort ongoing receive.
+  if (receive_callback_) {
+    receive_callback_var_ = NULL;
+    // Need to do a "Post" to avoid reentering the plugin.
+    receive_callback_->PostAbort();
+    receive_callback_ = NULL;
+  }
+
+  // Close connection.
+  state_ = PP_WEBSOCKETREADYSTATE_CLOSING;
+  close_request_id_ =
+      CallRenderer(PpapiHostMsg_WebSocket_Close((int32_t)event_code,

[brettw, 2012/10/03 04:54:02]

Nit: use C++ casts.

[Takashi Toyoshima, 2012/10/03 11:42:20]

Done.

+                                                reason_string));
+  return PP_OK_COMPLETIONPENDING;
+}
+
+int32_t WebSocketResource::ReceiveMessage(
+    PP_Var* message,
+    scoped_refptr<TrackedCallback> callback) {
+  if (TrackedCallback::IsPending(receive_callback_))
+    return PP_ERROR_INPROGRESS;
+
+  // Check state.
+  if (state_ == PP_WEBSOCKETREADYSTATE_INVALID ||
+      state_ == PP_WEBSOCKETREADYSTATE_CONNECTING)
+    return PP_ERROR_BADARGUMENT;
+
+  // Just return received message if any received message is queued.
+  if (!received_messages_.empty()) {
+    receive_callback_var_ = message;
+    return DoReceive();
+  }
+
+  // Check state again. In CLOSED state, no more messages will be received.
+  if (state_ == PP_WEBSOCKETREADYSTATE_CLOSED)
+    return PP_ERROR_BADARGUMENT;
+
+  // Returns PP_ERROR_FAILED after an error is received and received messages
+  // is exhausted.
+  if (error_was_received_)
+    return PP_ERROR_FAILED;
+
+  // Or retain |message| as buffer to store and install |callback|.
+  receive_callback_var_ = message;
+  receive_callback_ = callback;
+
+  return PP_OK_COMPLETIONPENDING;
+}
+
+int32_t WebSocketResource::SendMessage(PP_Var message) {
+  // Check state.
+  if (state_ == PP_WEBSOCKETREADYSTATE_INVALID ||
+      state_ == PP_WEBSOCKETREADYSTATE_CONNECTING)
+    return PP_ERROR_BADARGUMENT;
+
+  if (state_ == PP_WEBSOCKETREADYSTATE_CLOSING ||
+      state_ == PP_WEBSOCKETREADYSTATE_CLOSED) {
+    // Handle buffered_amount_after_close_.
+    uint64_t payload_size = 0;
+    if (message.type == PP_VARTYPE_STRING) {
+      scoped_refptr<StringVar> message_string = StringVar::FromPPVar(message);
+      if (message_string)
+        payload_size += message_string->value().length();
+    } else if (message.type == PP_VARTYPE_ARRAY_BUFFER) {
+      scoped_refptr<ArrayBufferVar> message_array_buffer =
+          ArrayBufferVar::FromPPVar(message);
+      if (message_array_buffer)
+        payload_size += message_array_buffer->ByteLength();
+    } else {
+      // TODO(toyoshim): Support Blob.
+      return PP_ERROR_NOTSUPPORTED;
+    }
+
+    buffered_amount_after_close_ =
+        SaturateAdd(buffered_amount_after_close_, GetFrameSize(payload_size));
+
+    return PP_ERROR_FAILED;
+  }
+
+  // Send the message.
+  // TODO(toyoshim): Check if ppapi::proxy::SerializedVar can reduce copy.

[brettw, 2012/10/03 04:54:02]

Nope, doing this won't save a copy, so you can just remove this todo.

[Takashi Toyoshima, 2012/10/03 11:42:20]

Thanks!

+  if (message.type == PP_VARTYPE_STRING) {
+    // Convert message to std::string, then send it.
+    scoped_refptr<StringVar> message_string = StringVar::FromPPVar(message);
+    if (!message_string)
+      return PP_ERROR_BADARGUMENT;
+    CallRenderer(PpapiHostMsg_WebSocket_SendText(message_string->value()));
+  } else if (message.type == PP_VARTYPE_ARRAY_BUFFER) {
+    // Convert message to std::vector<uint8_t>, then send it.
+    scoped_refptr<ArrayBufferVar> message_arraybuffer =
+        ArrayBufferVar::FromPPVar(message);
+    if (!message_arraybuffer)
+      return PP_ERROR_BADARGUMENT;
+    uint8_t* message_data = static_cast<uint8_t*>(message_arraybuffer->Map());
+    uint32 message_length = message_arraybuffer->ByteLength();
+    std::vector<uint8_t> message_vector(message_data,
+                                        message_data + message_length);
+    CallRenderer(PpapiHostMsg_WebSocket_SendBinary(message_vector));
+  } else {
+    // TODO(toyoshim): Support Blob.
+    return PP_ERROR_NOTSUPPORTED;
+  }
+  return PP_OK;
+}
+
+uint64_t WebSocketResource::GetBufferedAmount() {
+  return SaturateAdd(buffered_amount_, buffered_amount_after_close_);
+}
+
+uint16_t WebSocketResource::GetCloseCode() {
+  return close_code_;
+}
+
+PP_Var WebSocketResource::GetCloseReason() {
+  if (!close_reason_)
+    return empty_string_->GetPPVar();
+  return close_reason_->GetPPVar();
+}
+
+PP_Bool WebSocketResource::GetCloseWasClean() {
+  return close_was_clean_;
+}
+
+PP_Var WebSocketResource::GetExtensions() {
+  return StringVar::StringToPPVar("");

[brettw, 2012/10/03 04:54:02]

Can you use std::string() instead of ""? It's more accurate and theoretically
saves some work in the string constructor.

[Takashi Toyoshima, 2012/10/03 11:42:20]

Done.

+}
+
+PP_Var WebSocketResource::GetProtocol() {
+  if (!protocol_)
+    return empty_string_->GetPPVar();
+  return protocol_->GetPPVar();
+}
+
+PP_WebSocketReadyState WebSocketResource::GetReadyState() {
+  return state_;
+}
+
+PP_Var WebSocketResource::GetURL() {
+  if (!url_)
+    return empty_string_->GetPPVar();
+  return url_->GetPPVar();
+}
+
+void WebSocketResource::OnReplyReceived(
+    const ResourceMessageReplyParams& params,
+    const IPC::Message& msg) {
+  // TODO(toyoshim): DISPATCH_RESOURCE_REPLY will be replaced with other
+  // mechanisms. WebSocket dispatches IPC messages manually for now and must be
+  // replaced with the new one in the future.
+  int32_t sequence = params.sequence();
+  if (!sequence) {
+    // Handle unsolicited IPCs.
+    switch (msg.type()) {
+      case PpapiPluginMsg_WebSocket_ReceiveTextReply::ID: {
+        PpapiPluginMsg_WebSocket_ReceiveTextReply::Schema::Param p;
+        if (PpapiPluginMsg_WebSocket_ReceiveTextReply::Read(&msg, &p))
+          OnPluginMsgReceiveTextReply(params, p.a);
+        else
+          NOTREACHED();
+        break; }

[brettw, 2012/10/03 04:54:02]

I've usually seen the closing } on the next line. So 
    break;
  }
  case PpapiPl...

[Takashi Toyoshima, 2012/10/03 11:42:20]

Done.

+      case PpapiPluginMsg_WebSocket_ReceiveBinaryReply::ID: {
+        PpapiPluginMsg_WebSocket_ReceiveBinaryReply::Schema::Param p;
+        if (PpapiPluginMsg_WebSocket_ReceiveBinaryReply::Read(&msg, &p))
+          OnPluginMsgReceiveBinaryReply(params, p.a);
+        else
+          NOTREACHED();
+        break; }
+      case PpapiPluginMsg_WebSocket_ErrorReply::ID: {
+        OnPluginMsgErrorReply(params);
+        break; }
+      case PpapiPluginMsg_WebSocket_BufferedAmountReply::ID: {
+        PpapiPluginMsg_WebSocket_BufferedAmountReply::Schema::Param p;
+        if (PpapiPluginMsg_WebSocket_BufferedAmountReply::Read(&msg, &p))
+          OnPluginMsgBufferedAmountReply(params, p.a);
+        else
+          NOTREACHED();
+        break; }
+      case PpapiPluginMsg_WebSocket_StateReply::ID: {
+        PpapiPluginMsg_WebSocket_StateReply::Schema::Param p;
+        if (PpapiPluginMsg_WebSocket_StateReply::Read(&msg, &p))
+          OnPluginMsgStateReply(params, p.a);
+        else
+          NOTREACHED();
+        break; }
+      case PpapiPluginMsg_WebSocket_ClosedReply::ID: {
+        PpapiPluginMsg_WebSocket_ClosedReply::Schema::Param p;
+        if (PpapiPluginMsg_WebSocket_ClosedReply::Read(&msg, &p))
+          OnPluginMsgClosedReply(params, p.a, p.b, p.c, p.d);
+        else
+          NOTREACHED();
+        break; }
+      default:
+        NOTREACHED();
+    }
+  } else if (sequence == connect_request_id_) {
+    PpapiPluginMsg_WebSocket_ConnectReply::Schema::Param p;
+    if (PpapiPluginMsg_WebSocket_ConnectReply::Read(&msg, &p))
+      OnPluginMsgConnectReply(params, p.a, p.b);
+    else  // On error use default params and run the callback.
+      OnPluginMsgConnectReply(params, std::string(), std::string());
+  } else if (sequence == close_request_id_) {
+    PpapiPluginMsg_WebSocket_CloseReply::Schema::Param p;
+    if (PpapiPluginMsg_WebSocket_CloseReply::Read(&msg, &p))
+      OnPluginMsgCloseReply(params, p.a, p.b, p.c, p.d);
+    else  // On error use default params and run the callback.
+      OnPluginMsgCloseReply(params, 0, false, 0, std::string());
+  } else {
+    NOTREACHED();
+  }
+}
+
+void WebSocketResource::OnPluginMsgConnectReply(
+    const ResourceMessageReplyParams& params,
+    const std::string& url,
+    const std::string& protocol) {
+  if (!TrackedCallback::IsPending(connect_callback_))
+    return;
+
+  int32_t result = params.result();
+  if (result == PP_OK) {
+    state_ = PP_WEBSOCKETREADYSTATE_OPEN;
+    protocol_ = new StringVar(protocol);
+  } else {
+    url_ = new StringVar(url);
+  }
+  if (result != PP_OK_COMPLETIONPENDING)
+    TrackedCallback::ClearAndRun(&connect_callback_, params.result());
+}
+
+void WebSocketResource::OnPluginMsgCloseReply(
+    const ResourceMessageReplyParams& params,
+    unsigned long buffered_amount,
+    bool was_clean,
+    unsigned short code,
+    const std::string& reason) {
+  // Set close related properties.
+  state_ = PP_WEBSOCKETREADYSTATE_CLOSED;
+  buffered_amount_ = buffered_amount;
+  close_was_clean_ = PP_FromBool(was_clean);
+  close_code_ = code;
+  close_reason_ = new StringVar(reason);
+
+  bool post = false;
+  if (TrackedCallback::IsPending(connect_callback_)) {
+    connect_callback_->PostRun(PP_ERROR_FAILED);
+    post = true;
+  }
+
+  if (TrackedCallback::IsPending(receive_callback_)) {
+    receive_callback_var_ = NULL;
+    receive_callback_->PostRun(PP_ERROR_FAILED);
+    post = true;
+  }
+
+  if (!TrackedCallback::IsPending(close_callback_))
+    return;
+
+  if (post)
+    close_callback_->PostRun(params.result());
+  else
+    TrackedCallback::ClearAndRun(&close_callback_, params.result());
+}
+
+void WebSocketResource::OnPluginMsgReceiveTextReply(
+    const ResourceMessageReplyParams& params,
+    const std::string& message) {
+  // Dispose packets after receiving an error or in invalid state.
+  if (error_was_received_ || !InValidStateToReceive(state_))
+     return;
+
+  // Append received data to queue.
+  received_messages_.push(scoped_refptr<Var>(new StringVar(message)));
+
+  if (!TrackedCallback::IsPending(receive_callback_))
+    return;
+
+  TrackedCallback::ClearAndRun(&receive_callback_, DoReceive());
+}
+
+void WebSocketResource::OnPluginMsgReceiveBinaryReply(
+    const ResourceMessageReplyParams& params,
+    const std::vector<uint8_t>& message) {
+  // Dispose packets after receiving an error or in invalid state.
+  if (error_was_received_ || !InValidStateToReceive(state_))
+     return;
+
+  // Append received data to queue.
+  scoped_refptr<Var> message_var(ArrayBufferVar::FromPPVar(
+      PpapiGlobals::Get()->GetVarTracker()->MakeArrayBufferPPVar(
+          message.size(),
+          &message.front())));
+  received_messages_.push(message_var);
+
+  if (!TrackedCallback::IsPending(receive_callback_))
+    return;
+
+  TrackedCallback::ClearAndRun(&receive_callback_, DoReceive());
+

[brettw, 2012/10/03 04:54:02]

Nit: blank line

[Takashi Toyoshima, 2012/10/03 11:42:20]

Done.

+}
+
+void WebSocketResource::OnPluginMsgErrorReply(
+    const ResourceMessageReplyParams& params) {
+}
+
+void WebSocketResource::OnPluginMsgBufferedAmountReply(
+    const ResourceMessageReplyParams& params,
+    unsigned long buffered_amount) {
+  buffered_amount_ = buffered_amount;
+}
+
+void WebSocketResource::OnPluginMsgStateReply(
+    const ResourceMessageReplyParams& params,
+    int32_t state) {
+  state_ = static_cast<PP_WebSocketReadyState>(state);
+}
+
+void WebSocketResource::OnPluginMsgClosedReply(
+    const ResourceMessageReplyParams& params,
+    unsigned long buffered_amount,
+    bool was_clean,
+    unsigned short code,
+    const std::string& reason) {
+  OnPluginMsgCloseReply(params, buffered_amount, was_clean, code, reason);
+}
+
+int32_t WebSocketResource::DoReceive() {
+  if (!receive_callback_var_)
+    return PP_OK;
+
+  *receive_callback_var_ = received_messages_.front()->GetPPVar();
+  received_messages_.pop();
+  receive_callback_var_ = NULL;
+  return PP_OK;
+}
+
+}  // namespace proxy
+}  // namespace ppapi