xpcom/io/NonBlockingAsyncInputStream.cpp

   1 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* This Source Code Form is subject to the terms of the Mozilla Public
   3  * License, v. 2.0. If a copy of the MPL was not distributed with this
   4  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   5
   6 #include "NonBlockingAsyncInputStream.h"
   7 #include "mozilla/ipc/InputStreamUtils.h"
   8 #include "nsIAsyncInputStream.h"
   9 #include "nsICloneableInputStream.h"
  10 #include "nsIInputStream.h"
  11 #include "nsIIPCSerializableInputStream.h"
  12 #include "nsISeekableStream.h"
  13 #include "nsStreamUtils.h"
  14
  15 namespace mozilla {
  16
  17 using namespace ipc;
  18
  19 class NonBlockingAsyncInputStream::AsyncWaitRunnable final
  20     : public CancelableRunnable {
  21   RefPtr<NonBlockingAsyncInputStream> mStream;
  22   nsCOMPtr<nsIInputStreamCallback> mCallback;
  23
  24  public:
  25   AsyncWaitRunnable(NonBlockingAsyncInputStream* aStream,
  26                     nsIInputStreamCallback* aCallback)
  27       : CancelableRunnable("AsyncWaitRunnable"),
  28         mStream(aStream),
  29         mCallback(aCallback) {}
  30
  31   NS_IMETHOD
  32   Run() override {
  33     mStream->RunAsyncWaitCallback(this, mCallback.forget());
  34     return NS_OK;
  35   }
  36
  37   nsresult Cancel() override {
  38     mStream = nullptr;
  39     return NS_OK;
  40   }
  41 };
  42
  43 NS_IMPL_ADDREF(NonBlockingAsyncInputStream);
  44 NS_IMPL_RELEASE(NonBlockingAsyncInputStream);
  45
  46 NonBlockingAsyncInputStream::WaitClosureOnly::WaitClosureOnly(
  47     AsyncWaitRunnable* aRunnable, nsIEventTarget* aEventTarget)
  48     : mRunnable(aRunnable), mEventTarget(aEventTarget) {}
  49
  50 NS_INTERFACE_MAP_BEGIN(NonBlockingAsyncInputStream)
  51   NS_INTERFACE_MAP_ENTRY(nsIInputStream)
  52   NS_INTERFACE_MAP_ENTRY(nsIAsyncInputStream)
  53   NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsICloneableInputStream,
  54                                      mWeakCloneableInputStream)
  55   NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsIIPCSerializableInputStream,
  56                                      mWeakIPCSerializableInputStream)
  57   NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsISeekableStream,
  58                                      mWeakSeekableInputStream)
  59   NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsITellableStream,
  60                                      mWeakTellableInputStream)
  61   NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIInputStream)
  62 NS_INTERFACE_MAP_END
  63
  64 /* static */
  65 nsresult NonBlockingAsyncInputStream::Create(
  66     already_AddRefed<nsIInputStream> aInputStream,
  67     nsIAsyncInputStream** aResult) {
  68   MOZ_DIAGNOSTIC_ASSERT(aResult);
  69
  70   nsCOMPtr<nsIInputStream> inputStream = std::move(aInputStream);
  71
  72   bool nonBlocking = false;
  73   nsresult rv = inputStream->IsNonBlocking(&nonBlocking);
  74   if (NS_WARN_IF(NS_FAILED(rv))) {
  75     return rv;
  76   }
  77
  78   MOZ_DIAGNOSTIC_ASSERT(nonBlocking);
  79
  80 #ifdef MOZ_DIAGNOSTIC_ASSERT_ENABLED
  81   nsCOMPtr<nsIAsyncInputStream> asyncInputStream =
  82       do_QueryInterface(inputStream);
  83   MOZ_DIAGNOSTIC_ASSERT(!asyncInputStream);
  84 #endif  // MOZ_DIAGNOSTIC_ASSERT_ENABLED
  85
  86   RefPtr<NonBlockingAsyncInputStream> stream =
  87       new NonBlockingAsyncInputStream(inputStream.forget());
  88
  89   stream.forget(aResult);
  90   return NS_OK;
  91 }
  92
  93 NonBlockingAsyncInputStream::NonBlockingAsyncInputStream(
  94     already_AddRefed<nsIInputStream> aInputStream)
  95     : mInputStream(std::move(aInputStream)),
  96       mWeakCloneableInputStream(nullptr),
  97       mWeakIPCSerializableInputStream(nullptr),
  98       mWeakSeekableInputStream(nullptr),
  99       mWeakTellableInputStream(nullptr),
 100       mLock("NonBlockingAsyncInputStream::mLock"),
 101       mClosed(false) {
 102   MOZ_ASSERT(mInputStream);
 103
 104   nsCOMPtr<nsICloneableInputStream> cloneableStream =
 105       do_QueryInterface(mInputStream);
 106   if (cloneableStream && SameCOMIdentity(mInputStream, cloneableStream)) {
 107     mWeakCloneableInputStream = cloneableStream;
 108   }
 109
 110   nsCOMPtr<nsIIPCSerializableInputStream> serializableStream =
 111       do_QueryInterface(mInputStream);
 112   if (serializableStream && SameCOMIdentity(mInputStream, serializableStream)) {
 113     mWeakIPCSerializableInputStream = serializableStream;
 114   }
 115
 116   nsCOMPtr<nsISeekableStream> seekableStream = do_QueryInterface(mInputStream);
 117   if (seekableStream && SameCOMIdentity(mInputStream, seekableStream)) {
 118     mWeakSeekableInputStream = seekableStream;
 119   }
 120
 121   nsCOMPtr<nsITellableStream> tellableStream = do_QueryInterface(mInputStream);
 122   if (tellableStream && SameCOMIdentity(mInputStream, tellableStream)) {
 123     mWeakTellableInputStream = tellableStream;
 124   }
 125 }
 126
 127 NonBlockingAsyncInputStream::~NonBlockingAsyncInputStream() = default;
 128
 129 NS_IMETHODIMP
 130 NonBlockingAsyncInputStream::Close() {
 131   RefPtr<AsyncWaitRunnable> waitClosureOnlyRunnable;
 132   nsCOMPtr<nsIEventTarget> waitClosureOnlyEventTarget;
 133
 134   {
 135     MutexAutoLock lock(mLock);
 136
 137     if (mClosed) {
 138       // Here we could return NS_BASE_STREAM_CLOSED as well, but just to avoid
 139       // warning messages, let's make everybody happy with a NS_OK.
 140       return NS_OK;
 141     }
 142
 143     mClosed = true;
 144
 145     NS_ENSURE_STATE(mInputStream);
 146     nsresult rv = mInputStream->Close();
 147     if (NS_WARN_IF(NS_FAILED(rv))) {
 148       mWaitClosureOnly.reset();
 149       return rv;
 150     }
 151
 152     // If we have a WaitClosureOnly runnable, it's time to use it.
 153     if (mWaitClosureOnly.isSome()) {
 154       waitClosureOnlyRunnable = std::move(mWaitClosureOnly->mRunnable);
 155       waitClosureOnlyEventTarget = std::move(mWaitClosureOnly->mEventTarget);
 156
 157       mWaitClosureOnly.reset();
 158
 159       // Now we want to dispatch the asyncWaitCallback.
 160       mAsyncWaitCallback = waitClosureOnlyRunnable;
 161     }
 162   }
 163
 164   if (waitClosureOnlyRunnable) {
 165     if (waitClosureOnlyEventTarget) {
 166       waitClosureOnlyEventTarget->Dispatch(waitClosureOnlyRunnable,
 167                                            NS_DISPATCH_NORMAL);
 168     } else {
 169       waitClosureOnlyRunnable->Run();
 170     }
 171   }
 172
 173   return NS_OK;
 174 }
 175
 176 // nsIInputStream interface
 177
 178 NS_IMETHODIMP
 179 NonBlockingAsyncInputStream::Available(uint64_t* aLength) {
 180   nsresult rv = mInputStream->Available(aLength);
 181   // Don't issue warnings for legal condition NS_BASE_STREAM_CLOSED.
 182   if (rv == NS_BASE_STREAM_CLOSED || NS_WARN_IF(NS_FAILED(rv))) {
 183     return rv;
 184   }
 185
 186   // Nothing more to read. Let's close the stream now.
 187   if (*aLength == 0) {
 188     MutexAutoLock lock(mLock);
 189     mInputStream->Close();
 190     mClosed = true;
 191     return NS_BASE_STREAM_CLOSED;
 192   }
 193
 194   return NS_OK;
 195 }
 196
 197 NS_IMETHODIMP
 198 NonBlockingAsyncInputStream::StreamStatus() {
 199   return mInputStream->StreamStatus();
 200 }
 201
 202 NS_IMETHODIMP
 203 NonBlockingAsyncInputStream::Read(char* aBuffer, uint32_t aCount,
 204                                   uint32_t* aReadCount) {
 205   return mInputStream->Read(aBuffer, aCount, aReadCount);
 206 }
 207
 208 namespace {
 209
 210 class MOZ_RAII ReadSegmentsData {
 211  public:
 212   ReadSegmentsData(NonBlockingAsyncInputStream* aStream,
 213                    nsWriteSegmentFun aFunc, void* aClosure)
 214       : mStream(aStream), mFunc(aFunc), mClosure(aClosure) {}
 215
 216   NonBlockingAsyncInputStream* mStream;
 217   nsWriteSegmentFun mFunc;
 218   void* mClosure;
 219 };
 220
 221 nsresult ReadSegmentsWriter(nsIInputStream* aInStream, void* aClosure,
 222                             const char* aFromSegment, uint32_t aToOffset,
 223                             uint32_t aCount, uint32_t* aWriteCount) {
 224   ReadSegmentsData* data = static_cast<ReadSegmentsData*>(aClosure);
 225   return data->mFunc(data->mStream, data->mClosure, aFromSegment, aToOffset,
 226                      aCount, aWriteCount);
 227 }
 228
 229 }  // namespace
 230
 231 NS_IMETHODIMP
 232 NonBlockingAsyncInputStream::ReadSegments(nsWriteSegmentFun aWriter,
 233                                           void* aClosure, uint32_t aCount,
 234                                           uint32_t* aResult) {
 235   ReadSegmentsData data(this, aWriter, aClosure);
 236   return mInputStream->ReadSegments(ReadSegmentsWriter, &data, aCount, aResult);
 237 }
 238
 239 NS_IMETHODIMP
 240 NonBlockingAsyncInputStream::IsNonBlocking(bool* aNonBlocking) {
 241   *aNonBlocking = true;
 242   return NS_OK;
 243 }
 244
 245 // nsICloneableInputStream interface
 246
 247 NS_IMETHODIMP
 248 NonBlockingAsyncInputStream::GetCloneable(bool* aCloneable) {
 249   NS_ENSURE_STATE(mWeakCloneableInputStream);
 250   return mWeakCloneableInputStream->GetCloneable(aCloneable);
 251 }
 252
 253 NS_IMETHODIMP
 254 NonBlockingAsyncInputStream::Clone(nsIInputStream** aResult) {
 255   NS_ENSURE_STATE(mWeakCloneableInputStream);
 256
 257   nsCOMPtr<nsIInputStream> clonedStream;
 258   nsresult rv = mWeakCloneableInputStream->Clone(getter_AddRefs(clonedStream));
 259   if (NS_WARN_IF(NS_FAILED(rv))) {
 260     return rv;
 261   }
 262
 263   nsCOMPtr<nsIAsyncInputStream> asyncStream;
 264   rv = Create(clonedStream.forget(), getter_AddRefs(asyncStream));
 265   if (NS_WARN_IF(NS_FAILED(rv))) {
 266     return rv;
 267   }
 268
 269   asyncStream.forget(aResult);
 270   return NS_OK;
 271 }
 272
 273 // nsIAsyncInputStream interface
 274
 275 NS_IMETHODIMP
 276 NonBlockingAsyncInputStream::CloseWithStatus(nsresult aStatus) {
 277   return Close();
 278 }
 279
 280 NS_IMETHODIMP
 281 NonBlockingAsyncInputStream::AsyncWait(nsIInputStreamCallback* aCallback,
 282                                        uint32_t aFlags,
 283                                        uint32_t aRequestedCount,
 284                                        nsIEventTarget* aEventTarget) {
 285   RefPtr<AsyncWaitRunnable> runnable;
 286   {
 287     MutexAutoLock lock(mLock);
 288
 289     mWaitClosureOnly.reset();
 290     mAsyncWaitCallback = nullptr;
 291
 292     if (!aCallback) {
 293       // Canceling previous callbacks, which is done above.
 294       return NS_OK;
 295     }
 296
 297     // Maybe the stream is already closed.
 298     if (!mClosed) {
 299       uint64_t length;
 300       nsresult rv = mInputStream->Available(&length);
 301       if (NS_SUCCEEDED(rv) && length == 0) {
 302         mInputStream->Close();
 303         mClosed = true;
 304       }
 305     }
 306
 307     runnable = new AsyncWaitRunnable(this, aCallback);
 308     if ((aFlags & nsIAsyncInputStream::WAIT_CLOSURE_ONLY) && !mClosed) {
 309       mWaitClosureOnly.emplace(runnable, aEventTarget);
 310       return NS_OK;
 311     }
 312
 313     mAsyncWaitCallback = runnable;
 314   }
 315
 316   MOZ_ASSERT(runnable);
 317
 318   if (aEventTarget) {
 319     return aEventTarget->Dispatch(runnable.forget());
 320   }
 321
 322   return runnable->Run();
 323 }
 324
 325 // nsIIPCSerializableInputStream
 326
 327 void NonBlockingAsyncInputStream::SerializedComplexity(
 328     uint32_t aMaxSize, uint32_t* aSizeUsed, uint32_t* aPipes,
 329     uint32_t* aTransferables) {
 330   InputStreamHelper::SerializedComplexity(mInputStream, aMaxSize, aSizeUsed,
 331                                           aPipes, aTransferables);
 332 }
 333
 334 void NonBlockingAsyncInputStream::Serialize(
 335     mozilla::ipc::InputStreamParams& aParams, uint32_t aMaxSize,
 336     uint32_t* aSizeUsed) {
 337   MOZ_ASSERT(mWeakIPCSerializableInputStream);
 338   InputStreamHelper::SerializeInputStream(mInputStream, aParams, aMaxSize,
 339                                           aSizeUsed);
 340 }
 341
 342 bool NonBlockingAsyncInputStream::Deserialize(
 343     const mozilla::ipc::InputStreamParams& aParams) {
 344   MOZ_CRASH("NonBlockingAsyncInputStream cannot be deserialized!");
 345   return true;
 346 }
 347
 348 // nsISeekableStream
 349
 350 NS_IMETHODIMP
 351 NonBlockingAsyncInputStream::Seek(int32_t aWhence, int64_t aOffset) {
 352   NS_ENSURE_STATE(mWeakSeekableInputStream);
 353   return mWeakSeekableInputStream->Seek(aWhence, aOffset);
 354 }
 355
 356 NS_IMETHODIMP
 357 NonBlockingAsyncInputStream::SetEOF() {
 358   NS_ENSURE_STATE(mWeakSeekableInputStream);
 359   return NS_ERROR_NOT_IMPLEMENTED;
 360 }
 361
 362 // nsITellableStream
 363
 364 NS_IMETHODIMP
 365 NonBlockingAsyncInputStream::Tell(int64_t* aResult) {
 366   NS_ENSURE_STATE(mWeakTellableInputStream);
 367   return mWeakTellableInputStream->Tell(aResult);
 368 }
 369
 370 void NonBlockingAsyncInputStream::RunAsyncWaitCallback(
 371     NonBlockingAsyncInputStream::AsyncWaitRunnable* aRunnable,
 372     already_AddRefed<nsIInputStreamCallback> aCallback) {
 373   nsCOMPtr<nsIInputStreamCallback> callback = std::move(aCallback);
 374
 375   {
 376     MutexAutoLock lock(mLock);
 377     if (mAsyncWaitCallback != aRunnable) {
 378       // The callback has been canceled in the meantime.
 379       return;
 380     }
 381
 382     mAsyncWaitCallback = nullptr;
 383   }
 384
 385   callback->OnInputStreamReady(this);
 386 }
 387
 388 }  // namespace mozilla