Add support of sending blob data for RTCDataChannel

Source/modules/mediastream/RTCDataChannel.cpp

 /*
  * Copyright (C) 2012 Google 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
@@ skipping 13 matching lines @@
 
 #include "config.h"
 #include "modules/mediastream/RTCDataChannel.h"
 
 #include "bindings/v8/ExceptionState.h"
 #include "core/dom/ExceptionCode.h"
 #include "core/dom/ExecutionContext.h"
 #include "core/events/Event.h"
 #include "core/events/MessageEvent.h"
 #include "core/fileapi/Blob.h"
+#include "core/fileapi/FileReaderLoader.h"
 #include "public/platform/WebRTCPeerConnectionHandler.h"
 #include "wtf/ArrayBuffer.h"
 #include "wtf/ArrayBufferView.h"
+#include "wtf/OwnPtr.h"
 
 namespace WebCore {
 
 static void throwNotOpenException(ExceptionState& exceptionState)
 {
     exceptionState.throwDOMException(InvalidStateError, "RTCDataChannel.readyState is not 'open'");
 }
 
 static void throwCouldNotSendDataException(ExceptionState& exceptionState)
 {
     exceptionState.throwDOMException(NetworkError, "Could not send data");
 }
 
-static void throwNoBlobSupportException(ExceptionState& exceptionState)
-{
-    exceptionState.throwDOMException(NotSupportedError, "Blob support not implemented yet");
-}
-
 PassRefPtr<RTCDataChannel> RTCDataChannel::create(ExecutionContext* context, PassOwnPtr<blink::WebRTCDataChannelHandler> handler)
 {
     ASSERT(handler);
     return adoptRef(new RTCDataChannel(context, handler));
 }
 
 PassRefPtr<RTCDataChannel> RTCDataChannel::create(ExecutionContext* context, blink::WebRTCPeerConnectionHandler* peerConnectionHandler, const String& label, const blink::WebRTCDataChannelInit& init, ExceptionState& exceptionState)
 {
     OwnPtr<blink::WebRTCDataChannelHandler> handler = adoptPtr(peerConnectionHandler->createDataChannel(label, init));
     if (!handler) {
         exceptionState.throwDOMException(NotSupportedError, "RTCDataChannel is not supported");
         return nullptr;
     }
     return adoptRef(new RTCDataChannel(context, handler.release()));
 }
 
 RTCDataChannel::RTCDataChannel(ExecutionContext* context, PassOwnPtr<blink::WebRTCDataChannelHandler> handler)
     : m_executionContext(context)
     , m_handler(handler)
     , m_stopped(false)
     , m_readyState(ReadyStateConnecting)
-    , m_binaryType(BinaryTypeArrayBuffer)
+    , m_binaryType(BinaryTypeBlob)
+    , m_blobLoaderStatus(BlobLoaderNotStarted)
     , m_scheduledEventTimer(this, &RTCDataChannel::scheduledEventTimerFired)
 {
     ScriptWrappable::init(this);
     m_handler->setClient(this);
 }
 
 RTCDataChannel::~RTCDataChannel()
 {
 }
 
@@ skipping 67 matching lines @@
     case BinaryTypeArrayBuffer:
         return "arraybuffer";
     }
     ASSERT_NOT_REACHED();
     return String();
 }
 
 void RTCDataChannel::setBinaryType(const String& binaryType, ExceptionState& exceptionState)
 {
     if (binaryType == "blob")
-        throwNoBlobSupportException(exceptionState);
+        m_binaryType = BinaryTypeBlob;
     else if (binaryType == "arraybuffer")
         m_binaryType = BinaryTypeArrayBuffer;
     else
         exceptionState.throwDOMException(TypeMismatchError, "Unknown binary type : " + binaryType);
 }
 
 void RTCDataChannel::send(const String& data, ExceptionState& exceptionState)
 {
     if (m_readyState != ReadyStateOpen) {
         throwNotOpenException(exceptionState);
@@ skipping 27 matching lines @@
 void RTCDataChannel::send(PassRefPtr<ArrayBufferView> data, ExceptionState& exceptionState)
 {
     if (!m_handler->sendRawData(static_cast<const char*>(data->baseAddress()), data->byteLength())) {
         // FIXME: This should not throw an exception but instead forcefully close the data channel.
         throwCouldNotSendDataException(exceptionState);
     }
 }
 
 void RTCDataChannel::send(PassRefPtrWillBeRawPtr<Blob> data, ExceptionState& exceptionState)
 {
-    // FIXME: implement
-    throwNoBlobSupportException(exceptionState);
+    if (m_readyState != ReadyStateOpen) {
+        throwNotOpenException(exceptionState);
+        return;
+    }
+    m_outgoingBlobQueue.append(data->blobDataHandle());
+    processOutgoingBlobQueue();
 }
 
 void RTCDataChannel::close()
 {
     if (m_stopped)
         return;
 
     m_handler->close();
 }
 
@@ skipping 23 matching lines @@
 
     scheduleDispatchEvent(MessageEvent::create(text));
 }
 
 void RTCDataChannel::didReceiveRawData(const char* data, size_t dataLength)
 {
     if (m_stopped)
         return;
 
     if (m_binaryType == BinaryTypeBlob) {
-        // FIXME: Implement.
+        OwnPtr<BlobData> blobData = BlobData::create();
+        RefPtr<RawData> rawData = RawData::create();
+        OwnPtr<Vector<char> > binaryData = adoptPtr(new Vector<char>(dataLength));

[michaeln, 2014/05/06 02:36:49]

this also probably won't work well with large blobs

[Li Yin, 2014/05/06 07:09:37]

When blob is large, it needs much memory, do you mean we need a protection for
out of memory? I found WebSocket and XHR meets the same issue.

+        memcpy(binaryData->data(), data, dataLength);
+        binaryData->swap(*rawData->mutableData());
+        blobData->appendData(rawData, 0, BlobDataItem::toEndOfFile);
+        RefPtrWillBeRawPtr<Blob> blob = Blob::create(BlobDataHandle::create(blobData.release(), dataLength));
+        scheduleDispatchEvent(MessageEvent::create(blob.release()));
         return;
     }
     if (m_binaryType == BinaryTypeArrayBuffer) {
         RefPtr<ArrayBuffer> buffer = ArrayBuffer::create(data, dataLength);
         scheduleDispatchEvent(MessageEvent::create(buffer.release()));
         return;
     }
     ASSERT_NOT_REACHED();
 }
 
@@ skipping 39 matching lines @@
     WillBeHeapVector<RefPtrWillBeMember<Event> > events;
     events.swap(m_scheduledEvents);
 
     WillBeHeapVector<RefPtrWillBeMember<Event> >::iterator it = events.begin();
     for (; it != events.end(); ++it)
         dispatchEvent((*it).release());
 
     events.clear();
 }
 
+void RTCDataChannel::didStartLoading()
+{
+    ASSERT(m_blobLoader);
+    ASSERT(m_blobLoaderStatus == BlobLoaderStarted);
+}
+
+void RTCDataChannel::didReceiveData()
+{
+    ASSERT(m_blobLoader);
+    ASSERT(m_blobLoaderStatus == BlobLoaderStarted);
+}
+
+void RTCDataChannel::didFinishLoading()
+{
+    ASSERT(m_blobLoader);
+    ASSERT(m_blobLoaderStatus == BlobLoaderStarted);
+    m_blobLoaderStatus = BlobLoaderFinished;
+    processOutgoingBlobQueue();
+    deref();
+}
+
+void RTCDataChannel::didFail(FileError::ErrorCode errorCode)
+{
+    ASSERT(m_blobLoader);
+    ASSERT(m_blobLoaderStatus == BlobLoaderStarted);
+    m_blobLoader.clear();
+    m_blobLoaderStatus = BlobLoaderFailed;
+    close();
+    deref();
+}
+
+void RTCDataChannel::processOutgoingBlobQueue()
+{
+    while (!m_outgoingBlobQueue.isEmpty()) {
+        RefPtr<BlobDataHandle> blobDataHandle = m_outgoingBlobQueue.takeFirst();
+        switch (m_blobLoaderStatus) {
+        case BlobLoaderNotStarted:
+            ref(); // Will be derefed after didFinishLoading() or didFail().
+            ASSERT(!m_blobLoader);
+            m_blobLoader = adoptPtr(new FileReaderLoader(FileReaderLoader::ReadAsArrayBuffer, this));
+            m_blobLoaderStatus = BlobLoaderStarted;
+            m_blobLoader->start(m_executionContext, blobDataHandle.get());
+            m_outgoingBlobQueue.prepend(blobDataHandle.release());
+            return;
+
+        case BlobLoaderStarted:
+        case BlobLoaderFailed:
+            m_outgoingBlobQueue.prepend(blobDataHandle.release());
+            return;
+
+        case BlobLoaderFinished: {
+            RefPtr<ArrayBuffer> result = m_blobLoader->arrayBufferResult();

[michaeln, 2014/05/06 02:36:49]

this probably won't work well with large blobs, have you considered while
(!done) { readsome(); sendsome(); } to avoid having to read the entire thing
into memory at once?

[Li Yin, 2014/05/06 07:09:37]

Yeah, your concern looks reasonable, but unfortunately, DataChannelInterface in
libjingle doesn't have the splitting capability.
DataChannelInterface::Send() means data was sent completely, the remote peer
will receive a message.

+            m_blobLoader.clear();
+            m_blobLoaderStatus = BlobLoaderNotStarted;
+            if (!m_handler->sendRawData(static_cast<const char*>(result->data()), result->byteLength())) {
+                // FIXME: This should forcefully close the data channel.
+                return;
+            }
+            break;
+        }
+        }
+    }
+}
+
 } // namespace WebCore