IPC: Add attachment brokering support to the message header.

ipc/ipc_channel_win.cc

 // 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 "ipc/ipc_channel_win.h"
 
 #include <windows.h>
 
 #include "base/auto_reset.h"
 #include "base/bind.h"
@@ skipping 35 matching lines @@
       waiting_connect_(mode & MODE_SERVER_FLAG),
       processing_incoming_(false),
       validate_client_(false),
       client_secret_(0),
       broker_(broker),
       weak_factory_(this) {
   CreatePipe(channel_handle, mode);
 }
 
 ChannelWin::~ChannelWin() {
+  CleanUp();
   Close();
 }
 
 void ChannelWin::Close() {
   if (thread_check_.get())
     DCHECK(thread_check_->CalledOnValidThread());
 
   if (input_state_.is_pending || output_state_.is_pending)
     CancelIo(pipe_.Get());
 
   // Closing the handle at this point prevents us from issuing more requests
   // form OnIOCompleted().
   if (pipe_.IsValid())
     pipe_.Close();
 
   // Make sure all IO has completed.
   while (input_state_.is_pending || output_state_.is_pending) {
     base::MessageLoopForIO::current()->WaitForIOCompletion(INFINITE, this);
   }
 
   while (!output_queue_.empty()) {
-    Message* m = output_queue_.front();
+    OutputElement* element = output_queue_.front();
     output_queue_.pop();
-    delete m;
+    delete element;
   }
 }
 
 bool ChannelWin::Send(Message* message) {
   DCHECK(thread_check_->CalledOnValidThread());
   DVLOG(2) << "sending message @" << message << " on channel @" << this
            << " with type " << message->type()
            << " (" << output_queue_.size() << " in queue)";
 
   if (!prelim_queue_.empty()) {
@@ skipping 29 matching lines @@
 #ifdef IPC_MESSAGE_LOG_ENABLED
   Logging::GetInstance()->OnSendMessage(message, "");
 #endif
 
   TRACE_EVENT_WITH_FLOW0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
                          "ChannelWin::ProcessMessageForDelivery",
                          message->flags(),
                          TRACE_EVENT_FLAG_FLOW_OUT);
 
   // |output_queue_| takes ownership of |message|.
-  output_queue_.push(message);
+  OutputElement* element = new OutputElement(message);
+  output_queue_.push(element);
+
+#if USE_ATTACHMENT_BROKER
+  if (message->HasBrokerableAttachments()) {
+    // |output_queue_| takes ownership of |ids.buffer|.
+    Message::SerializedAttachmentIds ids =
+        message->SerializedIdsOfBrokerableAttachments();
+    OutputElement* new_element = new OutputElement(ids.buffer, ids.size);
+    output_queue_.push(new_element);
+  }
+#endif
+
   // ensure waiting to write
   if (!waiting_connect_) {
     if (!output_state_.is_pending) {
       if (!ProcessOutgoingMessages(NULL, 0))
         return false;
     }
   }
 
   return true;
 }
@@ skipping 216 matching lines @@
 
   // Don't send the secret to the untrusted process, and don't send a secret
   // if the value is zero (for IPC backwards compatability).
   int32 secret = validate_client_ ? 0 : client_secret_;
   if (!m->WriteInt(GetCurrentProcessId()) ||
       (secret && !m->WriteUInt32(secret))) {
     pipe_.Close();
     return false;
   }
 
-  output_queue_.push(m.release());
+  OutputElement* element = new OutputElement(m.release());
+  output_queue_.push(element);
   return true;
 }
 
 bool ChannelWin::Connect() {
   DLOG_IF(WARNING, thread_check_.get()) << "Connect called more than once";
 
   if (!thread_check_.get())
     thread_check_.reset(new base::ThreadChecker());
 
   if (!pipe_.IsValid())
@@ skipping 69 matching lines @@
   if (output_state_.is_pending) {
     DCHECK(context);
     output_state_.is_pending = false;
     if (!context || bytes_written == 0) {
       DWORD err = GetLastError();
       LOG(ERROR) << "pipe error: " << err;
       return false;
     }
     // Message was sent.
     CHECK(!output_queue_.empty());
-    Message* m = output_queue_.front();
+    OutputElement* element = output_queue_.front();
     output_queue_.pop();
-    delete m;
+    delete element;
   }
 
   if (output_queue_.empty())
     return true;
 
   if (!pipe_.IsValid())
     return false;
 
   // Write to pipe...
-  Message* m = output_queue_.front();
-  DCHECK(m->size() <= INT_MAX);
-  BOOL ok = WriteFile(pipe_.Get(),
-                      m->data(),
-                      static_cast<uint32>(m->size()),
-                      NULL,
+  OutputElement* element = output_queue_.front();
+  DCHECK(element->size() <= INT_MAX);
+  BOOL ok = WriteFile(pipe_.Get(), element->data(),
+                      static_cast<uint32>(element->size()), NULL,
                       &output_state_.context.overlapped);
   if (!ok) {
     DWORD write_error = GetLastError();
     if (write_error == ERROR_IO_PENDING) {
       output_state_.is_pending = true;
 
-      DVLOG(2) << "sent pending message @" << m << " on channel @" << this
-               << " with type " << m->type();
+      const Message* m = element->get_message();
+      if (m) {
+        DVLOG(2) << "sent pending message @" << m << " on channel @" << this
+                 << " with type " << m->type();
+      }
 
       return true;
     }
     LOG(ERROR) << "pipe error: " << write_error;
     return false;
   }
 
-  DVLOG(2) << "sent message @" << m << " on channel @" << this
-           << " with type " << m->type();
+  const Message* m = element->get_message();
+  if (m) {
+    DVLOG(2) << "sent message @" << m << " on channel @" << this
+             << " with type " << m->type();
+  }
 
   output_state_.is_pending = true;
   return true;
 }
 
 void ChannelWin::OnIOCompleted(
     base::MessageLoopForIO::IOContext* context,
     DWORD bytes_transfered,
     DWORD error) {
   bool ok = true;
@@ skipping 74 matching lines @@
   int secret;
   do {  // Guarantee we get a non-zero value.
     secret = base::RandInt(0, std::numeric_limits<int>::max());
   } while (secret == 0);
 
   id.append(GenerateUniqueRandomChannelID());
   return id.append(base::StringPrintf("\\%d", secret));
 }
 
 }  // namespace IPC