Binding media stream audio track to speech recognition [browser]

content/browser/speech/speech_recognition_audio_receiver.cc

Index: content/browser/speech/speech_recognition_audio_receiver.cc
diff --git a/content/browser/speech/speech_recognition_audio_receiver.cc b/content/browser/speech/speech_recognition_audio_receiver.cc
new file mode 100644
index 0000000000000000000000000000000000000000..275d140d4ddde0f0910ce0e7092bf4dbd1bbdb8b
--- /dev/null
+++ b/content/browser/speech/speech_recognition_audio_receiver.cc
@@ -0,0 +1,156 @@
+// Copyright 2014 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 "content/browser/speech/speech_recognition_audio_receiver.h"
+
+#include "base/bind.h"
+#include "base/process/process_handle.h"
+#include "base/threading/thread.h"
+#include "base/threading/thread_restrictions.h"
+#include "media/audio/audio_parameters.h"
+#include "media/base/audio_bus.h"
+
+namespace content {
+
+SpeechRecognitionAudioReceiver::SpeechRecognitionAudioReceiver(
+    const media::AudioParameters& params,
+    const base::ProcessHandle peer_process_handle,
+    const base::TimeDelta receive_timeout,
+    const OnDataCB& on_data_callback,
+    base::SharedMemoryHandle* foreign_memory_handle,
+    base::SyncSocket::TransitDescriptor* foreign_socket_descriptor) :
+      receive_timeout_(receive_timeout),
+      dropped_packets_(0) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK_LT(receive_timeout.InMilliseconds(), 1000);
+  DCHECK(foreign_memory_handle);
+  DCHECK(foreign_socket_descriptor);
+
+  // Create the shared memory and share with the renderer process.
+  uint32 shared_memory_size = sizeof(media::AudioInputBufferParameters) +
+                              media::AudioBus::CalculateMemorySize(params);
+  shared_memory_.reset(new base::SharedMemory());
+  if (!shared_memory_->CreateAndMapAnonymous(shared_memory_size)) {
+    DLOG(ERROR) << "Failed to create shared memory for the speech recognizer";
+    shared_memory_.reset();
+    return;
+  }
+
+  if (!shared_memory_->ShareToProcess(base::GetCurrentProcessHandle(),
+                                      foreign_memory_handle)) {
+    DLOG(ERROR) << "Failed to map and share the shared memory";
+    shared_memory_.reset();
+    return;
+  }
+
+  foreign_socket_.reset(new base::CancelableSyncSocket());
+  audio_socket_.reset(new base::CancelableSyncSocket());
+  if (!base::SyncSocket::CreatePair(audio_socket_.get(),
+                                    foreign_socket_.get())) {
+    shared_memory_.reset();
+    foreign_socket_.reset();
+    audio_socket_.reset();
+    DLOG(ERROR) << "Cannot create SyncSocket pair";
+    return;
+  }
+
+  if (!foreign_socket_->PrepareTransitDescriptor(peer_process_handle,
+                                                 foreign_socket_descriptor)) {
+    shared_memory_.reset();
+    foreign_socket_.reset();
+    audio_socket_.reset();
+    DLOG(ERROR) << "Cannot prepare syncsocket transit descriptor";
+    return;
+  }
+
+  media::AudioInputBuffer* buffer =
+      static_cast<media::AudioInputBuffer*>(shared_memory_->memory());
+  audio_bus_ = media::AudioBus::WrapMemory(params, buffer->audio);
+
+  // Use |buffer->params.size| from shared memory to synchronize by counting
+  // audio buffers when receiving from the renderer.
+  buffer_index_ = &(buffer->params.size);
+  *buffer_index_ = 0U;
+
+  // The thread check will be first invoked in |OnData()|.
+  audio_thread_checker_.DetachFromThread();
+  audio_thread_.reset(new base::Thread("SpeechRecognitionAudioTrackThread"));
+  audio_thread_->Start();
+
+  on_data_cb_ = on_data_callback;
+
+  listening_ = audio_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&SpeechRecognitionAudioReceiver::OnData,
+                            base::Unretained(this)));
+  DCHECK(listening_);
+}
+
+SpeechRecognitionAudioReceiver::~SpeechRecognitionAudioReceiver() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK(audio_thread_.get());
+
+  // We need to stop the |audio_thread_| to properly clean up.
+  // We must allow the IO until the task in the thread's message loop completes.
+  // This should execute almost immediately in most of cases since we expect
+  // the sockets were closed and |OnData()| returned without posting a new task.
+  const bool previous_io_allowed = base::ThreadRestrictions::SetIOAllowed(true);
+
+  // Make sure we close the socket so the thread doesn't block on it.
+  audio_socket_->Shutdown();
+  audio_thread_->Stop();
+  base::ThreadRestrictions::SetIOAllowed(previous_io_allowed);
+}
+
+bool SpeechRecognitionAudioReceiver::listening() {
+ DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+ return listening_;
+}
+
+void SpeechRecognitionAudioReceiver::OnData() {
+  DCHECK(!BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK(audio_thread_checker_.CalledOnValidThread());
+
+  // Maximum of consecutive dropped packets before we abort posting new tasks.
+  // Since |ReceiveWithTimeout()| can return 0U if either the socket was closed
+  // or a timeout occured, this constant is used to handle both situations:
+  // 1) If the renderer is slow to start sending, we allow 2 timeouts.
+  //    E.g. if audio chunk is 200 ms and timeout is 400 ms, we allow 800 ms.
+  // 2) If we haven't received any data, we will try to repost the task
+  //    2 times before we abort.
+  //    E.g. The socket was closed before we received any data. This happens
+  //    if the user refreshes the page (closing the socket) before we get the
+  //    the first packet.
+  const uint32 kMaxDroppedPackets = 2;
+
+  // Receive with timeout not to block indefinitely.
+  size_t bytes_received = audio_socket_->ReceiveWithTimeout(
+      buffer_index_, sizeof(*buffer_index_), receive_timeout_);
+
+  // Message must be either complete or empty.
+  DCHECK_EQ(0U, bytes_received % sizeof(*buffer_index_));
+
+  if (bytes_received == sizeof(*buffer_index_)) {
+
+    // Any previous issues with the socket are ignored.
+    dropped_packets_ = 0;
+
+    // Execute the client callback on this |audio_thread_|.
+    on_data_cb_.Run(NULL, audio_bus_.get());
+
+    // Notify host via shared memory that the buffer was consumed.
+    ++(*buffer_index_);
+
+  } else if (++dropped_packets_ > kMaxDroppedPackets) {
+    audio_socket_->Shutdown();
+
+    return;
+  }
+
+  // Post this task again to |audio_thread_| for next round of data.
+  base::MessageLoopProxy::current()->PostTask(
+      FROM_HERE, base::Bind(&SpeechRecognitionAudioReceiver::OnData,
+                            base::Unretained(this)));
+}
+
+}  // namespace content