summaryrefslogtreecommitdiffstats
path: root/netwerk/sctp/datachannel/DataChannel.h
diff options
context:
space:
mode:
Diffstat (limited to 'netwerk/sctp/datachannel/DataChannel.h')
-rw-r--r--netwerk/sctp/datachannel/DataChannel.h670
1 files changed, 670 insertions, 0 deletions
diff --git a/netwerk/sctp/datachannel/DataChannel.h b/netwerk/sctp/datachannel/DataChannel.h
new file mode 100644
index 0000000000..76fb6aeced
--- /dev/null
+++ b/netwerk/sctp/datachannel/DataChannel.h
@@ -0,0 +1,670 @@
+/* -*- 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/. */
+
+#ifndef NETWERK_SCTP_DATACHANNEL_DATACHANNEL_H_
+#define NETWERK_SCTP_DATACHANNEL_DATACHANNEL_H_
+
+#ifdef MOZ_WEBRTC_SIGNALING
+# define SCTP_DTLS_SUPPORTED 1
+#endif
+
+#include <memory>
+#include <string>
+#include <vector>
+#include <errno.h>
+#include "nsISupports.h"
+#include "nsCOMPtr.h"
+#include "mozilla/UniquePtr.h"
+#include "mozilla/WeakPtr.h"
+#include "nsString.h"
+#include "nsThreadUtils.h"
+#include "nsTArray.h"
+#include "nsDeque.h"
+#include "mozilla/dom/Blob.h"
+#include "mozilla/Mutex.h"
+#include "DataChannelProtocol.h"
+#include "DataChannelListener.h"
+#include "mozilla/net/NeckoTargetHolder.h"
+#include "DataChannelLog.h"
+
+#ifdef SCTP_DTLS_SUPPORTED
+# include "transport/sigslot.h"
+# include "transport/transportlayer.h" // For TransportLayer::State
+#endif
+
+#ifndef EALREADY
+# define EALREADY WSAEALREADY
+#endif
+
+extern "C" {
+struct socket;
+struct sctp_rcvinfo;
+}
+
+namespace mozilla {
+
+class DataChannelConnection;
+class DataChannel;
+class DataChannelOnMessageAvailable;
+class MediaPacket;
+class MediaTransportHandler;
+namespace dom {
+struct RTCStatsCollection;
+};
+
+enum class DataChannelState { Connecting, Open, Closing, Closed };
+enum class DataChannelConnectionState { Connecting, Open, Closed };
+enum class DataChannelReliabilityPolicy {
+ Reliable,
+ LimitedRetransmissions,
+ LimitedLifetime
+};
+
+// For sending outgoing messages.
+// This class only holds a reference to the data and the info structure but does
+// not copy it.
+class OutgoingMsg {
+ public:
+ OutgoingMsg(struct sctp_sendv_spa& info, const uint8_t* data, size_t length);
+ OutgoingMsg(OutgoingMsg&& other) = default;
+ OutgoingMsg& operator=(OutgoingMsg&& other) = default;
+ ~OutgoingMsg() = default;
+
+ void Advance(size_t offset);
+ struct sctp_sendv_spa& GetInfo() const { return *mInfo; };
+ size_t GetLength() const { return mLength; };
+ size_t GetLeft() const { return mLength - mPos; };
+ const uint8_t* GetData() const { return (const uint8_t*)(mData + mPos); };
+
+ protected:
+ OutgoingMsg() // Use this for inheritance only
+ : mLength(0), mData(nullptr), mInfo(nullptr), mPos(0){};
+ size_t mLength;
+ const uint8_t* mData;
+ struct sctp_sendv_spa* mInfo;
+ size_t mPos;
+};
+
+// For queuing outgoing messages
+// This class copies data of an outgoing message.
+class BufferedOutgoingMsg : public OutgoingMsg {
+ public:
+ explicit BufferedOutgoingMsg(OutgoingMsg& msg);
+ BufferedOutgoingMsg(BufferedOutgoingMsg&& other) = default;
+ BufferedOutgoingMsg& operator=(BufferedOutgoingMsg&& other) = default;
+ ~BufferedOutgoingMsg();
+};
+
+// for queuing incoming data messages before the Open or
+// external negotiation is indicated to us
+class QueuedDataMessage {
+ public:
+ QueuedDataMessage(uint16_t stream, uint32_t ppid, int flags, const void* data,
+ uint32_t length)
+ : mStream(stream), mPpid(ppid), mFlags(flags), mLength(length) {
+ mData = static_cast<uint8_t*>(moz_xmalloc((size_t)length)); // infallible
+ memcpy(mData, data, (size_t)length);
+ }
+ QueuedDataMessage(QueuedDataMessage&& other) = default;
+ QueuedDataMessage& operator=(QueuedDataMessage&& other) = default;
+ ~QueuedDataMessage() { free(mData); }
+
+ uint16_t mStream;
+ uint32_t mPpid;
+ int mFlags;
+ uint32_t mLength;
+ uint8_t* mData;
+};
+
+// One per PeerConnection
+class DataChannelConnection final : public net::NeckoTargetHolder
+#ifdef SCTP_DTLS_SUPPORTED
+ ,
+ public sigslot::has_slots<>
+#endif
+{
+ friend class DataChannel;
+ friend class DataChannelOnMessageAvailable;
+ friend class DataChannelConnectRunnable;
+
+ virtual ~DataChannelConnection();
+
+ public:
+ enum class PendingType {
+ None, // No outgoing messages are pending.
+ Dcep, // Outgoing DCEP messages are pending.
+ Data, // Outgoing data channel messages are pending.
+ };
+
+ class DataConnectionListener : public SupportsWeakPtr {
+ public:
+ virtual ~DataConnectionListener() = default;
+
+ // Called when a new DataChannel has been opened by the other side.
+ virtual void NotifyDataChannel(already_AddRefed<DataChannel> channel) = 0;
+
+ // Called when a DataChannel transitions to state open
+ virtual void NotifyDataChannelOpen(DataChannel* aChannel) = 0;
+
+ // Called when a DataChannel (that was open at some point in the past)
+ // transitions to state closed
+ virtual void NotifyDataChannelClosed(DataChannel* aChannel) = 0;
+
+ // Called when SCTP connects
+ virtual void NotifySctpConnected() = 0;
+
+ // Called when SCTP closes
+ virtual void NotifySctpClosed() = 0;
+ };
+
+ // Create a new DataChannel Connection
+ // Must be called on Main thread
+ static Maybe<RefPtr<DataChannelConnection>> Create(
+ DataConnectionListener* aListener, nsISerialEventTarget* aTarget,
+ MediaTransportHandler* aHandler, const uint16_t aLocalPort,
+ const uint16_t aNumStreams, const Maybe<uint64_t>& aMaxMessageSize);
+
+ DataChannelConnection(const DataChannelConnection&) = delete;
+ DataChannelConnection& operator=(const DataChannelConnection&) = delete;
+
+ NS_INLINE_DECL_THREADSAFE_REFCOUNTING(DataChannelConnection)
+
+ void Destroy(); // So we can spawn refs tied to runnables in shutdown
+ // Finish Destroy on STS to avoid SCTP race condition with ABORT from far end
+ void DestroyOnSTS(struct socket* aMasterSocket, struct socket* aSocket);
+ void DestroyOnSTSFinal();
+
+ void SetMaxMessageSize(bool aMaxMessageSizeSet, uint64_t aMaxMessageSize);
+ uint64_t GetMaxMessageSize();
+
+ void AppendStatsToReport(const UniquePtr<dom::RTCStatsCollection>& aReport,
+ const DOMHighResTimeStamp aTimestamp) const;
+#ifdef ALLOW_DIRECT_SCTP_LISTEN_CONNECT
+ // These block; they require something to decide on listener/connector
+ // (though you can do simultaneous Connect()). Do not call these from
+ // the main thread!
+ bool Listen(unsigned short port);
+ bool Connect(const char* addr, unsigned short port);
+#endif
+
+#ifdef SCTP_DTLS_SUPPORTED
+ bool ConnectToTransport(const std::string& aTransportId, const bool aClient,
+ const uint16_t aLocalPort,
+ const uint16_t aRemotePort);
+ void TransportStateChange(const std::string& aTransportId,
+ TransportLayer::State aState);
+ void CompleteConnect();
+ void SetSignals(const std::string& aTransportId);
+#endif
+
+ [[nodiscard]] already_AddRefed<DataChannel> Open(
+ const nsACString& label, const nsACString& protocol,
+ DataChannelReliabilityPolicy prPolicy, bool inOrder, uint32_t prValue,
+ DataChannelListener* aListener, nsISupports* aContext,
+ bool aExternalNegotiated, uint16_t aStream);
+
+ void Stop();
+ void Close(DataChannel* aChannel);
+ void CloseLocked(DataChannel* aChannel) MOZ_REQUIRES(mLock);
+ void CloseAll();
+
+ // Returns a POSIX error code.
+ int SendMsg(uint16_t stream, const nsACString& aMsg) {
+ return SendDataMsgCommon(stream, aMsg, false);
+ }
+
+ // Returns a POSIX error code.
+ int SendBinaryMsg(uint16_t stream, const nsACString& aMsg) {
+ return SendDataMsgCommon(stream, aMsg, true);
+ }
+
+ // Returns a POSIX error code.
+ int SendBlob(uint16_t stream, nsIInputStream* aBlob);
+
+ // Called on data reception from the SCTP library
+ // must(?) be public so my c->c++ trampoline can call it
+ // May be called with (STS thread) or without the lock
+ int ReceiveCallback(struct socket* sock, void* data, size_t datalen,
+ struct sctp_rcvinfo rcv, int flags);
+
+ void ReadBlob(already_AddRefed<DataChannelConnection> aThis, uint16_t aStream,
+ nsIInputStream* aBlob);
+
+ bool SendDeferredMessages() MOZ_REQUIRES(mLock);
+
+#ifdef SCTP_DTLS_SUPPORTED
+ int SctpDtlsOutput(void* addr, void* buffer, size_t length, uint8_t tos,
+ uint8_t set_df);
+#endif
+
+ bool InShutdown() const {
+#ifdef MOZ_DIAGNOSTIC_ASSERT_ENABLED
+ return mShutdown;
+#else
+ return false;
+#endif
+ }
+
+ private:
+ class Channels {
+ public:
+ using ChannelArray = AutoTArray<RefPtr<DataChannel>, 16>;
+
+ Channels() : mMutex("DataChannelConnection::Channels::mMutex") {}
+ Channels(const Channels&) = delete;
+ Channels& operator=(const Channels&) = delete;
+
+ void Insert(const RefPtr<DataChannel>& aChannel);
+ bool Remove(const RefPtr<DataChannel>& aChannel);
+ RefPtr<DataChannel> Get(uint16_t aId) const;
+ ChannelArray GetAll() const {
+ MutexAutoLock lock(mMutex);
+ return mChannels.Clone();
+ }
+ RefPtr<DataChannel> GetNextChannel(uint16_t aCurrentId) const;
+
+ private:
+ struct IdComparator {
+ bool Equals(const RefPtr<DataChannel>& aChannel, uint16_t aId) const;
+ bool LessThan(const RefPtr<DataChannel>& aChannel, uint16_t aId) const;
+ bool Equals(const RefPtr<DataChannel>& a1,
+ const RefPtr<DataChannel>& a2) const;
+ bool LessThan(const RefPtr<DataChannel>& a1,
+ const RefPtr<DataChannel>& a2) const;
+ };
+ mutable Mutex mMutex;
+ ChannelArray mChannels MOZ_GUARDED_BY(mMutex);
+ };
+
+ DataChannelConnection(DataConnectionListener* aListener,
+ nsISerialEventTarget* aTarget,
+ MediaTransportHandler* aHandler);
+
+ bool Init(const uint16_t aLocalPort, const uint16_t aNumStreams,
+ const Maybe<uint64_t>& aMaxMessageSize);
+
+ DataChannelConnectionState GetState() const MOZ_REQUIRES(mLock) {
+ mLock.AssertCurrentThreadOwns();
+
+ return mState;
+ }
+
+ void SetState(DataChannelConnectionState aState) MOZ_REQUIRES(mLock);
+ static int OnThresholdEvent(struct socket* sock, uint32_t sb_free,
+ void* ulp_info);
+
+#ifdef SCTP_DTLS_SUPPORTED
+ static void DTLSConnectThread(void* data);
+ void SendPacket(std::unique_ptr<MediaPacket>&& packet);
+ void SctpDtlsInput(const std::string& aTransportId,
+ const MediaPacket& packet);
+#endif
+ DataChannel* FindChannelByStream(uint16_t stream) MOZ_REQUIRES(mLock);
+ uint16_t FindFreeStream() const MOZ_REQUIRES(mLock);
+ bool RequestMoreStreams(int32_t aNeeded = 16) MOZ_REQUIRES(mLock);
+ uint32_t UpdateCurrentStreamIndex() MOZ_REQUIRES(mLock);
+ uint32_t GetCurrentStreamIndex() MOZ_REQUIRES(mLock);
+ int SendControlMessage(const uint8_t* data, uint32_t len, uint16_t stream)
+ MOZ_REQUIRES(mLock);
+ int SendOpenAckMessage(uint16_t stream) MOZ_REQUIRES(mLock);
+ int SendOpenRequestMessage(const nsACString& label,
+ const nsACString& protocol, uint16_t stream,
+ bool unordered,
+ DataChannelReliabilityPolicy prPolicy,
+ uint32_t prValue) MOZ_REQUIRES(mLock);
+ bool SendBufferedMessages(nsTArray<UniquePtr<BufferedOutgoingMsg>>& buffer,
+ size_t* aWritten);
+ int SendMsgInternal(OutgoingMsg& msg, size_t* aWritten);
+ int SendMsgInternalOrBuffer(nsTArray<UniquePtr<BufferedOutgoingMsg>>& buffer,
+ OutgoingMsg& msg, bool& buffered,
+ size_t* aWritten) MOZ_REQUIRES(mLock);
+ int SendDataMsgInternalOrBuffer(DataChannel& channel, const uint8_t* data,
+ size_t len, uint32_t ppid)
+ MOZ_REQUIRES(mLock);
+ int SendDataMsg(DataChannel& channel, const uint8_t* data, size_t len,
+ uint32_t ppidPartial, uint32_t ppidFinal) MOZ_REQUIRES(mLock);
+ int SendDataMsgCommon(uint16_t stream, const nsACString& aMsg, bool isBinary);
+
+ void DeliverQueuedData(uint16_t stream) MOZ_REQUIRES(mLock);
+
+ already_AddRefed<DataChannel> OpenFinish(
+ already_AddRefed<DataChannel>&& aChannel) MOZ_REQUIRES(mLock);
+
+ void ProcessQueuedOpens() MOZ_REQUIRES(mLock);
+ void ClearResets() MOZ_REQUIRES(mLock);
+ void SendOutgoingStreamReset() MOZ_REQUIRES(mLock);
+ void ResetOutgoingStream(uint16_t stream) MOZ_REQUIRES(mLock);
+ void HandleOpenRequestMessage(
+ const struct rtcweb_datachannel_open_request* req, uint32_t length,
+ uint16_t stream) MOZ_REQUIRES(mLock);
+ void HandleOpenAckMessage(const struct rtcweb_datachannel_ack* ack,
+ uint32_t length, uint16_t stream);
+ void HandleUnknownMessage(uint32_t ppid, uint32_t length, uint16_t stream)
+ MOZ_REQUIRES(mLock);
+ uint8_t BufferMessage(nsACString& recvBuffer, const void* data,
+ uint32_t length, uint32_t ppid, int flags);
+ void HandleDataMessage(const void* data, size_t length, uint32_t ppid,
+ uint16_t stream, int flags) MOZ_REQUIRES(mLock);
+ void HandleDCEPMessage(const void* buffer, size_t length, uint32_t ppid,
+ uint16_t stream, int flags) MOZ_REQUIRES(mLock);
+ void HandleMessage(const void* buffer, size_t length, uint32_t ppid,
+ uint16_t stream, int flags) MOZ_REQUIRES(mLock);
+ void HandleAssociationChangeEvent(const struct sctp_assoc_change* sac)
+ MOZ_REQUIRES(mLock);
+ void HandlePeerAddressChangeEvent(const struct sctp_paddr_change* spc)
+ MOZ_REQUIRES(mLock);
+ void HandleRemoteErrorEvent(const struct sctp_remote_error* sre)
+ MOZ_REQUIRES(mLock);
+ void HandleShutdownEvent(const struct sctp_shutdown_event* sse)
+ MOZ_REQUIRES(mLock);
+ void HandleAdaptationIndication(const struct sctp_adaptation_event* sai)
+ MOZ_REQUIRES(mLock);
+ void HandlePartialDeliveryEvent(const struct sctp_pdapi_event* spde)
+ MOZ_REQUIRES(mLock);
+ void HandleSendFailedEvent(const struct sctp_send_failed_event* ssfe)
+ MOZ_REQUIRES(mLock);
+ void HandleStreamResetEvent(const struct sctp_stream_reset_event* strrst)
+ MOZ_REQUIRES(mLock);
+ void HandleStreamChangeEvent(const struct sctp_stream_change_event* strchg)
+ MOZ_REQUIRES(mLock);
+ void HandleNotification(const union sctp_notification* notif, size_t n)
+ MOZ_REQUIRES(mLock);
+
+#ifdef SCTP_DTLS_SUPPORTED
+ bool IsSTSThread() const {
+ bool on = false;
+ if (mSTS) {
+ mSTS->IsOnCurrentThread(&on);
+ }
+ return on;
+ }
+#endif
+
+ mutable Mutex mLock;
+ // Avoid cycles with PeerConnectionImpl
+ // Use from main thread only as WeakPtr is not threadsafe
+ WeakPtr<DataConnectionListener> mListener;
+ bool mSendInterleaved MOZ_GUARDED_BY(mLock) = false;
+ // MainThread only
+ bool mMaxMessageSizeSet = false;
+ // mMaxMessageSize is only set on MainThread, but read off-main-thread
+ uint64_t mMaxMessageSize MOZ_GUARDED_BY(mLock) = 0;
+ // Main thread only
+ Maybe<bool> mAllocateEven;
+ // Data:
+ // NOTE: while this container will auto-expand, increases in the number of
+ // channels available from the stack must be negotiated!
+ // Accessed from both main and sts, API is threadsafe
+ Channels mChannels;
+ // STS only
+ uint32_t mCurrentStream = 0;
+ nsRefPtrDeque<DataChannel> mPending;
+ // STS and main
+ size_t mNegotiatedIdLimit MOZ_GUARDED_BY(mLock) = 0;
+ PendingType mPendingType MOZ_GUARDED_BY(mLock) = PendingType::None;
+ // holds data that's come in before a channel is open
+ nsTArray<UniquePtr<QueuedDataMessage>> mQueuedData MOZ_GUARDED_BY(mLock);
+ // holds outgoing control messages
+ nsTArray<UniquePtr<BufferedOutgoingMsg>> mBufferedControl
+ MOZ_GUARDED_BY(mLock);
+
+ // Streams pending reset. Accessed from main and STS.
+ AutoTArray<uint16_t, 4> mStreamsResetting MOZ_GUARDED_BY(mLock);
+ // accessed from STS thread
+ struct socket* mMasterSocket = nullptr;
+ // cloned from mMasterSocket on successful Connect on STS thread
+ struct socket* mSocket = nullptr;
+ DataChannelConnectionState mState MOZ_GUARDED_BY(mLock) =
+ DataChannelConnectionState::Closed;
+
+#ifdef SCTP_DTLS_SUPPORTED
+ std::string mTransportId;
+ bool mConnectedToTransportHandler = false;
+ RefPtr<MediaTransportHandler> mTransportHandler;
+ nsCOMPtr<nsIEventTarget> mSTS;
+#endif
+ uint16_t mLocalPort = 0; // Accessed from connect thread
+ uint16_t mRemotePort = 0;
+
+ nsCOMPtr<nsIThread> mInternalIOThread = nullptr;
+ nsCString mRecvBuffer;
+
+ // Workaround to prevent a message from being received on main before the
+ // sender sees the decrease in bufferedAmount.
+ bool mDeferSend = false;
+ std::vector<std::unique_ptr<MediaPacket>> mDeferredSend;
+
+#ifdef MOZ_DIAGNOSTIC_ASSERT_ENABLED
+ bool mShutdown;
+#endif
+ uintptr_t mId = 0;
+};
+
+class DataChannel {
+ friend class DataChannelOnMessageAvailable;
+ friend class DataChannelConnection;
+
+ public:
+ struct TrafficCounters {
+ uint32_t mMessagesSent = 0;
+ uint64_t mBytesSent = 0;
+ uint32_t mMessagesReceived = 0;
+ uint64_t mBytesReceived = 0;
+ };
+
+ DataChannel(DataChannelConnection* connection, uint16_t stream,
+ DataChannelState state, const nsACString& label,
+ const nsACString& protocol, DataChannelReliabilityPolicy policy,
+ uint32_t value, bool ordered, bool negotiated,
+ DataChannelListener* aListener, nsISupports* aContext)
+ : mListener(aListener),
+ mContext(aContext),
+ mConnection(connection),
+ mLabel(label),
+ mProtocol(protocol),
+ mReadyState(state),
+ mStream(stream),
+ mPrPolicy(policy),
+ mPrValue(value),
+ mNegotiated(negotiated),
+ mOrdered(ordered),
+ mIsRecvBinary(false),
+ mBufferedThreshold(0), // default from spec
+ mBufferedAmount(0),
+ mMainThreadEventTarget(connection->GetNeckoTarget()),
+ mStatsLock("netwer::sctp::DataChannel::mStatsLock") {
+ NS_ASSERTION(mConnection, "NULL connection");
+ }
+ DataChannel(const DataChannel&) = delete;
+ DataChannel& operator=(const DataChannel&) = delete;
+
+ private:
+ ~DataChannel();
+
+ public:
+ NS_INLINE_DECL_THREADSAFE_REFCOUNTING(DataChannel)
+
+ // when we disconnect from the connection after stream RESET
+ void StreamClosedLocked();
+
+ // Complete dropping of the link between DataChannel and the connection.
+ // After this, except for a few methods below listed to be safe, you can't
+ // call into DataChannel.
+ void ReleaseConnection();
+
+ // Close this DataChannel. Can be called multiple times. MUST be called
+ // before destroying the DataChannel (state must be CLOSED or CLOSING).
+ void Close();
+
+ // Set the listener (especially for channels created from the other side)
+ void SetListener(DataChannelListener* aListener, nsISupports* aContext);
+
+ // Helper for send methods that converts POSIX error codes to an ErrorResult.
+ static void SendErrnoToErrorResult(int error, size_t aMessageSize,
+ ErrorResult& aRv);
+
+ // Send a string
+ void SendMsg(const nsACString& aMsg, ErrorResult& aRv);
+
+ // Send a binary message (TypedArray)
+ void SendBinaryMsg(const nsACString& aMsg, ErrorResult& aRv);
+
+ // Send a binary blob
+ void SendBinaryBlob(dom::Blob& aBlob, ErrorResult& aRv);
+
+ DataChannelReliabilityPolicy GetType() const { return mPrPolicy; }
+
+ dom::Nullable<uint16_t> GetMaxPacketLifeTime() const;
+
+ dom::Nullable<uint16_t> GetMaxRetransmits() const;
+
+ bool GetNegotiated() const { return mNegotiated; }
+
+ bool GetOrdered() const { return mOrdered; }
+
+ void IncrementBufferedAmount(uint32_t aSize, ErrorResult& aRv);
+ void DecrementBufferedAmount(uint32_t aSize);
+
+ // Amount of data buffered to send
+ uint32_t GetBufferedAmount() const {
+ MOZ_ASSERT(NS_IsMainThread());
+ return mBufferedAmount;
+ }
+
+ // Trigger amount for generating BufferedAmountLow events
+ uint32_t GetBufferedAmountLowThreshold() const;
+ void SetBufferedAmountLowThreshold(uint32_t aThreshold);
+
+ void AnnounceOpen();
+ // TODO(bug 843625): Optionally pass an error here.
+ void AnnounceClosed();
+
+ // Find out state
+ DataChannelState GetReadyState() const {
+ MOZ_ASSERT(NS_IsMainThread());
+ return mReadyState;
+ }
+
+ // Set ready state
+ void SetReadyState(DataChannelState aState);
+
+ void GetLabel(nsAString& aLabel) { CopyUTF8toUTF16(mLabel, aLabel); }
+ void GetProtocol(nsAString& aProtocol) {
+ CopyUTF8toUTF16(mProtocol, aProtocol);
+ }
+ uint16_t GetStream() const { return mStream; }
+
+ void SendOrQueue(DataChannelOnMessageAvailable* aMessage);
+
+ TrafficCounters GetTrafficCounters() const;
+
+ private:
+ nsresult AddDataToBinaryMsg(const char* data, uint32_t size);
+ bool EnsureValidStream(ErrorResult& aRv);
+ void WithTrafficCounters(const std::function<void(TrafficCounters&)>&);
+
+ // These are both mainthread only
+ DataChannelListener* mListener;
+ nsCOMPtr<nsISupports> mContext;
+
+ RefPtr<DataChannelConnection> mConnection;
+ // mainthread only
+ bool mEverOpened = false;
+ const nsCString mLabel;
+ const nsCString mProtocol;
+ // This is mainthread only
+ DataChannelState mReadyState;
+ uint16_t mStream;
+ const DataChannelReliabilityPolicy mPrPolicy;
+ const uint32_t mPrValue;
+ // Accessed on main and STS
+ const bool mNegotiated;
+ const bool mOrdered;
+ // The data channel has completed the open procedure and the client has been
+ // notified about it.
+ bool mHasFinishedOpen = false;
+ // The channel has been opened, but the peer has not yet acked - ensures that
+ // the messages are sent ordered until this is cleared.
+ bool mWaitingForAck = false;
+ // A too large message was attempted to be sent - closing data channel.
+ bool mClosingTooLarge = false;
+ bool mIsRecvBinary;
+ size_t mBufferedThreshold;
+ // Read/written on main only. Decremented via message-passing, because the
+ // spec requires us to queue a task for this.
+ size_t mBufferedAmount;
+ nsCString mRecvBuffer;
+ nsTArray<UniquePtr<BufferedOutgoingMsg>>
+ mBufferedData; // MOZ_GUARDED_BY(mConnection->mLock)
+ nsCOMPtr<nsISerialEventTarget> mMainThreadEventTarget;
+ mutable Mutex mStatsLock;
+ TrafficCounters mTrafficCounters MOZ_GUARDED_BY(mStatsLock);
+};
+
+// used to dispatch notifications of incoming data to the main thread
+// Patterned on CallOnMessageAvailable in WebSockets
+// Also used to proxy other items to MainThread
+class DataChannelOnMessageAvailable : public Runnable {
+ public:
+ enum class EventType {
+ OnConnection,
+ OnDisconnected,
+ OnChannelCreated,
+ OnDataString,
+ OnDataBinary,
+ };
+
+ DataChannelOnMessageAvailable(
+ EventType aType, DataChannelConnection* aConnection,
+ DataChannel* aChannel,
+ nsCString& aData) // XXX this causes inefficiency
+ : Runnable("DataChannelOnMessageAvailable"),
+ mType(aType),
+ mChannel(aChannel),
+ mConnection(aConnection),
+ mData(aData) {}
+
+ DataChannelOnMessageAvailable(EventType aType, DataChannel* aChannel)
+ : Runnable("DataChannelOnMessageAvailable"),
+ mType(aType),
+ mChannel(aChannel) {}
+ // XXX is it safe to leave mData uninitialized? This should only be
+ // used for notifications that don't use them, but I'd like more
+ // bulletproof compile-time checking.
+
+ DataChannelOnMessageAvailable(EventType aType,
+ DataChannelConnection* aConnection,
+ DataChannel* aChannel)
+ : Runnable("DataChannelOnMessageAvailable"),
+ mType(aType),
+ mChannel(aChannel),
+ mConnection(aConnection) {}
+
+ // for ON_CONNECTION/ON_DISCONNECTED
+ DataChannelOnMessageAvailable(EventType aType,
+ DataChannelConnection* aConnection)
+ : Runnable("DataChannelOnMessageAvailable"),
+ mType(aType),
+ mConnection(aConnection) {}
+ DataChannelOnMessageAvailable(const DataChannelOnMessageAvailable&) = delete;
+ DataChannelOnMessageAvailable& operator=(
+ const DataChannelOnMessageAvailable&) = delete;
+
+ NS_IMETHOD Run() override;
+
+ private:
+ ~DataChannelOnMessageAvailable() = default;
+
+ EventType mType;
+ // XXX should use union
+ RefPtr<DataChannel> mChannel;
+ RefPtr<DataChannelConnection> mConnection;
+ nsCString mData;
+};
+
+} // namespace mozilla
+
+#endif // NETWERK_SCTP_DATACHANNEL_DATACHANNEL_H_