summaryrefslogtreecommitdiffstats
path: root/media/mtransport/nr_socket_prsock.cpp
diff options
context:
space:
mode:
authorMatt A. Tobin <mattatobin@localhost.localdomain>2018-02-02 04:16:08 -0500
committerMatt A. Tobin <mattatobin@localhost.localdomain>2018-02-02 04:16:08 -0500
commit5f8de423f190bbb79a62f804151bc24824fa32d8 (patch)
tree10027f336435511475e392454359edea8e25895d /media/mtransport/nr_socket_prsock.cpp
parent49ee0794b5d912db1f95dce6eb52d781dc210db5 (diff)
downloadUXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar.gz
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar.lz
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar.xz
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.zip
Add m-esr52 at 52.6.0
Diffstat (limited to 'media/mtransport/nr_socket_prsock.cpp')
-rw-r--r--media/mtransport/nr_socket_prsock.cpp2299
1 files changed, 2299 insertions, 0 deletions
diff --git a/media/mtransport/nr_socket_prsock.cpp b/media/mtransport/nr_socket_prsock.cpp
new file mode 100644
index 000000000..f22229df8
--- /dev/null
+++ b/media/mtransport/nr_socket_prsock.cpp
@@ -0,0 +1,2299 @@
+/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* 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/. */
+/*
+Modified version of nr_socket_local, adapted for NSPR
+*/
+
+/* 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/. */
+
+/*
+Original code from nICEr and nrappkit.
+
+nICEr copyright:
+
+Copyright (c) 2007, Adobe Systems, Incorporated
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+* Redistributions of source code must retain the above copyright
+ notice, this list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright
+ notice, this list of conditions and the following disclaimer in the
+ documentation and/or other materials provided with the distribution.
+
+* Neither the name of Adobe Systems, Network Resonance nor the names of its
+ contributors may be used to endorse or promote products derived from
+ this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
+nrappkit copyright:
+
+ Copyright (C) 2001-2003, Network Resonance, Inc.
+ Copyright (C) 2006, Network Resonance, Inc.
+ All Rights Reserved
+
+ Redistribution and use in source and binary forms, with or without
+ modification, are permitted provided that the following conditions
+ are met:
+
+ 1. Redistributions of source code must retain the above copyright
+ notice, this list of conditions and the following disclaimer.
+ 2. Redistributions in binary form must reproduce the above copyright
+ notice, this list of conditions and the following disclaimer in the
+ documentation and/or other materials provided with the distribution.
+ 3. Neither the name of Network Resonance, Inc. nor the name of any
+ contributors to this software may be used to endorse or promote
+ products derived from this software without specific prior written
+ permission.
+
+ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS ``AS IS''
+ AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ POSSIBILITY OF SUCH DAMAGE.
+
+
+ ekr@rtfm.com Thu Dec 20 20:14:49 2001
+*/
+
+#include <csi_platform.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/types.h>
+#include <assert.h>
+#include <errno.h>
+#include <string>
+
+#include "nspr.h"
+#include "prerror.h"
+#include "prio.h"
+#include "prnetdb.h"
+
+#include "mozilla/net/DNS.h"
+#include "nsCOMPtr.h"
+#include "nsASocketHandler.h"
+#include "nsISocketTransportService.h"
+#include "nsNetCID.h"
+#include "nsISupportsImpl.h"
+#include "nsServiceManagerUtils.h"
+#include "nsComponentManagerUtils.h"
+#include "nsXPCOM.h"
+#include "nsXULAppAPI.h"
+#include "runnable_utils.h"
+#include "mozilla/SyncRunnable.h"
+#include "nsTArray.h"
+#include "mozilla/dom/TCPSocketBinding.h"
+#include "nsITCPSocketCallback.h"
+#include "nsIPrefService.h"
+#include "nsIPrefBranch.h"
+#include "nsISocketFilter.h"
+
+#ifdef XP_WIN
+#include "mozilla/WindowsVersion.h"
+#endif
+
+#if defined(MOZILLA_INTERNAL_API)
+// csi_platform.h deep in nrappkit defines LOG_INFO and LOG_WARNING
+#ifdef LOG_INFO
+#define LOG_TEMP_INFO LOG_INFO
+#undef LOG_INFO
+#endif
+#ifdef LOG_WARNING
+#define LOG_TEMP_WARNING LOG_WARNING
+#undef LOG_WARNING
+#endif
+#if defined(LOG_DEBUG)
+#define LOG_TEMP_DEBUG LOG_DEBUG
+#undef LOG_DEBUG
+#endif
+#undef strlcpy
+
+#include "mozilla/dom/network/TCPSocketChild.h"
+
+#ifdef LOG_TEMP_INFO
+#define LOG_INFO LOG_TEMP_INFO
+#endif
+#ifdef LOG_TEMP_WARNING
+#define LOG_WARNING LOG_TEMP_WARNING
+#endif
+
+#ifdef LOG_TEMP_DEBUG
+#define LOG_DEBUG LOG_TEMP_DEBUG
+#endif
+#ifdef XP_WIN
+#ifdef LOG_DEBUG
+#undef LOG_DEBUG
+#endif
+// cloned from csi_platform.h. Win32 doesn't like how we hide symbols
+#define LOG_DEBUG 7
+#endif
+#endif
+
+
+extern "C" {
+#include "nr_api.h"
+#include "async_wait.h"
+#include "nr_socket.h"
+#include "nr_socket_local.h"
+#include "stun_hint.h"
+}
+#include "nr_socket_prsock.h"
+#include "simpletokenbucket.h"
+#include "test_nr_socket.h"
+
+// Implement the nsISupports ref counting
+namespace mozilla {
+
+#if defined(MOZILLA_INTERNAL_API)
+class SingletonThreadHolder final
+{
+private:
+ ~SingletonThreadHolder()
+ {
+ r_log(LOG_GENERIC,LOG_DEBUG,"Deleting SingletonThreadHolder");
+ MOZ_ASSERT(!mThread, "SingletonThreads should be Released and shut down before exit!");
+ if (mThread) {
+ mThread->Shutdown();
+ mThread = nullptr;
+ }
+ }
+
+ DISALLOW_COPY_ASSIGN(SingletonThreadHolder);
+
+public:
+ // Must be threadsafe for StaticRefPtr/ClearOnShutdown
+ NS_INLINE_DECL_THREADSAFE_REFCOUNTING(SingletonThreadHolder)
+
+ explicit SingletonThreadHolder(const nsCSubstring& aName)
+ : mName(aName)
+ {
+ mParentThread = NS_GetCurrentThread();
+ }
+
+ nsIThread* GetThread() {
+ return mThread;
+ }
+
+ /*
+ * Keep track of how many instances are using a SingletonThreadHolder.
+ * When no one is using it, shut it down
+ */
+ MozExternalRefCountType AddUse()
+ {
+ MOZ_ASSERT(mParentThread == NS_GetCurrentThread());
+ MOZ_ASSERT(int32_t(mUseCount) >= 0, "illegal refcnt");
+ nsrefcnt count = ++mUseCount;
+ if (count == 1) {
+ // idle -> in-use
+ nsresult rv = NS_NewThread(getter_AddRefs(mThread));
+ MOZ_RELEASE_ASSERT(NS_SUCCEEDED(rv) && mThread,
+ "Should successfully create mtransport I/O thread");
+ NS_SetThreadName(mThread, mName);
+ r_log(LOG_GENERIC,LOG_DEBUG,"Created wrapped SingletonThread %p",
+ mThread.get());
+ }
+ r_log(LOG_GENERIC,LOG_DEBUG,"AddUse: %lu", (unsigned long) count);
+ return count;
+ }
+
+ MozExternalRefCountType ReleaseUse()
+ {
+ MOZ_ASSERT(mParentThread == NS_GetCurrentThread());
+ nsrefcnt count = --mUseCount;
+ MOZ_ASSERT(int32_t(mUseCount) >= 0, "illegal refcnt");
+ if (count == 0) {
+ // in-use -> idle -- no one forcing it to remain instantiated
+ r_log(LOG_GENERIC,LOG_DEBUG,"Shutting down wrapped SingletonThread %p",
+ mThread.get());
+ mThread->Shutdown();
+ mThread = nullptr;
+ // It'd be nice to use a timer instead... But be careful of
+ // xpcom-shutdown-threads in that case
+ }
+ r_log(LOG_GENERIC,LOG_DEBUG,"ReleaseUse: %lu", (unsigned long) count);
+ return count;
+ }
+
+private:
+ nsCString mName;
+ nsAutoRefCnt mUseCount;
+ nsCOMPtr<nsIThread> mParentThread;
+ nsCOMPtr<nsIThread> mThread;
+};
+
+static StaticRefPtr<SingletonThreadHolder> sThread;
+
+static void ClearSingletonOnShutdown()
+{
+ ClearOnShutdown(&sThread);
+}
+#endif
+
+static nsIThread* GetIOThreadAndAddUse_s()
+{
+ // Always runs on STS thread!
+#if defined(MOZILLA_INTERNAL_API)
+ // We need to safely release this on shutdown to avoid leaks
+ if (!sThread) {
+ sThread = new SingletonThreadHolder(NS_LITERAL_CSTRING("mtransport"));
+ NS_DispatchToMainThread(mozilla::WrapRunnableNM(&ClearSingletonOnShutdown));
+ }
+ // Mark that we're using the shared thread and need it to stick around
+ sThread->AddUse();
+ return sThread->GetThread();
+#else
+ static nsCOMPtr<nsIThread> sThread;
+ if (!sThread) {
+ (void) NS_NewNamedThread("mtransport", getter_AddRefs(sThread));
+ }
+ return sThread;
+#endif
+}
+
+NrSocketIpc::NrSocketIpc(nsIEventTarget *aThread)
+ : io_thread_(aThread)
+{}
+
+static TimeStamp nr_socket_short_term_violation_time;
+static TimeStamp nr_socket_long_term_violation_time;
+
+TimeStamp NrSocketBase::short_term_violation_time() {
+ return nr_socket_short_term_violation_time;
+}
+
+TimeStamp NrSocketBase::long_term_violation_time() {
+ return nr_socket_long_term_violation_time;
+}
+
+// NrSocketBase implementation
+// async_event APIs
+int NrSocketBase::async_wait(int how, NR_async_cb cb, void *cb_arg,
+ char *function, int line) {
+ uint16_t flag;
+
+ switch (how) {
+ case NR_ASYNC_WAIT_READ:
+ flag = PR_POLL_READ;
+ break;
+ case NR_ASYNC_WAIT_WRITE:
+ flag = PR_POLL_WRITE;
+ break;
+ default:
+ return R_BAD_ARGS;
+ }
+
+ cbs_[how] = cb;
+ cb_args_[how] = cb_arg;
+ poll_flags_ |= flag;
+
+ return 0;
+}
+
+int NrSocketBase::cancel(int how) {
+ uint16_t flag;
+
+ switch (how) {
+ case NR_ASYNC_WAIT_READ:
+ flag = PR_POLL_READ;
+ break;
+ case NR_ASYNC_WAIT_WRITE:
+ flag = PR_POLL_WRITE;
+ break;
+ default:
+ return R_BAD_ARGS;
+ }
+
+ poll_flags_ &= ~flag;
+
+ return 0;
+}
+
+void NrSocketBase::fire_callback(int how) {
+ // This can't happen unless we are armed because we only set
+ // the flags if we are armed
+ MOZ_ASSERT(cbs_[how]);
+
+ // Now cancel so that we need to be re-armed. Note that
+ // the re-arming probably happens in the callback we are
+ // about to fire.
+ cancel(how);
+
+ cbs_[how](this, how, cb_args_[how]);
+}
+
+// NrSocket implementation
+NS_IMPL_ISUPPORTS0(NrSocket)
+
+
+// The nsASocket callbacks
+void NrSocket::OnSocketReady(PRFileDesc *fd, int16_t outflags) {
+ if (outflags & PR_POLL_READ & poll_flags())
+ fire_callback(NR_ASYNC_WAIT_READ);
+ if (outflags & PR_POLL_WRITE & poll_flags())
+ fire_callback(NR_ASYNC_WAIT_WRITE);
+ if (outflags & (PR_POLL_ERR | PR_POLL_NVAL | PR_POLL_HUP))
+ // TODO: Bug 946423: how do we notify the upper layers about this?
+ close();
+}
+
+void NrSocket::OnSocketDetached(PRFileDesc *fd) {
+ r_log(LOG_GENERIC, LOG_DEBUG, "Socket %p detached", fd);
+}
+
+void NrSocket::IsLocal(bool *aIsLocal) {
+ // TODO(jesup): better check? Does it matter? (likely no)
+ *aIsLocal = false;
+}
+
+// async_event APIs
+int NrSocket::async_wait(int how, NR_async_cb cb, void *cb_arg,
+ char *function, int line) {
+ int r = NrSocketBase::async_wait(how, cb, cb_arg, function, line);
+
+ if (!r) {
+ mPollFlags = poll_flags();
+ }
+
+ return r;
+}
+
+int NrSocket::cancel(int how) {
+ int r = NrSocketBase::cancel(how);
+
+ if (!r) {
+ mPollFlags = poll_flags();
+ }
+
+ return r;
+}
+
+// Helper functions for addresses
+static int nr_transport_addr_to_praddr(nr_transport_addr *addr,
+ PRNetAddr *naddr)
+ {
+ int _status;
+
+ memset(naddr, 0, sizeof(*naddr));
+
+ switch(addr->protocol){
+ case IPPROTO_TCP:
+ break;
+ case IPPROTO_UDP:
+ break;
+ default:
+ ABORT(R_BAD_ARGS);
+ }
+
+ switch(addr->ip_version){
+ case NR_IPV4:
+ naddr->inet.family = PR_AF_INET;
+ naddr->inet.port = addr->u.addr4.sin_port;
+ naddr->inet.ip = addr->u.addr4.sin_addr.s_addr;
+ break;
+ case NR_IPV6:
+ naddr->ipv6.family = PR_AF_INET6;
+ naddr->ipv6.port = addr->u.addr6.sin6_port;
+ naddr->ipv6.flowinfo = addr->u.addr6.sin6_flowinfo;
+ memcpy(&naddr->ipv6.ip, &addr->u.addr6.sin6_addr, sizeof(in6_addr));
+ naddr->ipv6.scope_id = addr->u.addr6.sin6_scope_id;
+ break;
+ default:
+ ABORT(R_BAD_ARGS);
+ }
+
+ _status = 0;
+ abort:
+ return(_status);
+ }
+
+//XXX schien@mozilla.com: copy from PRNetAddrToNetAddr,
+// should be removed after fix the link error in signaling_unittests
+static int praddr_to_netaddr(const PRNetAddr *prAddr, net::NetAddr *addr)
+{
+ int _status;
+
+ switch (prAddr->raw.family) {
+ case PR_AF_INET:
+ addr->inet.family = AF_INET;
+ addr->inet.port = prAddr->inet.port;
+ addr->inet.ip = prAddr->inet.ip;
+ break;
+ case PR_AF_INET6:
+ addr->inet6.family = AF_INET6;
+ addr->inet6.port = prAddr->ipv6.port;
+ addr->inet6.flowinfo = prAddr->ipv6.flowinfo;
+ memcpy(&addr->inet6.ip, &prAddr->ipv6.ip, sizeof(addr->inet6.ip.u8));
+ addr->inet6.scope_id = prAddr->ipv6.scope_id;
+ break;
+ default:
+ MOZ_ASSERT(false);
+ ABORT(R_BAD_ARGS);
+ }
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+static int nr_transport_addr_to_netaddr(nr_transport_addr *addr,
+ net::NetAddr *naddr)
+{
+ int r, _status;
+ PRNetAddr praddr;
+
+ if((r = nr_transport_addr_to_praddr(addr, &praddr))) {
+ ABORT(r);
+ }
+
+ if((r = praddr_to_netaddr(&praddr, naddr))) {
+ ABORT(r);
+ }
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int nr_netaddr_to_transport_addr(const net::NetAddr *netaddr,
+ nr_transport_addr *addr, int protocol)
+ {
+ int _status;
+ int r;
+
+ switch(netaddr->raw.family) {
+ case AF_INET:
+ if ((r = nr_ip4_port_to_transport_addr(ntohl(netaddr->inet.ip),
+ ntohs(netaddr->inet.port),
+ protocol, addr)))
+ ABORT(r);
+ break;
+ case AF_INET6:
+ if ((r = nr_ip6_port_to_transport_addr((in6_addr *)&netaddr->inet6.ip.u8,
+ ntohs(netaddr->inet6.port),
+ protocol, addr)))
+ ABORT(r);
+ break;
+ default:
+ MOZ_ASSERT(false);
+ ABORT(R_BAD_ARGS);
+ }
+ _status = 0;
+ abort:
+ return(_status);
+ }
+
+int nr_praddr_to_transport_addr(const PRNetAddr *praddr,
+ nr_transport_addr *addr, int protocol,
+ int keep)
+ {
+ int _status;
+ int r;
+ struct sockaddr_in ip4;
+ struct sockaddr_in6 ip6;
+
+ switch(praddr->raw.family) {
+ case PR_AF_INET:
+ ip4.sin_family = PF_INET;
+ ip4.sin_addr.s_addr = praddr->inet.ip;
+ ip4.sin_port = praddr->inet.port;
+ if ((r = nr_sockaddr_to_transport_addr((sockaddr *)&ip4,
+ protocol, keep,
+ addr)))
+ ABORT(r);
+ break;
+ case PR_AF_INET6:
+ ip6.sin6_family = PF_INET6;
+ ip6.sin6_port = praddr->ipv6.port;
+ ip6.sin6_flowinfo = praddr->ipv6.flowinfo;
+ memcpy(&ip6.sin6_addr, &praddr->ipv6.ip, sizeof(in6_addr));
+ ip6.sin6_scope_id = praddr->ipv6.scope_id;
+ if ((r = nr_sockaddr_to_transport_addr((sockaddr *)&ip6,protocol,keep,addr)))
+ ABORT(r);
+ break;
+ default:
+ MOZ_ASSERT(false);
+ ABORT(R_BAD_ARGS);
+ }
+
+ _status = 0;
+ abort:
+ return(_status);
+ }
+
+/*
+ * nr_transport_addr_get_addrstring_and_port
+ * convert nr_transport_addr to IP address string and port number
+ */
+int nr_transport_addr_get_addrstring_and_port(nr_transport_addr *addr,
+ nsACString *host, int32_t *port) {
+ int r, _status;
+ char addr_string[64];
+
+ // We cannot directly use |nr_transport_addr.as_string| because it contains
+ // more than ip address, therefore, we need to explicity convert it
+ // from |nr_transport_addr_get_addrstring|.
+ if ((r=nr_transport_addr_get_addrstring(addr, addr_string, sizeof(addr_string)))) {
+ ABORT(r);
+ }
+
+ if ((r=nr_transport_addr_get_port(addr, port))) {
+ ABORT(r);
+ }
+
+ *host = addr_string;
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+// nr_socket APIs (as member functions)
+int NrSocket::create(nr_transport_addr *addr) {
+ int r,_status;
+
+ PRStatus status;
+ PRNetAddr naddr;
+
+ nsresult rv;
+ nsCOMPtr<nsISocketTransportService> stservice =
+ do_GetService(NS_SOCKETTRANSPORTSERVICE_CONTRACTID, &rv);
+
+ if (!NS_SUCCEEDED(rv)) {
+ ABORT(R_INTERNAL);
+ }
+
+ if((r=nr_transport_addr_to_praddr(addr, &naddr)))
+ ABORT(r);
+
+ switch (addr->protocol) {
+ case IPPROTO_UDP:
+ if (!(fd_ = PR_OpenUDPSocket(naddr.raw.family))) {
+ r_log(LOG_GENERIC,LOG_CRIT,"Couldn't create UDP socket, "
+ "family=%d, err=%d", naddr.raw.family, PR_GetError());
+ ABORT(R_INTERNAL);
+ }
+#ifdef XP_WIN
+ if (!mozilla::IsWin8OrLater()) {
+ // Increase default send and receive buffer sizes on <= Win7 to be able to
+ // receive and send an unpaced HD (>= 720p = 1280x720 - I Frame ~ 21K size)
+ // stream without losing packets.
+ // Manual testing showed that 100K buffer size was not enough and the
+ // packet loss dis-appeared with 256K buffer size.
+ // See bug 1252769 for future improvements of this.
+ PRSize min_buffer_size = 256 * 1024;
+ PRSocketOptionData opt_rcvbuf;
+ opt_rcvbuf.option = PR_SockOpt_RecvBufferSize;
+ if ((status = PR_GetSocketOption(fd_, &opt_rcvbuf)) == PR_SUCCESS) {
+ if (opt_rcvbuf.value.recv_buffer_size < min_buffer_size) {
+ opt_rcvbuf.value.recv_buffer_size = min_buffer_size;
+ if ((status = PR_SetSocketOption(fd_, &opt_rcvbuf)) != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_CRIT,
+ "Couldn't set socket receive buffer size: %d", status);
+ }
+ } else {
+ r_log(LOG_GENERIC, LOG_INFO,
+ "Socket receive buffer size is already: %d",
+ opt_rcvbuf.value.recv_buffer_size);
+ }
+ } else {
+ r_log(LOG_GENERIC, LOG_CRIT,
+ "Couldn't get socket receive buffer size: %d", status);
+ }
+ PRSocketOptionData opt_sndbuf;
+ opt_sndbuf.option = PR_SockOpt_SendBufferSize;
+ if ((status = PR_GetSocketOption(fd_, &opt_sndbuf)) == PR_SUCCESS) {
+ if (opt_sndbuf.value.recv_buffer_size < min_buffer_size) {
+ opt_sndbuf.value.recv_buffer_size = min_buffer_size;
+ if ((status = PR_SetSocketOption(fd_, &opt_sndbuf)) != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_CRIT,
+ "Couldn't set socket send buffer size: %d", status);
+ }
+ } else {
+ r_log(LOG_GENERIC, LOG_INFO,
+ "Socket send buffer size is already: %d",
+ opt_sndbuf.value.recv_buffer_size);
+ }
+ } else {
+ r_log(LOG_GENERIC, LOG_CRIT,
+ "Couldn't get socket send buffer size: %d", status);
+ }
+ }
+#endif
+ break;
+ case IPPROTO_TCP:
+ if (!(fd_ = PR_OpenTCPSocket(naddr.raw.family))) {
+ r_log(LOG_GENERIC,LOG_CRIT,"Couldn't create TCP socket, "
+ "family=%d, err=%d", naddr.raw.family, PR_GetError());
+ ABORT(R_INTERNAL);
+ }
+ // Set ReuseAddr for TCP sockets to enable having several
+ // sockets bound to same local IP and port
+ PRSocketOptionData opt_reuseaddr;
+ opt_reuseaddr.option = PR_SockOpt_Reuseaddr;
+ opt_reuseaddr.value.reuse_addr = PR_TRUE;
+ status = PR_SetSocketOption(fd_, &opt_reuseaddr);
+ if (status != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_CRIT,
+ "Couldn't set reuse addr socket option: %d", status);
+ ABORT(R_INTERNAL);
+ }
+ // And also set ReusePort for platforms supporting this socket option
+ PRSocketOptionData opt_reuseport;
+ opt_reuseport.option = PR_SockOpt_Reuseport;
+ opt_reuseport.value.reuse_port = PR_TRUE;
+ status = PR_SetSocketOption(fd_, &opt_reuseport);
+ if (status != PR_SUCCESS) {
+ if (PR_GetError() != PR_OPERATION_NOT_SUPPORTED_ERROR) {
+ r_log(LOG_GENERIC, LOG_CRIT,
+ "Couldn't set reuse port socket option: %d", status);
+ ABORT(R_INTERNAL);
+ }
+ }
+ // Try to speedup packet delivery by disabling TCP Nagle
+ PRSocketOptionData opt_nodelay;
+ opt_nodelay.option = PR_SockOpt_NoDelay;
+ opt_nodelay.value.no_delay = PR_TRUE;
+ status = PR_SetSocketOption(fd_, &opt_nodelay);
+ if (status != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_WARNING,
+ "Couldn't set Nodelay socket option: %d", status);
+ }
+ break;
+ default:
+ ABORT(R_INTERNAL);
+ }
+
+ status = PR_Bind(fd_, &naddr);
+ if (status != PR_SUCCESS) {
+ r_log(LOG_GENERIC,LOG_CRIT,"Couldn't bind socket to address %s",
+ addr->as_string);
+ ABORT(R_INTERNAL);
+ }
+
+ r_log(LOG_GENERIC,LOG_DEBUG,"Creating socket %p with addr %s",
+ fd_, addr->as_string);
+ nr_transport_addr_copy(&my_addr_,addr);
+
+ /* If we have a wildcard port, patch up the addr */
+ if(nr_transport_addr_is_wildcard(addr)){
+ status = PR_GetSockName(fd_, &naddr);
+ if (status != PR_SUCCESS){
+ r_log(LOG_GENERIC, LOG_CRIT, "Couldn't get sock name for socket");
+ ABORT(R_INTERNAL);
+ }
+
+ if((r=nr_praddr_to_transport_addr(&naddr,&my_addr_,addr->protocol,1)))
+ ABORT(r);
+ }
+
+
+ // Set nonblocking
+ PRSocketOptionData opt_nonblock;
+ opt_nonblock.option = PR_SockOpt_Nonblocking;
+ opt_nonblock.value.non_blocking = PR_TRUE;
+ status = PR_SetSocketOption(fd_, &opt_nonblock);
+ if (status != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_CRIT, "Couldn't make socket nonblocking");
+ ABORT(R_INTERNAL);
+ }
+
+ // Remember our thread.
+ ststhread_ = do_QueryInterface(stservice, &rv);
+ if (!NS_SUCCEEDED(rv))
+ ABORT(R_INTERNAL);
+
+ // Finally, register with the STS
+ rv = stservice->AttachSocket(fd_, this);
+ if (!NS_SUCCEEDED(rv)) {
+ r_log(LOG_GENERIC, LOG_CRIT, "Couldn't attach socket to STS, rv=%u",
+ static_cast<unsigned>(rv));
+ ABORT(R_INTERNAL);
+ }
+
+ _status = 0;
+
+abort:
+ return(_status);
+}
+
+static int ShouldDrop(size_t len) {
+ // Global rate limiting for stun requests, to mitigate the ice hammer DoS
+ // (see http://tools.ietf.org/html/draft-thomson-mmusic-ice-webrtc)
+
+ // Tolerate rate of 8k/sec, for one second.
+ static SimpleTokenBucket burst(16384*1, 16384);
+ // Tolerate rate of 7.2k/sec over twenty seconds.
+ static SimpleTokenBucket sustained(7372*20, 7372);
+
+ // Check number of tokens in each bucket.
+ if (burst.getTokens(UINT32_MAX) < len) {
+ r_log(LOG_GENERIC, LOG_ERR,
+ "Short term global rate limit for STUN requests exceeded.");
+#ifdef MOZILLA_INTERNAL_API
+ nr_socket_short_term_violation_time = TimeStamp::Now();
+#endif
+
+// Bug 1013007
+#if !EARLY_BETA_OR_EARLIER
+ return R_WOULDBLOCK;
+#else
+ MOZ_ASSERT(false,
+ "Short term global rate limit for STUN requests exceeded. Go "
+ "bug bcampen@mozilla.com if you weren't intentionally "
+ "spamming ICE candidates, or don't know what that means.");
+#endif
+ }
+
+ if (sustained.getTokens(UINT32_MAX) < len) {
+ r_log(LOG_GENERIC, LOG_ERR,
+ "Long term global rate limit for STUN requests exceeded.");
+#ifdef MOZILLA_INTERNAL_API
+ nr_socket_long_term_violation_time = TimeStamp::Now();
+#endif
+// Bug 1013007
+#if !EARLY_BETA_OR_EARLIER
+ return R_WOULDBLOCK;
+#else
+ MOZ_ASSERT(false,
+ "Long term global rate limit for STUN requests exceeded. Go "
+ "bug bcampen@mozilla.com if you weren't intentionally "
+ "spamming ICE candidates, or don't know what that means.");
+#endif
+ }
+
+ // Take len tokens from both buckets.
+ // (not threadsafe, but no problem since this is only called from STS)
+ burst.getTokens(len);
+ sustained.getTokens(len);
+ return 0;
+}
+
+// This should be called on the STS thread.
+int NrSocket::sendto(const void *msg, size_t len,
+ int flags, nr_transport_addr *to) {
+ ASSERT_ON_THREAD(ststhread_);
+ int r,_status;
+ PRNetAddr naddr;
+ int32_t status;
+
+ if ((r=nr_transport_addr_to_praddr(to, &naddr)))
+ ABORT(r);
+
+ if(fd_==nullptr)
+ ABORT(R_EOD);
+
+ if (nr_is_stun_request_message((UCHAR*)msg, len) && ShouldDrop(len)) {
+ ABORT(R_WOULDBLOCK);
+ }
+
+ // TODO: Convert flags?
+ status = PR_SendTo(fd_, msg, len, flags, &naddr, PR_INTERVAL_NO_WAIT);
+ if (status < 0 || (size_t)status != len) {
+ if (PR_GetError() == PR_WOULD_BLOCK_ERROR)
+ ABORT(R_WOULDBLOCK);
+
+ r_log(LOG_GENERIC, LOG_INFO, "Error in sendto %s: %d",
+ to->as_string, PR_GetError());
+ ABORT(R_IO_ERROR);
+ }
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int NrSocket::recvfrom(void * buf, size_t maxlen,
+ size_t *len, int flags,
+ nr_transport_addr *from) {
+ ASSERT_ON_THREAD(ststhread_);
+ int r,_status;
+ PRNetAddr nfrom;
+ int32_t status;
+
+ status = PR_RecvFrom(fd_, buf, maxlen, flags, &nfrom, PR_INTERVAL_NO_WAIT);
+ if (status <= 0) {
+ if (PR_GetError() == PR_WOULD_BLOCK_ERROR)
+ ABORT(R_WOULDBLOCK);
+ r_log(LOG_GENERIC, LOG_INFO, "Error in recvfrom: %d", (int)PR_GetError());
+ ABORT(R_IO_ERROR);
+ }
+ *len = status;
+
+ if((r=nr_praddr_to_transport_addr(&nfrom,from,my_addr_.protocol,0)))
+ ABORT(r);
+
+ //r_log(LOG_GENERIC,LOG_DEBUG,"Read %d bytes from %s",*len,addr->as_string);
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int NrSocket::getaddr(nr_transport_addr *addrp) {
+ ASSERT_ON_THREAD(ststhread_);
+ return nr_transport_addr_copy(addrp, &my_addr_);
+}
+
+// Close the socket so that the STS will detach and then kill it
+void NrSocket::close() {
+ ASSERT_ON_THREAD(ststhread_);
+ mCondition = NS_BASE_STREAM_CLOSED;
+}
+
+
+int NrSocket::connect(nr_transport_addr *addr) {
+ ASSERT_ON_THREAD(ststhread_);
+ int r,_status;
+ PRNetAddr naddr;
+ int32_t connect_status, getsockname_status;
+
+ if ((r=nr_transport_addr_to_praddr(addr, &naddr)))
+ ABORT(r);
+
+ if(!fd_)
+ ABORT(R_EOD);
+
+ // Note: this just means we tried to connect, not that we
+ // are actually live.
+ connect_invoked_ = true;
+ connect_status = PR_Connect(fd_, &naddr, PR_INTERVAL_NO_WAIT);
+ if (connect_status != PR_SUCCESS) {
+ if (PR_GetError() != PR_IN_PROGRESS_ERROR)
+ ABORT(R_IO_ERROR);
+ }
+
+ // If our local address is wildcard, then fill in the
+ // address now.
+ if(nr_transport_addr_is_wildcard(&my_addr_)){
+ getsockname_status = PR_GetSockName(fd_, &naddr);
+ if (getsockname_status != PR_SUCCESS){
+ r_log(LOG_GENERIC, LOG_CRIT, "Couldn't get sock name for socket");
+ ABORT(R_INTERNAL);
+ }
+
+ if((r=nr_praddr_to_transport_addr(&naddr,&my_addr_,addr->protocol,1)))
+ ABORT(r);
+ }
+
+ // Now return the WOULDBLOCK if needed.
+ if (connect_status != PR_SUCCESS) {
+ ABORT(R_WOULDBLOCK);
+ }
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+
+int NrSocket::write(const void *msg, size_t len, size_t *written) {
+ ASSERT_ON_THREAD(ststhread_);
+ int _status;
+ int32_t status;
+
+ if (!connect_invoked_)
+ ABORT(R_FAILED);
+
+ status = PR_Write(fd_, msg, len);
+ if (status < 0) {
+ if (PR_GetError() == PR_WOULD_BLOCK_ERROR)
+ ABORT(R_WOULDBLOCK);
+ r_log(LOG_GENERIC, LOG_INFO, "Error in write");
+ ABORT(R_IO_ERROR);
+ }
+
+ *written = status;
+
+ _status = 0;
+abort:
+ return _status;
+}
+
+int NrSocket::read(void* buf, size_t maxlen, size_t *len) {
+ ASSERT_ON_THREAD(ststhread_);
+ int _status;
+ int32_t status;
+
+ if (!connect_invoked_)
+ ABORT(R_FAILED);
+
+ status = PR_Read(fd_, buf, maxlen);
+ if (status < 0) {
+ if (PR_GetError() == PR_WOULD_BLOCK_ERROR)
+ ABORT(R_WOULDBLOCK);
+ r_log(LOG_GENERIC, LOG_INFO, "Error in read");
+ ABORT(R_IO_ERROR);
+ }
+ if (status == 0)
+ ABORT(R_EOD);
+
+ *len = (size_t)status; // Guaranteed to be > 0
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int NrSocket::listen(int backlog) {
+ ASSERT_ON_THREAD(ststhread_);
+ int32_t status;
+ int _status;
+
+ assert(fd_);
+ status = PR_Listen(fd_, backlog);
+ if (status != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_CRIT, "%s: PR_GetError() == %d",
+ __FUNCTION__, PR_GetError());
+ ABORT(R_IO_ERROR);
+ }
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int NrSocket::accept(nr_transport_addr *addrp, nr_socket **sockp) {
+ ASSERT_ON_THREAD(ststhread_);
+ int _status, r;
+ PRStatus status;
+ PRFileDesc *prfd;
+ PRNetAddr nfrom;
+ NrSocket *sock=nullptr;
+ nsresult rv;
+ PRSocketOptionData opt_nonblock, opt_nodelay;
+ nsCOMPtr<nsISocketTransportService> stservice =
+ do_GetService(NS_SOCKETTRANSPORTSERVICE_CONTRACTID, &rv);
+
+ if (NS_FAILED(rv)) {
+ ABORT(R_INTERNAL);
+ }
+
+ if(!fd_)
+ ABORT(R_EOD);
+
+ prfd = PR_Accept(fd_, &nfrom, PR_INTERVAL_NO_WAIT);
+
+ if (!prfd) {
+ if (PR_GetError() == PR_WOULD_BLOCK_ERROR)
+ ABORT(R_WOULDBLOCK);
+
+ ABORT(R_IO_ERROR);
+ }
+
+ sock = new NrSocket();
+
+ sock->fd_=prfd;
+ nr_transport_addr_copy(&sock->my_addr_, &my_addr_);
+
+ if((r=nr_praddr_to_transport_addr(&nfrom, addrp, my_addr_.protocol, 0)))
+ ABORT(r);
+
+ // Set nonblocking
+ opt_nonblock.option = PR_SockOpt_Nonblocking;
+ opt_nonblock.value.non_blocking = PR_TRUE;
+ status = PR_SetSocketOption(prfd, &opt_nonblock);
+ if (status != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_CRIT,
+ "Failed to make accepted socket nonblocking: %d", status);
+ ABORT(R_INTERNAL);
+ }
+ // Disable TCP Nagle
+ opt_nodelay.option = PR_SockOpt_NoDelay;
+ opt_nodelay.value.no_delay = PR_TRUE;
+ status = PR_SetSocketOption(prfd, &opt_nodelay);
+ if (status != PR_SUCCESS) {
+ r_log(LOG_GENERIC, LOG_WARNING,
+ "Failed to set Nodelay on accepted socket: %d", status);
+ }
+
+ // Should fail only with OOM
+ if ((r=nr_socket_create_int(static_cast<void *>(sock), sock->vtbl(), sockp)))
+ ABORT(r);
+
+ // Remember our thread.
+ sock->ststhread_ = do_QueryInterface(stservice, &rv);
+ if (NS_FAILED(rv))
+ ABORT(R_INTERNAL);
+
+ // Finally, register with the STS
+ rv = stservice->AttachSocket(prfd, sock);
+ if (NS_FAILED(rv)) {
+ ABORT(R_INTERNAL);
+ }
+
+ sock->connect_invoked_ = true;
+
+ // Add a reference so that we can delete it in destroy()
+ sock->AddRef();
+ _status = 0;
+abort:
+ if (_status) {
+ delete sock;
+ }
+
+ return(_status);
+}
+
+NS_IMPL_ISUPPORTS(NrUdpSocketIpcProxy, nsIUDPSocketInternal)
+
+nsresult
+NrUdpSocketIpcProxy::Init(const RefPtr<NrUdpSocketIpc>& socket)
+{
+ nsresult rv;
+ sts_thread_ = do_GetService(NS_SOCKETTRANSPORTSERVICE_CONTRACTID, &rv);
+ if (NS_FAILED(rv)) {
+ MOZ_ASSERT(false, "Failed to get STS thread");
+ return rv;
+ }
+
+ socket_ = socket;
+ return NS_OK;
+}
+
+NrUdpSocketIpcProxy::~NrUdpSocketIpcProxy()
+{
+ // Send our ref to STS to be released
+ RUN_ON_THREAD(sts_thread_,
+ mozilla::WrapRelease(socket_.forget()),
+ NS_DISPATCH_NORMAL);
+}
+
+// IUDPSocketInternal interfaces
+// callback while error happened in UDP socket operation
+NS_IMETHODIMP NrUdpSocketIpcProxy::CallListenerError(const nsACString &message,
+ const nsACString &filename,
+ uint32_t line_number) {
+ return socket_->CallListenerError(message, filename, line_number);
+}
+
+// callback while receiving UDP packet
+NS_IMETHODIMP NrUdpSocketIpcProxy::CallListenerReceivedData(const nsACString &host,
+ uint16_t port,
+ const uint8_t *data,
+ uint32_t data_length) {
+ return socket_->CallListenerReceivedData(host, port, data, data_length);
+}
+
+// callback while UDP socket is opened
+NS_IMETHODIMP NrUdpSocketIpcProxy::CallListenerOpened() {
+ return socket_->CallListenerOpened();
+}
+
+// callback while UDP socket is connected
+NS_IMETHODIMP NrUdpSocketIpcProxy::CallListenerConnected() {
+ return socket_->CallListenerConnected();
+}
+
+// callback while UDP socket is closed
+NS_IMETHODIMP NrUdpSocketIpcProxy::CallListenerClosed() {
+ return socket_->CallListenerClosed();
+}
+
+// NrUdpSocketIpc Implementation
+NrUdpSocketIpc::NrUdpSocketIpc()
+ : NrSocketIpc(GetIOThreadAndAddUse_s()),
+ monitor_("NrUdpSocketIpc"),
+ err_(false),
+ state_(NR_INIT) {
+}
+
+NrUdpSocketIpc::~NrUdpSocketIpc()
+{
+ // also guarantees socket_child_ is released from the io_thread, and
+ // tells the SingletonThreadHolder we're done with it
+
+#if defined(MOZILLA_INTERNAL_API)
+ // close(), but transfer the socket_child_ reference to die as well
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnableNM(&NrUdpSocketIpc::release_child_i,
+ socket_child_.forget().take(),
+ sts_thread_),
+ NS_DISPATCH_NORMAL);
+#endif
+}
+
+// IUDPSocketInternal interfaces
+// callback while error happened in UDP socket operation
+NS_IMETHODIMP NrUdpSocketIpc::CallListenerError(const nsACString &message,
+ const nsACString &filename,
+ uint32_t line_number) {
+ ASSERT_ON_THREAD(io_thread_);
+
+ r_log(LOG_GENERIC, LOG_ERR, "UDP socket error:%s at %s:%d this=%p",
+ message.BeginReading(), filename.BeginReading(), line_number, (void*) this );
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+ err_ = true;
+ monitor_.NotifyAll();
+
+ return NS_OK;
+}
+
+// callback while receiving UDP packet
+NS_IMETHODIMP NrUdpSocketIpc::CallListenerReceivedData(const nsACString &host,
+ uint16_t port,
+ const uint8_t *data,
+ uint32_t data_length) {
+ ASSERT_ON_THREAD(io_thread_);
+
+ PRNetAddr addr;
+ memset(&addr, 0, sizeof(addr));
+
+ {
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ if (PR_SUCCESS != PR_StringToNetAddr(host.BeginReading(), &addr)) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to convert remote host to PRNetAddr");
+ return NS_OK;
+ }
+
+ // Use PR_IpAddrNull to avoid address being reset to 0.
+ if (PR_SUCCESS != PR_SetNetAddr(PR_IpAddrNull, addr.raw.family, port, &addr)) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to set port in PRNetAddr");
+ return NS_OK;
+ }
+ }
+
+ nsAutoPtr<DataBuffer> buf(new DataBuffer(data, data_length));
+ RefPtr<nr_udp_message> msg(new nr_udp_message(addr, buf));
+
+ RUN_ON_THREAD(sts_thread_,
+ mozilla::WrapRunnable(RefPtr<NrUdpSocketIpc>(this),
+ &NrUdpSocketIpc::recv_callback_s,
+ msg),
+ NS_DISPATCH_NORMAL);
+ return NS_OK;
+}
+
+nsresult NrUdpSocketIpc::SetAddress() {
+ uint16_t port;
+ if (NS_FAILED(socket_child_->GetLocalPort(&port))) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to get local port");
+ return NS_OK;
+ }
+
+ nsAutoCString address;
+ if(NS_FAILED(socket_child_->GetLocalAddress(address))) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to get local address");
+ return NS_OK;
+ }
+
+ PRNetAddr praddr;
+ if (PR_SUCCESS != PR_InitializeNetAddr(PR_IpAddrAny, port, &praddr)) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to set port in PRNetAddr");
+ return NS_OK;
+ }
+
+ if (PR_SUCCESS != PR_StringToNetAddr(address.BeginReading(), &praddr)) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to convert local host to PRNetAddr");
+ return NS_OK;
+ }
+
+ nr_transport_addr expected_addr;
+ if(nr_transport_addr_copy(&expected_addr, &my_addr_)) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to copy my_addr_");
+ }
+
+ if (nr_praddr_to_transport_addr(&praddr, &my_addr_, IPPROTO_UDP, 1)) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to copy local host to my_addr_");
+ }
+
+ if (!nr_transport_addr_is_wildcard(&expected_addr) &&
+ nr_transport_addr_cmp(&expected_addr, &my_addr_,
+ NR_TRANSPORT_ADDR_CMP_MODE_ADDR)) {
+ err_ = true;
+ MOZ_ASSERT(false, "Address of opened socket is not expected");
+ }
+
+ return NS_OK;
+}
+
+// callback while UDP socket is opened
+NS_IMETHODIMP NrUdpSocketIpc::CallListenerOpened() {
+ ASSERT_ON_THREAD(io_thread_);
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ r_log(LOG_GENERIC, LOG_DEBUG, "UDP socket opened this=%p", (void*) this);
+ nsresult rv = SetAddress();
+ if (NS_FAILED(rv)) {
+ return rv;
+ }
+
+ mon.NotifyAll();
+
+ return NS_OK;
+}
+
+// callback while UDP socket is connected
+NS_IMETHODIMP NrUdpSocketIpc::CallListenerConnected() {
+ ASSERT_ON_THREAD(io_thread_);
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ r_log(LOG_GENERIC, LOG_DEBUG, "UDP socket connected this=%p", (void*) this);
+ MOZ_ASSERT(state_ == NR_CONNECTED);
+
+ nsresult rv = SetAddress();
+ if (NS_FAILED(rv)) {
+ mon.NotifyAll();
+ return rv;
+ }
+
+ r_log(LOG_GENERIC, LOG_INFO, "Exit UDP socket connected");
+ mon.NotifyAll();
+
+ return NS_OK;
+}
+
+// callback while UDP socket is closed
+NS_IMETHODIMP NrUdpSocketIpc::CallListenerClosed() {
+ ASSERT_ON_THREAD(io_thread_);
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ r_log(LOG_GENERIC, LOG_DEBUG, "UDP socket closed this=%p", (void*) this);
+ MOZ_ASSERT(state_ == NR_CONNECTED || state_ == NR_CLOSING);
+ state_ = NR_CLOSED;
+
+ return NS_OK;
+}
+
+//
+// NrSocketBase methods.
+//
+int NrUdpSocketIpc::create(nr_transport_addr *addr) {
+ ASSERT_ON_THREAD(sts_thread_);
+
+ int r, _status;
+ nsresult rv;
+ int32_t port;
+ nsCString host;
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ if (state_ != NR_INIT) {
+ ABORT(R_INTERNAL);
+ }
+
+ sts_thread_ = do_GetService(NS_SOCKETTRANSPORTSERVICE_CONTRACTID, &rv);
+ if (NS_FAILED(rv)) {
+ MOZ_ASSERT(false, "Failed to get STS thread");
+ ABORT(R_INTERNAL);
+ }
+
+ if ((r=nr_transport_addr_get_addrstring_and_port(addr, &host, &port))) {
+ ABORT(r);
+ }
+
+ // wildcard address will be resolved at NrUdpSocketIpc::CallListenerVoid
+ if ((r=nr_transport_addr_copy(&my_addr_, addr))) {
+ ABORT(r);
+ }
+
+ state_ = NR_CONNECTING;
+
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnable(RefPtr<NrUdpSocketIpc>(this),
+ &NrUdpSocketIpc::create_i,
+ host, static_cast<uint16_t>(port)),
+ NS_DISPATCH_NORMAL);
+
+ // Wait until socket creation complete.
+ mon.Wait();
+
+ if (err_) {
+ close();
+ ABORT(R_INTERNAL);
+ }
+
+ state_ = NR_CONNECTED;
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int NrUdpSocketIpc::sendto(const void *msg, size_t len, int flags,
+ nr_transport_addr *to) {
+ ASSERT_ON_THREAD(sts_thread_);
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ //If send err happened before, simply return the error.
+ if (err_) {
+ return R_IO_ERROR;
+ }
+
+ if (state_ != NR_CONNECTED) {
+ return R_INTERNAL;
+ }
+
+ int r;
+ net::NetAddr addr;
+ if ((r=nr_transport_addr_to_netaddr(to, &addr))) {
+ return r;
+ }
+
+ if (nr_is_stun_request_message((UCHAR*)msg, len) && ShouldDrop(len)) {
+ return R_WOULDBLOCK;
+ }
+
+ nsAutoPtr<DataBuffer> buf(new DataBuffer(static_cast<const uint8_t*>(msg), len));
+
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnable(RefPtr<NrUdpSocketIpc>(this),
+ &NrUdpSocketIpc::sendto_i,
+ addr, buf),
+ NS_DISPATCH_NORMAL);
+ return 0;
+}
+
+void NrUdpSocketIpc::close() {
+ r_log(LOG_GENERIC, LOG_DEBUG, "NrUdpSocketIpc::close()");
+
+ ASSERT_ON_THREAD(sts_thread_);
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+ state_ = NR_CLOSING;
+
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnable(RefPtr<NrUdpSocketIpc>(this),
+ &NrUdpSocketIpc::close_i),
+ NS_DISPATCH_NORMAL);
+
+ //remove all enqueued messages
+ std::queue<RefPtr<nr_udp_message> > empty;
+ std::swap(received_msgs_, empty);
+}
+
+int NrUdpSocketIpc::recvfrom(void *buf, size_t maxlen, size_t *len, int flags,
+ nr_transport_addr *from) {
+ ASSERT_ON_THREAD(sts_thread_);
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ int r, _status;
+ uint32_t consumed_len;
+
+ *len = 0;
+
+ if (state_ != NR_CONNECTED) {
+ ABORT(R_INTERNAL);
+ }
+
+ if (received_msgs_.empty()) {
+ ABORT(R_WOULDBLOCK);
+ }
+
+ {
+ RefPtr<nr_udp_message> msg(received_msgs_.front());
+
+ received_msgs_.pop();
+
+ if ((r=nr_praddr_to_transport_addr(&msg->from, from, IPPROTO_UDP, 0))) {
+ err_ = true;
+ MOZ_ASSERT(false, "Get bogus address for received UDP packet");
+ ABORT(r);
+ }
+
+ consumed_len = std::min(maxlen, msg->data->len());
+ if (consumed_len < msg->data->len()) {
+ r_log(LOG_GENERIC, LOG_DEBUG, "Partial received UDP packet will be discard");
+ }
+
+ memcpy(buf, msg->data->data(), consumed_len);
+ *len = consumed_len;
+ }
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int NrUdpSocketIpc::getaddr(nr_transport_addr *addrp) {
+ ASSERT_ON_THREAD(sts_thread_);
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ if (state_ != NR_CONNECTED) {
+ return R_INTERNAL;
+ }
+
+ return nr_transport_addr_copy(addrp, &my_addr_);
+}
+
+int NrUdpSocketIpc::connect(nr_transport_addr *addr) {
+ int r,_status;
+ int32_t port;
+ nsCString host;
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+ r_log(LOG_GENERIC, LOG_DEBUG, "NrUdpSocketIpc::connect(%s) this=%p", addr->as_string,
+ (void*) this);
+
+ if ((r=nr_transport_addr_get_addrstring_and_port(addr, &host, &port))) {
+ ABORT(r);
+ }
+
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnable(RefPtr<NrUdpSocketIpc>(this),
+ &NrUdpSocketIpc::connect_i,
+ host, static_cast<uint16_t>(port)),
+ NS_DISPATCH_NORMAL);
+
+ // Wait until connect() completes.
+ mon.Wait();
+
+ r_log(LOG_GENERIC, LOG_DEBUG, "NrUdpSocketIpc::connect this=%p completed err_ = %s",
+ (void*) this, err_ ? "true" : "false");
+
+ if (err_) {
+ ABORT(R_INTERNAL);
+ }
+
+ _status = 0;
+abort:
+ return _status;
+}
+
+int NrUdpSocketIpc::write(const void *msg, size_t len, size_t *written) {
+ MOZ_ASSERT(false);
+ return R_INTERNAL;
+}
+
+int NrUdpSocketIpc::read(void* buf, size_t maxlen, size_t *len) {
+ MOZ_ASSERT(false);
+ return R_INTERNAL;
+}
+
+int NrUdpSocketIpc::listen(int backlog) {
+ MOZ_ASSERT(false);
+ return R_INTERNAL;
+}
+
+int NrUdpSocketIpc::accept(nr_transport_addr *addrp, nr_socket **sockp) {
+ MOZ_ASSERT(false);
+ return R_INTERNAL;
+}
+
+// IO thread executors
+void NrUdpSocketIpc::create_i(const nsACString &host, const uint16_t port) {
+ ASSERT_ON_THREAD(io_thread_);
+
+ uint32_t minBuffSize = 0;
+ nsresult rv;
+ nsCOMPtr<nsIUDPSocketChild> socketChild = do_CreateInstance("@mozilla.org/udp-socket-child;1", &rv);
+ if (NS_FAILED(rv)) {
+ ReentrantMonitorAutoEnter mon(monitor_);
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to create UDPSocketChild");
+ return;
+ }
+
+ // This can spin the event loop; don't do that with the monitor held
+ socketChild->SetBackgroundSpinsEvents();
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+ if (!socket_child_) {
+ socket_child_ = socketChild;
+ socket_child_->SetFilterName(nsCString(NS_NETWORK_SOCKET_FILTER_HANDLER_STUN_SUFFIX));
+ } else {
+ socketChild = nullptr;
+ }
+
+ RefPtr<NrUdpSocketIpcProxy> proxy(new NrUdpSocketIpcProxy);
+ rv = proxy->Init(this);
+ if (NS_FAILED(rv)) {
+ err_ = true;
+ mon.NotifyAll();
+ return;
+ }
+
+#ifdef XP_WIN
+ if (!mozilla::IsWin8OrLater()) {
+ // Increase default receive and send buffer size on <= Win7 to be able to
+ // receive and send an unpaced HD (>= 720p = 1280x720 - I Frame ~ 21K size)
+ // stream without losing packets.
+ // Manual testing showed that 100K buffer size was not enough and the
+ // packet loss dis-appeared with 256K buffer size.
+ // See bug 1252769 for future improvements of this.
+ minBuffSize = 256 * 1024;
+ }
+#endif
+ // XXX bug 1126232 - don't use null Principal!
+ if (NS_FAILED(socket_child_->Bind(proxy, nullptr, host, port,
+ /* reuse = */ false,
+ /* loopback = */ false,
+ /* recv buffer size */ minBuffSize,
+ /* send buffer size */ minBuffSize))) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to create UDP socket");
+ mon.NotifyAll();
+ return;
+ }
+}
+
+void NrUdpSocketIpc::connect_i(const nsACString &host, const uint16_t port) {
+ ASSERT_ON_THREAD(io_thread_);
+ nsresult rv;
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ RefPtr<NrUdpSocketIpcProxy> proxy(new NrUdpSocketIpcProxy);
+ rv = proxy->Init(this);
+ if (NS_FAILED(rv)) {
+ err_ = true;
+ mon.NotifyAll();
+ return;
+ }
+
+ if (NS_FAILED(socket_child_->Connect(proxy, host, port))) {
+ err_ = true;
+ MOZ_ASSERT(false, "Failed to connect UDP socket");
+ mon.NotifyAll();
+ return;
+ }
+}
+
+
+void NrUdpSocketIpc::sendto_i(const net::NetAddr &addr, nsAutoPtr<DataBuffer> buf) {
+ ASSERT_ON_THREAD(io_thread_);
+
+ ReentrantMonitorAutoEnter mon(monitor_);
+
+ if (!socket_child_) {
+ MOZ_ASSERT(false);
+ err_ = true;
+ return;
+ }
+ if (NS_FAILED(socket_child_->SendWithAddress(&addr,
+ buf->data(),
+ buf->len()))) {
+ err_ = true;
+ }
+}
+
+void NrUdpSocketIpc::close_i() {
+ ASSERT_ON_THREAD(io_thread_);
+
+ if (socket_child_) {
+ socket_child_->Close();
+ socket_child_ = nullptr;
+ }
+}
+
+#if defined(MOZILLA_INTERNAL_API)
+// close(), but transfer the socket_child_ reference to die as well
+// static
+void NrUdpSocketIpc::release_child_i(nsIUDPSocketChild* aChild,
+ nsCOMPtr<nsIEventTarget> sts_thread) {
+ RefPtr<nsIUDPSocketChild> socket_child_ref =
+ already_AddRefed<nsIUDPSocketChild>(aChild);
+ if (socket_child_ref) {
+ socket_child_ref->Close();
+ }
+ // Tell SingletonThreadHolder we're done with it
+ RUN_ON_THREAD(sts_thread,
+ mozilla::WrapRunnableNM(&NrUdpSocketIpc::release_use_s),
+ NS_DISPATCH_NORMAL);
+}
+
+void NrUdpSocketIpc::release_use_s() {
+ sThread->ReleaseUse();
+}
+#endif
+
+void NrUdpSocketIpc::recv_callback_s(RefPtr<nr_udp_message> msg) {
+ ASSERT_ON_THREAD(sts_thread_);
+
+ {
+ ReentrantMonitorAutoEnter mon(monitor_);
+ if (state_ != NR_CONNECTED) {
+ return;
+ }
+ }
+
+ //enqueue received message
+ received_msgs_.push(msg);
+
+ if ((poll_flags() & PR_POLL_READ)) {
+ fire_callback(NR_ASYNC_WAIT_READ);
+ }
+}
+
+#if defined(MOZILLA_INTERNAL_API)
+// TCPSocket.
+class NrTcpSocketIpc::TcpSocketReadyRunner: public Runnable
+{
+public:
+ explicit TcpSocketReadyRunner(NrTcpSocketIpc *sck)
+ : socket_(sck) {}
+
+ NS_IMETHOD Run() override {
+ socket_->maybe_post_socket_ready();
+ return NS_OK;
+ }
+
+private:
+ RefPtr<NrTcpSocketIpc> socket_;
+};
+
+
+NS_IMPL_ISUPPORTS(NrTcpSocketIpc,
+ nsITCPSocketCallback)
+
+NrTcpSocketIpc::NrTcpSocketIpc(nsIThread* aThread)
+ : NrSocketIpc(static_cast<nsIEventTarget*>(aThread)),
+ mirror_state_(NR_INIT),
+ state_(NR_INIT),
+ buffered_bytes_(0),
+ tracking_number_(0) {
+}
+
+NrTcpSocketIpc::~NrTcpSocketIpc()
+{
+ // also guarantees socket_child_ is released from the io_thread
+
+ // close(), but transfer the socket_child_ reference to die as well
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnableNM(&NrTcpSocketIpc::release_child_i,
+ socket_child_.forget().take(),
+ sts_thread_),
+ NS_DISPATCH_NORMAL);
+}
+
+//
+// nsITCPSocketCallback methods
+//
+NS_IMETHODIMP NrTcpSocketIpc::UpdateReadyState(uint32_t aReadyState) {
+ NrSocketIpcState temp = NR_INIT;
+ switch (static_cast<dom::TCPReadyState>(aReadyState)) {
+ case dom::TCPReadyState::Connecting:
+ temp = NR_CONNECTING;
+ break;
+ case dom::TCPReadyState::Open:
+ temp = NR_CONNECTED;
+ break;
+ case dom::TCPReadyState::Closing:
+ temp = NR_CLOSING;
+ break;
+ case dom::TCPReadyState::Closed:
+ temp = NR_CLOSED;
+ break;
+ default:
+ MOZ_ASSERT(false, "Invalid ReadyState");
+ return NS_OK;
+ }
+ if (mirror_state_ != temp) {
+ mirror_state_ = temp;
+ RUN_ON_THREAD(sts_thread_,
+ mozilla::WrapRunnable(RefPtr<NrTcpSocketIpc>(this),
+ &NrTcpSocketIpc::update_state_s,
+ temp),
+ NS_DISPATCH_NORMAL);
+ }
+ return NS_OK;
+}
+
+NS_IMETHODIMP NrTcpSocketIpc::UpdateBufferedAmount(uint32_t buffered_amount,
+ uint32_t tracking_number) {
+ RUN_ON_THREAD(sts_thread_,
+ mozilla::WrapRunnable(RefPtr<NrTcpSocketIpc>(this),
+ &NrTcpSocketIpc::message_sent_s,
+ buffered_amount,
+ tracking_number),
+ NS_DISPATCH_NORMAL);
+
+ return NS_OK;
+}
+
+NS_IMETHODIMP NrTcpSocketIpc::FireDataArrayEvent(const nsAString& aType,
+ const InfallibleTArray<uint8_t>& buffer) {
+ // Called when we received data.
+ uint8_t *buf = const_cast<uint8_t*>(buffer.Elements());
+
+ nsAutoPtr<DataBuffer> data_buf(new DataBuffer(buf, buffer.Length()));
+ RefPtr<nr_tcp_message> msg = new nr_tcp_message(data_buf);
+
+ RUN_ON_THREAD(sts_thread_,
+ mozilla::WrapRunnable(RefPtr<NrTcpSocketIpc>(this),
+ &NrTcpSocketIpc::recv_message_s,
+ msg),
+ NS_DISPATCH_NORMAL);
+ return NS_OK;
+}
+
+NS_IMETHODIMP NrTcpSocketIpc::FireErrorEvent(const nsAString &type,
+ const nsAString &name) {
+ r_log(LOG_GENERIC, LOG_ERR,
+ "Error from TCPSocketChild: type: %s, name: %s",
+ NS_LossyConvertUTF16toASCII(type).get(), NS_LossyConvertUTF16toASCII(name).get());
+ socket_child_ = nullptr;
+
+ mirror_state_ = NR_CLOSED;
+ RUN_ON_THREAD(sts_thread_,
+ mozilla::WrapRunnable(RefPtr<NrTcpSocketIpc>(this),
+ &NrTcpSocketIpc::update_state_s,
+ NR_CLOSED),
+ NS_DISPATCH_NORMAL);
+
+ return NS_OK;
+}
+
+// methods of nsITCPSocketCallback that we are not going to implement.
+
+NS_IMETHODIMP NrTcpSocketIpc::FireDataStringEvent(const nsAString &type,
+ const nsACString &data) {
+ return NS_ERROR_NOT_IMPLEMENTED;
+}
+
+NS_IMETHODIMP NrTcpSocketIpc::FireEvent(const nsAString &type) {
+ // XXX support type.mData == 'close' at least
+ return NS_ERROR_NOT_IMPLEMENTED;
+}
+
+//
+// NrSocketBase methods.
+//
+int NrTcpSocketIpc::create(nr_transport_addr *addr) {
+ int r, _status;
+ nsresult rv;
+ int32_t port;
+ nsCString host;
+
+ if (state_ != NR_INIT) {
+ ABORT(R_INTERNAL);
+ }
+
+ sts_thread_ = do_GetService(NS_SOCKETTRANSPORTSERVICE_CONTRACTID, &rv);
+ if (NS_FAILED(rv)) {
+ MOZ_ASSERT(false, "Failed to get STS thread");
+ ABORT(R_INTERNAL);
+ }
+
+ // Sanity check
+ if ((r=nr_transport_addr_get_addrstring_and_port(addr, &host, &port))) {
+ ABORT(r);
+ }
+
+ if ((r=nr_transport_addr_copy(&my_addr_, addr))) {
+ ABORT(r);
+ }
+
+ _status = 0;
+abort:
+ return(_status);
+}
+
+int NrTcpSocketIpc::sendto(const void *msg, size_t len,
+ int flags, nr_transport_addr *to) {
+ MOZ_ASSERT(false);
+ return R_INTERNAL;
+}
+
+int NrTcpSocketIpc::recvfrom(void * buf, size_t maxlen,
+ size_t *len, int flags,
+ nr_transport_addr *from) {
+ MOZ_ASSERT(false);
+ return R_INTERNAL;
+}
+
+int NrTcpSocketIpc::getaddr(nr_transport_addr *addrp) {
+ ASSERT_ON_THREAD(sts_thread_);
+ return nr_transport_addr_copy(addrp, &my_addr_);
+}
+
+void NrTcpSocketIpc::close() {
+ ASSERT_ON_THREAD(sts_thread_);
+
+ if (state_ == NR_CLOSED || state_ == NR_CLOSING) {
+ return;
+ }
+
+ state_ = NR_CLOSING;
+
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnable(RefPtr<NrTcpSocketIpc>(this),
+ &NrTcpSocketIpc::close_i),
+ NS_DISPATCH_NORMAL);
+
+ //remove all enqueued messages
+ std::queue<RefPtr<nr_tcp_message>> empty;
+ std::swap(msg_queue_, empty);
+}
+
+int NrTcpSocketIpc::connect(nr_transport_addr *addr) {
+ nsCString remote_addr, local_addr;
+ int32_t remote_port, local_port;
+ int r, _status;
+ if ((r=nr_transport_addr_get_addrstring_and_port(addr,
+ &remote_addr,
+ &remote_port))) {
+ ABORT(r);
+ }
+
+ if ((r=nr_transport_addr_get_addrstring_and_port(&my_addr_,
+ &local_addr,
+ &local_port))) {
+ MOZ_ASSERT(false); // shouldn't fail as it was sanity-checked in ::create()
+ ABORT(r);
+ }
+
+ state_ = mirror_state_ = NR_CONNECTING;
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnable(RefPtr<NrTcpSocketIpc>(this),
+ &NrTcpSocketIpc::connect_i,
+ remote_addr,
+ static_cast<uint16_t>(remote_port),
+ local_addr,
+ static_cast<uint16_t>(local_port)),
+ NS_DISPATCH_NORMAL);
+
+ // Make caller wait for ready to write.
+ _status = R_WOULDBLOCK;
+ abort:
+ return _status;
+}
+
+int NrTcpSocketIpc::write(const void *msg, size_t len, size_t *written) {
+ ASSERT_ON_THREAD(sts_thread_);
+ int _status = 0;
+ if (state_ != NR_CONNECTED) {
+ ABORT(R_FAILED);
+ }
+
+ if (buffered_bytes_ + len >= nsITCPSocketCallback::BUFFER_SIZE) {
+ ABORT(R_WOULDBLOCK);
+ }
+
+ buffered_bytes_ += len;
+ {
+ InfallibleTArray<uint8_t>* arr = new InfallibleTArray<uint8_t>();
+ arr->AppendElements(static_cast<const uint8_t*>(msg), len);
+ // keep track of un-acknowleged writes by tracking number.
+ writes_in_flight_.push_back(len);
+ RUN_ON_THREAD(io_thread_,
+ mozilla::WrapRunnable(RefPtr<NrTcpSocketIpc>(this),
+ &NrTcpSocketIpc::write_i,
+ nsAutoPtr<InfallibleTArray<uint8_t>>(arr),
+ ++tracking_number_),
+ NS_DISPATCH_NORMAL);
+ }
+ *written = len;
+ abort:
+ return _status;
+}
+
+int NrTcpSocketIpc::read(void* buf, size_t maxlen, size_t *len) {
+ int _status = 0;
+ if (state_ != NR_CONNECTED) {
+ ABORT(R_FAILED);
+ }
+
+ if (msg_queue_.size() == 0) {
+ ABORT(R_WOULDBLOCK);
+ }
+
+ {
+ RefPtr<nr_tcp_message> msg(msg_queue_.front());
+ size_t consumed_len = std::min(maxlen, msg->unread_bytes());
+ memcpy(buf, msg->reading_pointer(), consumed_len);
+ if (consumed_len < msg->unread_bytes()) {
+ // There is still something left in buffer.
+ msg->read_bytes += consumed_len;
+ } else {
+ msg_queue_.pop();
+ }
+ *len = consumed_len;
+ }
+
+ abort:
+ return _status;
+}
+
+int NrTcpSocketIpc::listen(int backlog) {
+ return R_INTERNAL;
+}
+
+int NrTcpSocketIpc::accept(nr_transport_addr *addrp, nr_socket **sockp) {
+ return R_INTERNAL;
+}
+
+void NrTcpSocketIpc::connect_i(const nsACString &remote_addr,
+ uint16_t remote_port,
+ const nsACString &local_addr,
+ uint16_t local_port) {
+ ASSERT_ON_THREAD(io_thread_);
+ mirror_state_ = NR_CONNECTING;
+
+ dom::TCPSocketChild* child = new dom::TCPSocketChild(NS_ConvertUTF8toUTF16(remote_addr), remote_port);
+ socket_child_ = child;
+
+ // Bug 1285330: put filtering back in here
+
+ // XXX remove remote!
+ socket_child_->SendWindowlessOpenBind(this,
+ remote_addr, remote_port,
+ local_addr, local_port,
+ /* use ssl */ false);
+}
+
+void NrTcpSocketIpc::write_i(nsAutoPtr<InfallibleTArray<uint8_t>> arr,
+ uint32_t tracking_number) {
+ ASSERT_ON_THREAD(io_thread_);
+ if (!socket_child_) {
+ return;
+ }
+ socket_child_->SendSendArray(*arr, tracking_number);
+}
+
+void NrTcpSocketIpc::close_i() {
+ ASSERT_ON_THREAD(io_thread_);
+ mirror_state_ = NR_CLOSING;
+ if (!socket_child_) {
+ return;
+ }
+ socket_child_->SendClose();
+}
+
+// close(), but transfer the socket_child_ reference to die as well
+// static
+void NrTcpSocketIpc::release_child_i(dom::TCPSocketChild* aChild,
+ nsCOMPtr<nsIEventTarget> sts_thread) {
+ RefPtr<dom::TCPSocketChild> socket_child_ref =
+ already_AddRefed<dom::TCPSocketChild>(aChild);
+ if (socket_child_ref) {
+ socket_child_ref->SendClose();
+ }
+ // io_thread_ is MainThread, so no use to release
+}
+
+void NrTcpSocketIpc::message_sent_s(uint32_t buffered_amount,
+ uint32_t tracking_number) {
+ ASSERT_ON_THREAD(sts_thread_);
+
+ size_t num_unacked_writes = tracking_number_ - tracking_number;
+ while (writes_in_flight_.size() > num_unacked_writes) {
+ writes_in_flight_.pop_front();
+ }
+
+ for (size_t unacked_write_len : writes_in_flight_) {
+ buffered_amount += unacked_write_len;
+ }
+
+ r_log(LOG_GENERIC, LOG_ERR,
+ "UpdateBufferedAmount: (tracking %u): %u, waiting: %s",
+ tracking_number, buffered_amount,
+ (poll_flags() & PR_POLL_WRITE) ? "yes" : "no");
+
+ buffered_bytes_ = buffered_amount;
+ maybe_post_socket_ready();
+}
+
+void NrTcpSocketIpc::recv_message_s(nr_tcp_message *msg) {
+ ASSERT_ON_THREAD(sts_thread_);
+ msg_queue_.push(msg);
+ maybe_post_socket_ready();
+}
+
+void NrTcpSocketIpc::update_state_s(NrSocketIpcState next_state) {
+ ASSERT_ON_THREAD(sts_thread_);
+ // only allow valid transitions
+ switch (state_) {
+ case NR_CONNECTING:
+ if (next_state == NR_CONNECTED) {
+ state_ = NR_CONNECTED;
+ maybe_post_socket_ready();
+ } else {
+ state_ = next_state; // all states are valid from CONNECTING
+ }
+ break;
+ case NR_CONNECTED:
+ if (next_state != NR_CONNECTING) {
+ state_ = next_state;
+ }
+ break;
+ case NR_CLOSING:
+ if (next_state == NR_CLOSED) {
+ state_ = next_state;
+ }
+ break;
+ case NR_CLOSED:
+ break;
+ default:
+ MOZ_CRASH("update_state_s while in illegal state");
+ }
+}
+
+void NrTcpSocketIpc::maybe_post_socket_ready() {
+ bool has_event = false;
+ if (state_ == NR_CONNECTED) {
+ if (poll_flags() & PR_POLL_WRITE) {
+ // This effectively polls via the event loop until the
+ // NR_ASYNC_WAIT_WRITE is no longer armed.
+ if (buffered_bytes_ < nsITCPSocketCallback::BUFFER_SIZE) {
+ r_log(LOG_GENERIC, LOG_INFO, "Firing write callback (%u)",
+ (uint32_t)buffered_bytes_);
+ fire_callback(NR_ASYNC_WAIT_WRITE);
+ has_event = true;
+ }
+ }
+ if (poll_flags() & PR_POLL_READ) {
+ if (msg_queue_.size()) {
+ r_log(LOG_GENERIC, LOG_INFO, "Firing read callback (%u)",
+ (uint32_t)msg_queue_.size());
+ fire_callback(NR_ASYNC_WAIT_READ);
+ has_event = true;
+ }
+ }
+ }
+
+ // If any event has been posted, we post a runnable to see
+ // if the events have to be posted again.
+ if (has_event) {
+ RefPtr<TcpSocketReadyRunner> runnable = new TcpSocketReadyRunner(this);
+ NS_DispatchToCurrentThread(runnable);
+ }
+}
+#endif
+
+} // close namespace
+
+
+using namespace mozilla;
+
+// Bridge to the nr_socket interface
+static int nr_socket_local_destroy(void **objp);
+static int nr_socket_local_sendto(void *obj,const void *msg, size_t len,
+ int flags, nr_transport_addr *to);
+static int nr_socket_local_recvfrom(void *obj,void * restrict buf,
+ size_t maxlen, size_t *len, int flags, nr_transport_addr *from);
+static int nr_socket_local_getfd(void *obj, NR_SOCKET *fd);
+static int nr_socket_local_getaddr(void *obj, nr_transport_addr *addrp);
+static int nr_socket_local_close(void *obj);
+static int nr_socket_local_connect(void *sock, nr_transport_addr *addr);
+static int nr_socket_local_write(void *obj,const void *msg, size_t len,
+ size_t *written);
+static int nr_socket_local_read(void *obj,void * restrict buf, size_t maxlen,
+ size_t *len);
+static int nr_socket_local_listen(void *obj, int backlog);
+static int nr_socket_local_accept(void *obj, nr_transport_addr *addrp,
+ nr_socket **sockp);
+
+static nr_socket_vtbl nr_socket_local_vtbl={
+ 2,
+ nr_socket_local_destroy,
+ nr_socket_local_sendto,
+ nr_socket_local_recvfrom,
+ nr_socket_local_getfd,
+ nr_socket_local_getaddr,
+ nr_socket_local_connect,
+ nr_socket_local_write,
+ nr_socket_local_read,
+ nr_socket_local_close,
+ nr_socket_local_listen,
+ nr_socket_local_accept
+};
+
+/* static */
+int
+NrSocketBase::CreateSocket(nr_transport_addr *addr, RefPtr<NrSocketBase> *sock)
+{
+ int r, _status;
+
+ // create IPC bridge for content process
+ if (XRE_IsParentProcess()) {
+ *sock = new NrSocket();
+ } else {
+ switch (addr->protocol) {
+ case IPPROTO_UDP:
+ *sock = new NrUdpSocketIpc();
+ break;
+ case IPPROTO_TCP:
+#if defined(MOZILLA_INTERNAL_API)
+ {
+ nsCOMPtr<nsIThread> main_thread;
+ NS_GetMainThread(getter_AddRefs(main_thread));
+ *sock = new NrTcpSocketIpc(main_thread.get());
+ }
+#else
+ ABORT(R_REJECTED);
+#endif
+ break;
+ }
+ }
+
+ r = (*sock)->create(addr);
+ if (r)
+ ABORT(r);
+
+ _status = 0;
+abort:
+ if (_status) {
+ *sock = nullptr;
+ }
+ return _status;
+}
+
+int nr_socket_local_create(void *obj, nr_transport_addr *addr, nr_socket **sockp) {
+ RefPtr<NrSocketBase> sock;
+ int r, _status;
+
+ r = NrSocketBase::CreateSocket(addr, &sock);
+ if (r) {
+ ABORT(r);
+ }
+
+ r = nr_socket_create_int(static_cast<void *>(sock),
+ sock->vtbl(), sockp);
+ if (r)
+ ABORT(r);
+
+ _status = 0;
+
+ {
+ // We will release this reference in destroy(), not exactly the normal
+ // ownership model, but it is what it is.
+ NrSocketBase* dummy = sock.forget().take();
+ (void)dummy;
+ }
+
+abort:
+ return _status;
+}
+
+
+static int nr_socket_local_destroy(void **objp) {
+ if(!objp || !*objp)
+ return 0;
+
+ NrSocketBase *sock = static_cast<NrSocketBase *>(*objp);
+ *objp = 0;
+
+ sock->close(); // Signal STS that we want not to listen
+ sock->Release(); // Decrement the ref count
+
+ return 0;
+}
+
+static int nr_socket_local_sendto(void *obj,const void *msg, size_t len,
+ int flags, nr_transport_addr *addr) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->sendto(msg, len, flags, addr);
+}
+
+static int nr_socket_local_recvfrom(void *obj,void * restrict buf,
+ size_t maxlen, size_t *len, int flags,
+ nr_transport_addr *addr) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->recvfrom(buf, maxlen, len, flags, addr);
+}
+
+static int nr_socket_local_getfd(void *obj, NR_SOCKET *fd) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ *fd = sock;
+
+ return 0;
+}
+
+static int nr_socket_local_getaddr(void *obj, nr_transport_addr *addrp) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->getaddr(addrp);
+}
+
+
+static int nr_socket_local_close(void *obj) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ sock->close();
+
+ return 0;
+}
+
+static int nr_socket_local_write(void *obj, const void *msg, size_t len,
+ size_t *written) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->write(msg, len, written);
+}
+
+static int nr_socket_local_read(void *obj, void * restrict buf, size_t maxlen,
+ size_t *len) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->read(buf, maxlen, len);
+}
+
+static int nr_socket_local_connect(void *obj, nr_transport_addr *addr) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->connect(addr);
+}
+
+static int nr_socket_local_listen(void *obj, int backlog) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->listen(backlog);
+}
+
+static int nr_socket_local_accept(void *obj, nr_transport_addr *addrp,
+ nr_socket **sockp) {
+ NrSocketBase *sock = static_cast<NrSocketBase *>(obj);
+
+ return sock->accept(addrp, sockp);
+}
+
+// Implement async api
+int NR_async_wait(NR_SOCKET sock, int how, NR_async_cb cb,void *cb_arg,
+ char *function,int line) {
+ NrSocketBase *s = static_cast<NrSocketBase *>(sock);
+
+ return s->async_wait(how, cb, cb_arg, function, line);
+}
+
+int NR_async_cancel(NR_SOCKET sock,int how) {
+ NrSocketBase *s = static_cast<NrSocketBase *>(sock);
+
+ return s->cancel(how);
+}
+
+nr_socket_vtbl* NrSocketBase::vtbl() {
+ return &nr_socket_local_vtbl;
+}