Backed out 3 changesets (bug 1497249) for causing topcrash bug 1663718.

Backed out changeset 1cd6d2f9a603 (bug 1497249)
Backed out changeset 2502449c2e24 (bug 1497249)
Backed out changeset 432c7e9dc19f (bug 1497249)

media/mtransport/ipc/WebrtcTCPSocket.cpp

@@ -555,12 +555,6 @@ WebrtcTCPSocket::OnTransportAvailable(nsISocketTransport* aTransport,
   return NS_OK;
 }
 
-NS_IMETHODIMP
-WebrtcTCPSocket::OnWebSocketConnectionAvailable(
-    nsIWebSocketConnection* aConnection) {
-  return NS_OK;
-}
-
 void WebrtcTCPSocket::FinishOpen() {
   MOZ_ASSERT(OnSocketThread());
   // mTransport, mSocketIn, and mSocketOut are all set. We may have set them in

netwerk/protocol/http/HttpConnectionMgrChild.cpp

@@ -11,7 +11,6 @@
 #include "HttpTransactionChild.h"
 #include "AltSvcTransactionChild.h"
 #include "EventTokenBucket.h"
-#include "mozilla/net/WebSocketConnectionChild.h"
 #include "nsHttpConnectionInfo.h"
 #include "nsHttpConnectionMgr.h"
 #include "nsHttpHandler.h"
@@ -178,25 +177,5 @@ mozilla::ipc::IPCResult HttpConnectionMgrChild::RecvSpeculativeConnect(
   return IPC_OK();
 }
 
-already_AddRefed<PWebSocketConnectionChild>
-HttpConnectionMgrChild::AllocPWebSocketConnectionChild(
-    PHttpTransactionChild* aTransWithStickyConn) {
-  RefPtr<WebSocketConnectionChild> actor = new WebSocketConnectionChild();
-  return actor.forget();
-}
-
-mozilla::ipc::IPCResult
-HttpConnectionMgrChild::RecvPWebSocketConnectionConstructor(
-    PWebSocketConnectionChild* aActor,
-    PHttpTransactionChild* aTransWithStickyConn) {
-  RefPtr<WebSocketConnectionChild> child =
-      static_cast<WebSocketConnectionChild*>(aActor);
-  nsCOMPtr<nsIHttpUpgradeListener> listener =
-      static_cast<nsIHttpUpgradeListener*>(child.get());
-  Unused << mConnMgr->CompleteUpgrade(
-      ToRealHttpTransaction(aTransWithStickyConn), listener);
-  return IPC_OK();
-}
-
 }  // namespace net
 }  // namespace mozilla

netwerk/protocol/http/HttpConnectionMgrChild.h

@@ -41,12 +41,6 @@ class HttpConnectionMgrChild final : public PHttpConnectionMgrChild {
       Maybe<SpeculativeConnectionOverriderArgs> aOverriderArgs, uint32_t aCaps,
       Maybe<PAltSvcTransactionChild*> aTrans);
 
-  already_AddRefed<PWebSocketConnectionChild> AllocPWebSocketConnectionChild(
-      PHttpTransactionChild* aTransWithStickyConn);
-  mozilla::ipc::IPCResult RecvPWebSocketConnectionConstructor(
-      PWebSocketConnectionChild* aActor,
-      PHttpTransactionChild* aTransWithStickyConn) override;
-
  private:
   virtual ~HttpConnectionMgrChild();
 

netwerk/protocol/http/HttpConnectionMgrParent.cpp

@@ -10,7 +10,6 @@
 #include "HttpConnectionMgrParent.h"
 #include "AltSvcTransactionParent.h"
 #include "mozilla/net/HttpTransactionParent.h"
-#include "mozilla/net/WebSocketConnectionParent.h"
 #include "nsHttpConnectionInfo.h"
 #include "nsIInterfaceRequestorUtils.h"
 #include "nsISpeculativeConnect.h"
@@ -253,13 +252,8 @@ nsresult HttpConnectionMgrParent::ClearConnectionHistory() {
 
 nsresult HttpConnectionMgrParent::CompleteUpgrade(
     HttpTransactionShell* aTrans, nsIHttpUpgradeListener* aUpgradeListener) {
-  MOZ_ASSERT(aTrans->AsHttpTransactionParent());
-
-  RefPtr<WebSocketConnectionParent> wsConnParent =
-      new WebSocketConnectionParent(aUpgradeListener);
-  Unused << SendPWebSocketConnectionConstructor(
-      wsConnParent, aTrans->AsHttpTransactionParent());
-  return NS_OK;
+  // TODO: fix this in bug 1497249
+  return NS_ERROR_NOT_IMPLEMENTED;
 }
 
 nsHttpConnectionMgr* HttpConnectionMgrParent::AsHttpConnectionMgr() {

netwerk/protocol/http/PHttpConnectionMgr.ipdl

@@ -8,7 +8,6 @@
 include protocol PAltSvcTransaction;
 include protocol PSocketProcess;
 include protocol PHttpTransaction;
-include protocol PWebSocketConnection;
 
 include NeckoChannelParams;
 
@@ -18,7 +17,6 @@ namespace net {
 async refcounted protocol PHttpConnectionMgr
 {
   manager PSocketProcess;
-  manages PWebSocketConnection;
 
 child:
   async __delete__();
@@ -35,7 +33,6 @@ child:
   async SpeculativeConnect(HttpConnectionInfoCloneArgs aConnInfo,
                            SpeculativeConnectionOverriderArgs? aOverriderArgs,
                            uint32_t aCaps, PAltSvcTransaction? aTrans);
-  async PWebSocketConnection(PHttpTransaction aTransWithStickyConn);
 };
 
 } // namespace net

netwerk/protocol/http/nsHttpChannel.cpp

@@ -8037,13 +8037,6 @@ nsresult nsHttpChannel::ContinueOnStopRequestAfterAuthRetry(
                         mResponseHead->Status() == 200;
 
   if (upgradeWebsocket || upgradeConnect) {
-    // TODO: Support connection upgrade for socket process in bug 1632809.
-    if (nsIOService::UseSocketProcess() && upgradeConnect) {
-      Unused << mUpgradeProtocolCallback->OnUpgradeFailed(
-          NS_ERROR_NOT_IMPLEMENTED);
-      return ContinueOnStopRequest(aStatus, aIsFromNet, aContentComplete);
-    }
-
     nsresult rv = gHttpHandler->CompleteUpgrade(aTransWithStickyConn,
                                                 mUpgradeProtocolCallback);
     if (NS_FAILED(rv)) {

netwerk/protocol/http/nsIHttpChannelInternal.idl

@@ -27,7 +27,6 @@ interface nsIProxyInfo;
 interface nsISecurityConsoleMessage;
 interface nsISocketTransport;
 interface nsIURI;
-interface nsIWebSocketConnection;
 
 /**
  * The callback interface for nsIHttpChannelInternal::HTTPUpgrade()
@@ -41,8 +40,6 @@ interface nsIHttpUpgradeListener : nsISupports
                                          in nsIAsyncOutputStream aSocketOut);
 
     [must_use] void onUpgradeFailed(in nsresult aErrorCode);
-
-    [must_use] void onWebSocketConnectionAvailable(in nsIWebSocketConnection aConnection);
 };
 
 /**

netwerk/protocol/websocket/IPCTransportProvider.cpp

@@ -48,12 +48,6 @@ TransportProviderParent::OnTransportAvailable(
   return NS_OK;
 }
 
-NS_IMETHODIMP
-TransportProviderParent::OnWebSocketConnectionAvailable(
-    nsIWebSocketConnection* aConnection) {
-  return NS_OK;
-}
-
 NS_IMETHODIMP
 TransportProviderParent::OnUpgradeFailed(nsresult aErrorCode) { return NS_OK; }
 

netwerk/protocol/websocket/PWebSocketConnection.ipdl

-/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
-/* vim: set sw=2 ts=8 et tw=80 ft=cpp : */
-/* 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 protocol PHttpConnectionMgr;
-
-namespace mozilla {
-namespace net {
-
-async refcounted protocol PWebSocketConnection
-{
-  manager PHttpConnectionMgr;
-
-parent:
-  async OnTransportAvailable(nsCString securityInfoSerialization);
-  async OnError(nsresult aStatus);
-  async OnTCPClosed();
-  async OnDataReceived(uint8_t[] aData);
-  async OnUpgradeFailed(nsresult aReason);
-
-child:
-  async EnqueueOutgoingData(uint8_t[] aData);
-  async StartReading();
-  async DrainSocketData();
-
-  async __delete__();
-
-};
-
-} //namespace net
-} //namespace mozilla

netwerk/protocol/websocket/WebSocketChannel.cpp

@@ -56,7 +56,6 @@
 #include "mozilla/TimeStamp.h"
 #include "nsSocketTransportService2.h"
 #include "nsINSSErrorsService.h"
-#include "nsWebSocketConnection.h"
 
 #include "plbase64.h"
 #include "prmem.h"
@@ -76,10 +75,10 @@ namespace net {
 
 NS_IMPL_ISUPPORTS(WebSocketChannel, nsIWebSocketChannel, nsIHttpUpgradeListener,
                   nsIRequestObserver, nsIStreamListener, nsIProtocolHandler,
+                  nsIInputStreamCallback, nsIOutputStreamCallback,
                   nsITimerCallback, nsIDNSListener, nsIProtocolProxyCallback,
                   nsIInterfaceRequestor, nsIChannelEventSink,
-                  nsIThreadRetargetableRequest, nsIObserver, nsINamed,
-                  nsIWebSocketConnectionListener)
+                  nsIThreadRetargetableRequest, nsIObserver, nsINamed)
 
 // We implement RFC 6455, which uses Sec-WebSocket-Version: 13 on the wire.
 #define SEC_WEBSOCKET_VERSION "13"
@@ -1125,7 +1124,8 @@ WebSocketChannel::WebSocketChannel()
       mBuffered(0),
       mBufferSize(kIncomingBufferInitialSize),
       mCurrentOut(nullptr),
-      mHdrOutSize(0),
+      mCurrentOutSent(0),
+      mHdrOutToSend(0),
       mHdrOut(nullptr),
       mDynamicOutputSize(0),
       mDynamicOutput(nullptr),
@@ -1898,7 +1898,7 @@ void WebSocketChannel::EnqueueOutgoingMessage(nsDeque<OutboundMessage>& aQueue,
        this, aMsg, msgNames[aMsg->GetMsgType()], aMsg->Length()));
 
   aQueue.Push(aMsg);
-  DoEnqueueOutgoingMessage();
+  OnOutputStreamReady(mSocketOut);
 }
 
 uint16_t WebSocketChannel::ResultToCloseCode(nsresult resultCode) {
@@ -1948,6 +1948,7 @@ void WebSocketChannel::PrimeNewOutgoingMessage() {
        "%p found queued msg %p [type=%s len=%d]\n",
        this, mCurrentOut, msgNames[msgType], mCurrentOut->Length()));
 
+  mCurrentOutSent = 0;
   mHdrOut = mOutHeader;
 
   uint8_t maskBit = mIsServerSide ? 0 : kMaskBit;
@@ -1978,25 +1979,25 @@ void WebSocketChannel::PrimeNewOutgoingMessage() {
       if (mScriptCloseCode) {
         NetworkEndian::writeUint16(payload, mScriptCloseCode);
         mOutHeader[1] += 2;
-        mHdrOutSize = 4 + maskSize;
+        mHdrOutToSend = 4 + maskSize;
         if (!mScriptCloseReason.IsEmpty()) {
           MOZ_ASSERT(mScriptCloseReason.Length() <= 123,
                      "Close Reason Too Long");
           mOutHeader[1] += mScriptCloseReason.Length();
-          mHdrOutSize += mScriptCloseReason.Length();
+          mHdrOutToSend += mScriptCloseReason.Length();
           memcpy(payload + 2, mScriptCloseReason.BeginReading(),
-                 std::min<uint32_t>(mScriptCloseReason.Length(), 123));
+                 mScriptCloseReason.Length());
         }
       } else {
         // No close code/reason, so payload length = 0.  We must still send mask
         // even though it's not used.  Keep payload offset so we write mask
         // below.
-        mHdrOutSize = 2 + maskSize;
+        mHdrOutToSend = 2 + maskSize;
       }
     } else {
       NetworkEndian::writeUint16(payload, ResultToCloseCode(mStopOnClose));
       mOutHeader[1] += 2;
-      mHdrOutSize = 4 + maskSize;
+      mHdrOutToSend = 4 + maskSize;
     }
 
     if (mServerClosed) {
@@ -2064,18 +2065,18 @@ void WebSocketChannel::PrimeNewOutgoingMessage() {
 
     if (mCurrentOut->Length() < 126) {
       mOutHeader[1] = mCurrentOut->Length() | maskBit;
-      mHdrOutSize = 2 + maskSize;
+      mHdrOutToSend = 2 + maskSize;
     } else if (mCurrentOut->Length() <= 0xffff) {
       mOutHeader[1] = 126 | maskBit;
       NetworkEndian::writeUint16(mOutHeader + sizeof(uint16_t),
                                  mCurrentOut->Length());
-      mHdrOutSize = 4 + maskSize;
+      mHdrOutToSend = 4 + maskSize;
     } else {
       mOutHeader[1] = 127 | maskBit;
       NetworkEndian::writeUint64(mOutHeader + 2, mCurrentOut->Length());
-      mHdrOutSize = 10 + maskSize;
+      mHdrOutToSend = 10 + maskSize;
     }
-    payload = mOutHeader + mHdrOutSize;
+    payload = mOutHeader + mHdrOutToSend;
   }
 
   MOZ_ASSERT(payload, "payload offset not found");
@@ -2117,7 +2118,7 @@ void WebSocketChannel::PrimeNewOutgoingMessage() {
       mOutHeader[0] & WebSocketChannel::kRsv3Bit,
       mOutHeader[0] & WebSocketChannel::kOpcodeBitsMask,
       mOutHeader[1] & WebSocketChannel::kMaskBit, mask, payload,
-      mHdrOutSize - (payload - mOutHeader), mCurrentOut->BeginOrigReading(),
+      mHdrOutToSend - (payload - mOutHeader), mCurrentOut->BeginOrigReading(),
       mCurrentOut->OrigLength());
 
   if (frame) {
@@ -2125,7 +2126,7 @@ void WebSocketChannel::PrimeNewOutgoingMessage() {
   }
 
   if (mask) {
-    while (payload < (mOutHeader + mHdrOutSize)) {
+    while (payload < (mOutHeader + mHdrOutToSend)) {
       *payload ^= mask >> 24;
       mask = RotateLeft(mask, 8);
       payload++;
@@ -2135,8 +2136,19 @@ void WebSocketChannel::PrimeNewOutgoingMessage() {
     ApplyMask(mask, mCurrentOut->BeginWriting(), mCurrentOut->Length());
   }
 
-  // Transmitting begins - mHdrOutSize bytes from mOutHeader and
-  // mCurrentOut->Length() bytes from mCurrentOut.
+  int32_t len = mCurrentOut->Length();
+
+  // for small frames, copy it all together for a contiguous write
+  if (len && len <= kCopyBreak) {
+    memcpy(mOutHeader + mHdrOutToSend, mCurrentOut->BeginWriting(), len);
+    mHdrOutToSend += len;
+    mCurrentOutSent = len;
+  }
+
+  // Transmitting begins - mHdrOutToSend bytes from mOutHeader and
+  // mCurrentOut->Length() bytes from mCurrentOut. The latter may be
+  // coaleseced into the former for small messages or as the result of the
+  // compression process.
 
   cleanupAfterFailure.release();
 }
@@ -2144,6 +2156,7 @@ void WebSocketChannel::PrimeNewOutgoingMessage() {
 void WebSocketChannel::DeleteCurrentOutGoingMessage() {
   delete mCurrentOut;
   mCurrentOut = nullptr;
+  mCurrentOutSent = 0;
 }
 
 void WebSocketChannel::EnsureHdrOut(uint32_t size) {
@@ -2189,9 +2202,23 @@ void WebSocketChannel::CleanupConnection() {
     mLingeringCloseTimer = nullptr;
   }
 
-  if (mConnection) {
-    mConnection->Close();
-    mConnection = nullptr;
+  if (mSocketIn) {
+    if (mDataStarted) {
+      mSocketIn->AsyncWait(nullptr, 0, 0, nullptr);
+    }
+    mSocketIn = nullptr;
+  }
+
+  if (mSocketOut) {
+    mSocketOut->AsyncWait(nullptr, 0, 0, nullptr);
+    mSocketOut = nullptr;
+  }
+
+  if (mTransport) {
+    mTransport->SetSecurityCallbacks(nullptr);
+    mTransport->SetEventSink(nullptr, nullptr);
+    mTransport->Close(NS_BASE_STREAM_CLOSED);
+    mTransport = nullptr;
   }
 
   if (mConnectionLogService && !mPrivateBrowsing) {
@@ -2259,15 +2286,30 @@ void WebSocketChannel::DoStopSession(nsresult reason) {
     mPingTimer = nullptr;
   }
 
-  if (mConnection && !mTCPClosed && mDataStarted) {
-    // Drain, within reason, this socket.
-    mConnection->DrainSocketData();
+  if (mSocketIn && !mTCPClosed && mDataStarted) {
+    // Drain, within reason, this socket. if we leave any data
+    // unconsumed (including the tcp fin) a RST will be generated
+    // The right thing to do here is shutdown(SHUT_WR) and then wait
+    // a little while to see if any data comes in.. but there is no
+    // reason to delay things for that when the websocket handshake
+    // is supposed to guarantee a quiet connection except for that fin.
+
+    char buffer[512];
+    uint32_t count = 0;
+    uint32_t total = 0;
+    nsresult rv;
+    do {
+      total += count;
+      rv = mSocketIn->Read(buffer, 512, &count);
+      if (rv != NS_BASE_STREAM_WOULD_BLOCK && (NS_FAILED(rv) || count == 0))
+        mTCPClosed = true;
+    } while (NS_SUCCEEDED(rv) && count > 0 && total < 32000);
   }
 
   int32_t sessionCount = kLingeringCloseThreshold;
   nsWSAdmissionManager::GetSessionCount(sessionCount);
 
-  if (!mTCPClosed && mConnection && sessionCount < kLingeringCloseThreshold) {
+  if (!mTCPClosed && mTransport && sessionCount < kLingeringCloseThreshold) {
     // 7.1.1 says that the client SHOULD wait for the server to close the TCP
     // connection. This is so we can reuse port numbers before 2 MSL expires,
     // which is not really as much of a concern for us as the amount of state
@@ -2334,7 +2376,7 @@ void WebSocketChannel::AbortSession(nsresult reason) {
       return;
     }
 
-    if (mConnection && reason != NS_BASE_STREAM_CLOSED && !mRequestedClose &&
+    if (mTransport && reason != NS_BASE_STREAM_CLOSED && !mRequestedClose &&
         !mClientClosed && !mServerClosed && mDataStarted) {
       mRequestedClose = true;
       mStopOnClose = reason;
@@ -2812,11 +2854,10 @@ nsresult WebSocketChannel::StartWebsocketData() {
     mDataStarted = true;
   }
 
-  rv = mConnection->StartReading();
+  rv = mSocketIn->AsyncWait(this, 0, 0, mSocketThread);
   if (NS_FAILED(rv)) {
     LOG(
-        ("WebSocketChannel::StartWebsocketData mConnection->StartReading() "
-         "failed "
+        ("WebSocketChannel::StartWebsocketData mSocketIn->AsyncWait() failed "
          "with error 0x%08" PRIx32,
          static_cast<uint32_t>(rv)));
     return mSocketThread->Dispatch(
@@ -2825,24 +2866,18 @@ nsresult WebSocketChannel::StartWebsocketData() {
         NS_DISPATCH_NORMAL);
   }
 
-  RefPtr<WebSocketChannel> self = this;
-  rv = mSocketThread->Dispatch(
-      NS_NewRunnableFunction("net::WebSocketChannel::StartPinging",
-                             [self]() {
-                               if (self->mPingInterval) {
-                                 Unused << self->StartPinging();
-                               }
-                               // Try to send the messages in the queue out
-                               // immediately.
-                               self->DoEnqueueOutgoingMessage();
-                             }),
-      NS_DISPATCH_NORMAL);
-  if (NS_FAILED(rv)) {
-    LOG(
-        ("WebSocketChannel::StartWebsocketData Could not start pinging, "
-         "rv=0x%08" PRIx32,
-         static_cast<uint32_t>(rv)));
-    return rv;
+  if (mPingInterval) {
+    rv = mSocketThread->Dispatch(
+        NewRunnableMethod("net::WebSocketChannel::StartPinging", this,
+                          &WebSocketChannel::StartPinging),
+        NS_DISPATCH_NORMAL);
+    if (NS_FAILED(rv)) {
+      LOG(
+          ("WebSocketChannel::StartWebsocketData Could not start pinging, "
+           "rv=0x%08" PRIx32,
+           static_cast<uint32_t>(rv)));
+      return rv;
+    }
   }
 
   LOG(("WebSocketChannel::StartWebsocketData Notifying Listener %p",
@@ -3178,8 +3213,8 @@ WebSocketChannel::GetSecurityInfo(nsISupports** aSecurityInfo) {
   LOG(("WebSocketChannel::GetSecurityInfo() %p\n", this));
   MOZ_ASSERT(NS_IsMainThread(), "not main thread");
 
-  if (mConnection) {
-    if (NS_FAILED(mConnection->GetSecurityInfo(aSecurityInfo)))
+  if (mTransport) {
+    if (NS_FAILED(mTransport->GetSecurityInfo(aSecurityInfo)))
       *aSecurityInfo = nullptr;
   }
   return NS_OK;
@@ -3529,32 +3564,14 @@ WebSocketChannel::OnTransportAvailable(nsISocketTransport* aTransport,
   MOZ_ASSERT(!mRecvdHttpUpgradeTransport, "OTA duplicated");
   MOZ_ASSERT(aSocketIn, "OTA with invalid socketIn");
 
-  return OnWebSocketConnectionAvailable(
-      new nsWebSocketConnection(aTransport, aSocketIn, aSocketOut));
-}
-
-NS_IMETHODIMP
-WebSocketChannel::OnWebSocketConnectionAvailable(
-    nsIWebSocketConnection* aConnection) {
-  LOG(
-      ("WebSocketChannel::OnWebSocketConnectionAvailable %p [%p] "
-       "rcvdonstart=%d\n",
-       this, aConnection, mGotUpgradeOK));
-
-  MOZ_ASSERT(NS_IsMainThread(), "not main thread");
-  MOZ_ASSERT(!mRecvdHttpUpgradeTransport,
-             "OnWebSocketConnectionAvailable duplicated");
-  MOZ_ASSERT(aConnection,
-             "OnWebSocketConnectionAvailable with invalid connection");
+  mTransport = aTransport;
+  mSocketIn = aSocketIn;
+  mSocketOut = aSocketOut;
 
-  if (mStopped) {
-    LOG(("WebSocketChannel::OnWebSocketConnectionAvailable: Already stopped"));
-    aConnection->Close();
-    return NS_OK;
-  }
-
-  mConnection = aConnection;
-  nsresult rv = mConnection->Init(this, mSocketThread);
+  nsresult rv;
+  rv = mTransport->SetEventSink(nullptr, nullptr);
+  if (NS_FAILED(rv)) return rv;
+  rv = mTransport->SetSecurityCallbacks(this);
   if (NS_FAILED(rv)) return rv;
 
   mRecvdHttpUpgradeTransport = 1;
@@ -3846,40 +3863,133 @@ WebSocketChannel::OnStopRequest(nsIRequest* aRequest, nsresult aStatusCode) {
   return NS_OK;
 }
 
-void WebSocketChannel::DoEnqueueOutgoingMessage() {
-  LOG(("WebSocketChannel::DoEnqueueOutgoingMessage() %p\n", this));
+// nsIInputStreamCallback
+
+NS_IMETHODIMP
+WebSocketChannel::OnInputStreamReady(nsIAsyncInputStream* aStream) {
+  LOG(("WebSocketChannel::OnInputStreamReady() %p\n", this));
   MOZ_ASSERT(OnSocketThread(), "not on socket thread");
 
-  if (!mCurrentOut) PrimeNewOutgoingMessage();
+  if (!mSocketIn)  // did we we clean up the socket after scheduling InputReady?
+    return NS_OK;
 
-  while (mCurrentOut && mConnection) {
-    LOG(
-        ("WebSocketChannel::DoEnqueueOutgoingMessage: "
-         "Try to send %u of hdr/copybreak and %u of data\n",
-         mHdrOutSize, mCurrentOut->Length()));
+  // this is after the http upgrade - so we are speaking websockets
+  char buffer[2048];
+  uint32_t count;
+  nsresult rv;
 
-    nsresult rv = mConnection->EnqueueOutputData(
-        mHdrOut, mHdrOutSize, (uint8_t*)mCurrentOut->BeginReading(),
-        mCurrentOut->Length());
+  do {
+    rv = mSocketIn->Read((char*)buffer, 2048, &count);
+    LOG(("WebSocketChannel::OnInputStreamReady: read %u rv %" PRIx32 "\n",
+         count, static_cast<uint32_t>(rv)));
 
-    LOG(("WebSocketChannel::DoEnqueueOutgoingMessage: rv %" PRIx32 "\n",
-         static_cast<uint32_t>(rv)));
+    if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
+      mSocketIn->AsyncWait(this, 0, 0, mSocketThread);
+      return NS_OK;
+    }
 
     if (NS_FAILED(rv)) {
       AbortSession(rv);
-      return;
+      return rv;
     }
 
-    if (!mStopped) {
-      mTargetThread->Dispatch(
-          new CallAcknowledge(this, mCurrentOut->OrigLength()),
-          NS_DISPATCH_NORMAL);
+    if (count == 0) {
+      AbortSession(NS_BASE_STREAM_CLOSED);
+      return NS_OK;
+    }
+
+    if (mStopped) {
+      continue;
+    }
+
+    rv = ProcessInput((uint8_t*)buffer, count);
+    if (NS_FAILED(rv)) {
+      AbortSession(rv);
+      return rv;
+    }
+  } while (NS_SUCCEEDED(rv) && mSocketIn);
+
+  return NS_OK;
+}
+
+// nsIOutputStreamCallback
+
+NS_IMETHODIMP
+WebSocketChannel::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
+  LOG(("WebSocketChannel::OnOutputStreamReady() %p\n", this));
+  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
+  nsresult rv;
+
+  if (!mCurrentOut) PrimeNewOutgoingMessage();
+
+  while (mCurrentOut && mSocketOut) {
+    const char* sndBuf;
+    uint32_t toSend;
+    uint32_t amtSent;
+
+    if (mHdrOut) {
+      sndBuf = (const char*)mHdrOut;
+      toSend = mHdrOutToSend;
+      LOG(
+          ("WebSocketChannel::OnOutputStreamReady: "
+           "Try to send %u of hdr/copybreak\n",
+           toSend));
+    } else {
+      sndBuf = (char*)mCurrentOut->BeginReading() + mCurrentOutSent;
+      toSend = mCurrentOut->Length() - mCurrentOutSent;