Bug 1171994: Add |RilSocket| and |RilSocketConsumer|, r=htsai

|RilSocket| and |RilSocketConsumer| are copies of the respective stream-
socket classes. Improvements to the RIL I/O code will be implemented on
top of the new classes.

--HG--
rename : ipc/unixsocket/StreamSocket.cpp => ipc/ril/RilSocket.cpp
rename : ipc/unixsocket/StreamSocket.h => ipc/ril/RilSocket.h
rename : ipc/unixsocket/StreamSocketConsumer.cpp => ipc/ril/RilSocketConsumer.cpp
rename : ipc/unixsocket/StreamSocketConsumer.h => ipc/ril/RilSocketConsumer.h

ipc/ril/RilSocket.cpp

@@ -0,0 +1,413 @@
+/* -*- Mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; tab-width: 40 -*- */
+/* vim: set ts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#include "RilSocket.h"
+#include <fcntl.h>
+#include "mozilla/ipc/UnixSocketConnector.h"
+#include "mozilla/RefPtr.h"
+#include "nsXULAppAPI.h"
+#include "RilSocketConsumer.h"
+
+static const size_t MAX_READ_SIZE = 1 << 16;
+
+namespace mozilla {
+namespace ipc {
+
+//
+// RilSocketIO
+//
+
+class RilSocketIO final : public ConnectionOrientedSocketIO
+{
+public:
+  class ConnectTask;
+  class DelayedConnectTask;
+  class ReceiveTask;
+
+  RilSocketIO(MessageLoop* aConsumerLoop,
+              MessageLoop* aIOLoop,
+              RilSocket* aRilSocket,
+              UnixSocketConnector* aConnector);
+  ~RilSocketIO();
+
+  RilSocket* GetRilSocket();
+  DataSocket* GetDataSocket();
+
+  // Delayed-task handling
+  //
+
+  void SetDelayedConnectTask(CancelableTask* aTask);
+  void ClearDelayedConnectTask();
+  void CancelDelayedConnectTask();
+
+  // Methods for |DataSocket|
+  //
+
+  nsresult QueryReceiveBuffer(UnixSocketIOBuffer** aBuffer) override;
+  void ConsumeBuffer() override;
+  void DiscardBuffer() override;
+
+  // Methods for |SocketIOBase|
+  //
+
+  SocketBase* GetSocketBase() override;
+
+  bool IsShutdownOnConsumerThread() const override;
+  bool IsShutdownOnIOThread() const override;
+
+  void ShutdownOnConsumerThread() override;
+  void ShutdownOnIOThread() override;
+
+private:
+  /**
+   * Consumer pointer. Non-thread safe RefPtr, so should only be manipulated
+   * directly from consumer thread. All non-consumer-thread accesses should
+   * happen with mIO as container.
+   */
+  RefPtr<RilSocket> mRilSocket;
+
+  /**
+   * If true, do not requeue whatever task we're running
+   */
+  bool mShuttingDownOnIOThread;
+
+  /**
+   * Task member for delayed connect task. Should only be access on consumer
+   * thread.
+   */
+  CancelableTask* mDelayedConnectTask;
+
+  /**
+   * I/O buffer for received data
+   */
+  nsAutoPtr<UnixSocketRawData> mBuffer;
+};
+
+RilSocketIO::RilSocketIO(MessageLoop* aConsumerLoop,
+                         MessageLoop* aIOLoop,
+                         RilSocket* aRilSocket,
+                         UnixSocketConnector* aConnector)
+  : ConnectionOrientedSocketIO(aConsumerLoop, aIOLoop, aConnector)
+  , mRilSocket(aRilSocket)
+  , mShuttingDownOnIOThread(false)
+  , mDelayedConnectTask(nullptr)
+{
+  MOZ_ASSERT(mRilSocket);
+}
+
+RilSocketIO::~RilSocketIO()
+{
+  MOZ_ASSERT(IsConsumerThread());
+  MOZ_ASSERT(IsShutdownOnConsumerThread());
+}
+
+RilSocket*
+RilSocketIO::GetRilSocket()
+{
+  return mRilSocket.get();
+}
+
+DataSocket*
+RilSocketIO::GetDataSocket()
+{
+  return mRilSocket.get();
+}
+
+void
+RilSocketIO::SetDelayedConnectTask(CancelableTask* aTask)
+{
+  MOZ_ASSERT(IsConsumerThread());
+
+  mDelayedConnectTask = aTask;
+}
+
+void
+RilSocketIO::ClearDelayedConnectTask()
+{
+  MOZ_ASSERT(IsConsumerThread());
+
+  mDelayedConnectTask = nullptr;
+}
+
+void
+RilSocketIO::CancelDelayedConnectTask()
+{
+  MOZ_ASSERT(IsConsumerThread());
+
+  if (!mDelayedConnectTask) {
+    return;
+  }
+
+  mDelayedConnectTask->Cancel();
+  ClearDelayedConnectTask();
+}
+
+// |DataSocketIO|
+
+nsresult
+RilSocketIO::QueryReceiveBuffer(UnixSocketIOBuffer** aBuffer)
+{
+  MOZ_ASSERT(aBuffer);
+
+  if (!mBuffer) {
+    mBuffer = new UnixSocketRawData(MAX_READ_SIZE);
+  }
+  *aBuffer = mBuffer.get();
+
+  return NS_OK;
+}
+
+/**
+ * |ReceiveTask| transfers data received on the I/O thread
+ * to an instance of |RilSocket| on the consumer thread.
+ */
+class RilSocketIO::ReceiveTask final : public SocketTask<RilSocketIO>
+{
+public:
+  ReceiveTask(RilSocketIO* aIO, UnixSocketBuffer* aBuffer)
+    : SocketTask<RilSocketIO>(aIO)
+    , mBuffer(aBuffer)
+  { }
+
+  void Run() override
+  {
+    RilSocketIO* io = SocketTask<RilSocketIO>::GetIO();
+
+    MOZ_ASSERT(io->IsConsumerThread());
+
+    if (NS_WARN_IF(io->IsShutdownOnConsumerThread())) {
+      // Since we've already explicitly closed and the close
+      // happened before this, this isn't really an error.
+      return;
+    }
+
+    RilSocket* rilSocket = io->GetRilSocket();
+    MOZ_ASSERT(rilSocket);
+
+    rilSocket->ReceiveSocketData(mBuffer);
+  }
+
+private:
+  nsAutoPtr<UnixSocketBuffer> mBuffer;
+};
+
+void
+RilSocketIO::ConsumeBuffer()
+{
+  GetConsumerThread()->PostTask(FROM_HERE,
+                                new ReceiveTask(this, mBuffer.forget()));
+}
+
+void
+RilSocketIO::DiscardBuffer()
+{
+  // Nothing to do.
+}
+
+// |SocketIOBase|
+
+SocketBase*
+RilSocketIO::GetSocketBase()
+{
+  return GetDataSocket();
+}
+
+bool
+RilSocketIO::IsShutdownOnConsumerThread() const
+{
+  MOZ_ASSERT(IsConsumerThread());
+
+  return mRilSocket == nullptr;
+}
+
+bool
+RilSocketIO::IsShutdownOnIOThread() const
+{
+  return mShuttingDownOnIOThread;
+}
+
+void
+RilSocketIO::ShutdownOnConsumerThread()
+{
+  MOZ_ASSERT(IsConsumerThread());
+  MOZ_ASSERT(!IsShutdownOnConsumerThread());
+
+  mRilSocket = nullptr;
+}
+
+void
+RilSocketIO::ShutdownOnIOThread()
+{
+  MOZ_ASSERT(!IsConsumerThread());
+  MOZ_ASSERT(!mShuttingDownOnIOThread);
+
+  Close(); // will also remove fd from I/O loop
+  mShuttingDownOnIOThread = true;
+}
+
+//
+// Socket tasks
+//
+
+class RilSocketIO::ConnectTask final
+  : public SocketIOTask<RilSocketIO>
+{
+public:
+  ConnectTask(RilSocketIO* aIO)
+    : SocketIOTask<RilSocketIO>(aIO)
+  { }
+
+  void Run() override
+  {
+    MOZ_ASSERT(!GetIO()->IsConsumerThread());
+    MOZ_ASSERT(!IsCanceled());
+
+    GetIO()->Connect();
+  }
+};
+
+class RilSocketIO::DelayedConnectTask final
+  : public SocketIOTask<RilSocketIO>
+{
+public:
+  DelayedConnectTask(RilSocketIO* aIO)
+    : SocketIOTask<RilSocketIO>(aIO)
+  { }
+
+  void Run() override
+  {
+    MOZ_ASSERT(GetIO()->IsConsumerThread());
+
+    if (IsCanceled()) {
+      return;
+    }
+
+    RilSocketIO* io = GetIO();
+    if (io->IsShutdownOnConsumerThread()) {
+      return;
+    }
+
+    io->ClearDelayedConnectTask();
+    io->GetIOLoop()->PostTask(FROM_HERE, new ConnectTask(io));
+  }
+};
+
+//
+// RilSocket
+//
+
+RilSocket::RilSocket(RilSocketConsumer* aConsumer, int aIndex)
+  : mIO(nullptr)
+  , mConsumer(aConsumer)
+  , mIndex(aIndex)
+{
+  MOZ_ASSERT(mConsumer);
+}
+
+RilSocket::~RilSocket()
+{
+  MOZ_ASSERT(!mIO);
+}
+
+void
+RilSocket::ReceiveSocketData(nsAutoPtr<UnixSocketBuffer>& aBuffer)
+{
+  mConsumer->ReceiveSocketData(mIndex, aBuffer);
+}
+
+nsresult
+RilSocket::Connect(UnixSocketConnector* aConnector, int aDelayMs,
+                   MessageLoop* aConsumerLoop, MessageLoop* aIOLoop)
+{
+  MOZ_ASSERT(!mIO);
+
+  mIO = new RilSocketIO(aConsumerLoop, aIOLoop, this, aConnector);
+  SetConnectionStatus(SOCKET_CONNECTING);
+
+  if (aDelayMs > 0) {
+    RilSocketIO::DelayedConnectTask* connectTask =
+      new RilSocketIO::DelayedConnectTask(mIO);
+    mIO->SetDelayedConnectTask(connectTask);
+    MessageLoop::current()->PostDelayedTask(FROM_HERE, connectTask, aDelayMs);
+  } else {
+    aIOLoop->PostTask(FROM_HERE, new RilSocketIO::ConnectTask(mIO));
+  }
+
+  return NS_OK;
+}
+
+nsresult
+RilSocket::Connect(UnixSocketConnector* aConnector, int aDelayMs)
+{
+  return Connect(aConnector, aDelayMs,
+                 MessageLoop::current(), XRE_GetIOMessageLoop());
+}
+
+// |ConnectionOrientedSocket|
+
+nsresult
+RilSocket::PrepareAccept(UnixSocketConnector* aConnector,
+                         MessageLoop* aConsumerLoop,
+                         MessageLoop* aIOLoop,
+                         ConnectionOrientedSocketIO*& aIO)
+{
+  MOZ_CRASH("|RilSocket| does not support accepting connections.");
+}
+
+// |DataSocket|
+
+void
+RilSocket::SendSocketData(UnixSocketIOBuffer* aBuffer)
+{
+  MOZ_ASSERT(mIO);
+  MOZ_ASSERT(mIO->IsConsumerThread());
+  MOZ_ASSERT(!mIO->IsShutdownOnConsumerThread());
+
+  mIO->GetIOLoop()->PostTask(
+    FROM_HERE,
+    new SocketIOSendTask<RilSocketIO, UnixSocketIOBuffer>(mIO, aBuffer));
+}
+
+// |SocketBase|
+
+void
+RilSocket::Close()
+{
+  MOZ_ASSERT(mIO);
+  MOZ_ASSERT(mIO->IsConsumerThread());
+
+  mIO->CancelDelayedConnectTask();
+
+  // From this point on, we consider |mIO| as being deleted. We sever
+  // the relationship here so any future calls to |Connect| will create
+  // a new I/O object.
+  mIO->ShutdownOnConsumerThread();
+  mIO->GetIOLoop()->PostTask(FROM_HERE, new SocketIOShutdownTask(mIO));
+  mIO = nullptr;
+
+  NotifyDisconnect();
+}
+
+void
+RilSocket::OnConnectSuccess()
+{
+  mConsumer->OnConnectSuccess(mIndex);
+}
+
+void
+RilSocket::OnConnectError()
+{
+  mConsumer->OnConnectError(mIndex);
+}
+
+void
+RilSocket::OnDisconnect()
+{
+  mConsumer->OnDisconnect(mIndex);
+}
+
+} // namespace ipc
+} // namespace mozilla

ipc/ril/RilSocket.h

@@ -0,0 +1,95 @@
+/* -*- Mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; tab-width: 40 -*- */
+/* vim: set ts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#ifndef mozilla_ipc_RilSocket_h
+#define mozilla_ipc_RilSocket_h
+
+#include "mozilla/ipc/ConnectionOrientedSocket.h"
+
+class MessageLoop;
+
+namespace mozilla {
+namespace ipc {
+
+class RilSocketConsumer;
+class RilSocketIO;
+class UnixSocketConnector;
+
+class RilSocket final : public ConnectionOrientedSocket
+{
+public:
+  /**
+   * Constructs an instance of |RilSocket|.
+   *
+   * @param aConsumer The consumer for the socket.
+   * @param aIndex An arbitrary index.
+   */
+  RilSocket(RilSocketConsumer* aConsumer, int aIndex);
+
+  /**
+   * Method to be called whenever data is received. Consumer-thread only.
+   *
+   * @param aBuffer Data received from the socket.
+   */
+  void ReceiveSocketData(nsAutoPtr<UnixSocketBuffer>& aBuffer);
+
+  /**
+   * Starts a task on the socket that will try to connect to a socket in a
+   * non-blocking manner.
+   *
+   * @param aConnector Connector object for socket type specific functions
+   * @param aDelayMs Time delay in milliseconds.
+   * @param aConsumerLoop The socket's consumer thread.
+   * @param aIOLoop The socket's I/O thread.
+   * @return NS_OK on success, or an XPCOM error code otherwise.
+   */
+  nsresult Connect(UnixSocketConnector* aConnector, int aDelayMs,
+                   MessageLoop* aConsumerLoop, MessageLoop* aIOLoop);
+
+  /**
+   * Starts a task on the socket that will try to connect to a socket in a
+   * non-blocking manner.
+   *
+   * @param aConnector Connector object for socket type specific functions
+   * @param aDelayMs Time delay in milliseconds.
+   * @return NS_OK on success, or an XPCOM error code otherwise.
+   */
+  nsresult Connect(UnixSocketConnector* aConnector, int aDelayMs = 0);
+
+  // Methods for |ConnectionOrientedSocket|
+  //
+
+  nsresult PrepareAccept(UnixSocketConnector* aConnector,
+                         MessageLoop* aConsumerLoop,
+                         MessageLoop* aIOLoop,
+                         ConnectionOrientedSocketIO*& aIO) override;
+
+  // Methods for |DataSocket|
+  //
+
+  void SendSocketData(UnixSocketIOBuffer* aBuffer) override;
+
+  // Methods for |SocketBase|
+  //
+
+  void Close() override;
+  void OnConnectSuccess() override;
+  void OnConnectError() override;
+  void OnDisconnect() override;
+
+protected:
+  virtual ~RilSocket();
+
+private:
+  RilSocketIO* mIO;
+  RilSocketConsumer* mConsumer;
+  int mIndex;
+};
+
+} // namespace ipc
+} // namepsace mozilla
+
+#endif // mozilla_ipc_RilSocket_h

ipc/ril/RilSocketConsumer.cpp

@@ -0,0 +1,20 @@
+/* -*- Mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; tab-width: 40 -*- */
+/* vim: set ts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#include "RilSocketConsumer.h"
+
+namespace mozilla {
+namespace ipc {
+
+//
+// RilSocketConsumer
+//
+
+RilSocketConsumer::~RilSocketConsumer()
+{ }
+
+}
+}

ipc/ril/RilSocketConsumer.h

@@ -0,0 +1,60 @@
+/* -*- Mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; tab-width: 40 -*- */
+/* vim: set ts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#ifndef mozilla_ipc_RilSocketConsumer_h
+#define mozilla_ipc_RilSocketConsumer_h
+
+#include "nsAutoPtr.h"
+
+namespace mozilla {
+namespace ipc {
+
+class UnixSocketBuffer;
+
+/**
+ * |RilSocketConsumer| handles socket events and received data.
+ */
+class RilSocketConsumer
+{
+public:
+  /**
+   * Method to be called whenever data is received. Consumer-thread only.
+   *
+   * @param aIndex The index that has been given to the stream socket.
+   * @param aBuffer Data received from the socket.
+   */
+  virtual void ReceiveSocketData(int aIndex,
+                                 nsAutoPtr<UnixSocketBuffer>& aBuffer) = 0;
+
+  /**
+   * Callback for socket success. Consumer-thread only.
+   *
+   * @param aIndex The index that has been given to the stream socket.
+   */
+  virtual void OnConnectSuccess(int aIndex) = 0;
+
+  /**
+   * Callback for socket errors. Consumer-thread only.
+   *
+   * @param aIndex The index that has been given to the stream socket.
+   */
+  virtual void OnConnectError(int aIndex) = 0;
+
+  /**
+   * Callback for socket disconnect. Consumer-thread only.
+   *
+   * @param aIndex The index that has been given to the stream socket.
+   */
+  virtual void OnDisconnect(int aIndex) = 0;
+
+protected:
+  virtual ~RilSocketConsumer();
+};
+
+}
+}
+
+#endif

ipc/ril/moz.build

@@ -6,11 +6,15 @@
 
 EXPORTS.mozilla.ipc += [
     'Ril.h',
+    'RilSocket.h',
+    'RilSocketConsumer.h'
 ]
 
 SOURCES += [
     'Ril.cpp',
-    'RilConnector.cpp'
+    'RilConnector.cpp',
+    'RilSocket.cpp',
+    'RilSocketConsumer.cpp'
 ]
 
 include('/ipc/chromium/chromium-config.mozbuild')