/* -*- 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 <string>

#include <errno.h>

#include "nsISupports.h"

#include "nsCOMPtr.h"

#include "mozilla/WeakPtr.h"

#include "nsString.h"

#include "nsThreadUtils.h"

#include "nsTArray.h"

#include "nsDeque.h"

#include "nsIInputStream.h"

#include "mozilla/Mutex.h"

#include "DataChannelProtocol.h"

#include "DataChannelListener.h"

#include "mozilla/net/NeckoTargetHolder.h"

#ifdef SCTP_DTLS_SUPPORTED

#include "mtransport/sigslot.h"

#include "mtransport/transportflow.h"

#include "mtransport/transportlayer.h"

#include "mtransport/transportlayerdtls.h"

#endif

#ifndef DATACHANNEL_LOG

#define DATACHANNEL_LOG(args)

#endif

#ifndef EALREADY

#define EALREADY  WSAEALREADY

#endif

extern "C" {

  struct socket;

  struct sctp_rcvinfo;

}

namespace mozilla {

class DataChannelConnection;

class DataChannel;

class DataChannelOnMessageAvailable;

// 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() = default;;

  void Advance(size_t offset);

  struct sctp_sendv_spa &GetInfo() { return *mInfo; };

  size_t GetLength() { return mLength; };

  size_t GetLeft() { return mLength - mPos; };

  const uint8_t *GetData() { 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 &message);

  ~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()

  {

    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

{

  virtual ~DataChannelConnection();

public:

  enum {

      PENDING_NONE = 0U, // No outgoing messages are pending

      PENDING_DCEP = 1U, // Outgoing DCEP messages are pending

      PENDING_DATA = 2U, // Outgoing data channel messages are pending

  };

  NS_INLINE_DECL_THREADSAFE_REFCOUNTING(DataChannelConnection)

  class DataConnectionListener : public SupportsWeakPtr<DataConnectionListener>

  {

  public:

    MOZ_DECLARE_WEAKREFERENCE_TYPENAME(DataChannelConnection::DataConnectionListener)

    virtual ~DataConnectionListener() = default;

    // Called when a new DataChannel has been opened by the other side.

    virtual void NotifyDataChannel(already_AddRefed<DataChannel> channel) = 0;

  };

  DataChannelConnection(DataConnectionListener *listener,

                        nsIEventTarget *aTarget);

  bool Init(unsigned short aPort, uint16_t aNumStreams, bool aMaxMessageSizeSet,

            uint64_t aMaxMessageSize);

  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();

#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

  // Connect using a TransportFlow (DTLS) channel

  void SetEvenOdd();

  bool ConnectViaTransportFlow(TransportFlow *aFlow, uint16_t localport, uint16_t remoteport);

  void CompleteConnect(TransportLayer *layer, TransportLayer::State state);

  void SetSignals();

#endif

  typedef enum {

    RELIABLE=0,

    PARTIAL_RELIABLE_REXMIT = 1,

    PARTIAL_RELIABLE_TIMED = 2

  } Type;

  MOZ_MUST_USE

  already_AddRefed<DataChannel> Open(const nsACString& label,

                                     const nsACString& protocol,

                                     Type type, bool inOrder,

                                     uint32_t prValue,

                                     DataChannelListener *aListener,

                                     nsISupports *aContext,

                                     bool aExternalNegotiated,

                                     uint16_t aStream);

  void Stop();

  void Close(DataChannel *aChannel);

  // CloseInt() must be called with mLock held

  void CloseInt(DataChannel *aChannel);

  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

  int ReceiveCallback(struct socket* sock, void *data, size_t datalen,

                      struct sctp_rcvinfo rcv, int flags);

  // Find out state

  enum {

    CONNECTING = 0U,

    OPEN = 1U,

    CLOSING = 2U,

    CLOSED = 3U

  };

  uint16_t GetReadyState() { MutexAutoLock lock(mLock); return mState; }

  friend class DataChannel;

  Mutex  mLock;

  void ReadBlob(already_AddRefed<DataChannelConnection> aThis, uint16_t aStream, nsIInputStream* aBlob);

  void GetStreamIds(std::vector<uint16_t>* aStreamList);

  bool SendDeferredMessages();

protected:

  friend class DataChannelOnMessageAvailable;

  // Avoid cycles with PeerConnectionImpl

  // Use from main thread only as WeakPtr is not threadsafe

  WeakPtr<DataConnectionListener> mListener;

private:

  friend class DataChannelConnectRunnable;

#ifdef SCTP_DTLS_SUPPORTED

  static void DTLSConnectThread(void *data);

  int SendPacket(nsAutoPtr<MediaPacket> packet);

  void SctpDtlsInput(TransportLayer *layer, MediaPacket& packet);

  static int SctpDtlsOutput(void *addr, void *buffer, size_t length, uint8_t tos, uint8_t set_df);

#endif

  DataChannel* FindChannelByStream(uint16_t stream);

  uint16_t FindFreeStream();

  bool RequestMoreStreams(int32_t aNeeded = 16);

  uint32_t UpdateCurrentStreamIndex();

  uint32_t GetCurrentStreamIndex();

  int SendControlMessage(const uint8_t *data, uint32_t len, uint16_t stream);

  int SendOpenAckMessage(uint16_t stream);

  int SendOpenRequestMessage(const nsACString& label, const nsACString& protocol, uint16_t stream,

                             bool unordered, uint16_t prPolicy, uint32_t prValue);

  bool SendBufferedMessages(nsTArray<nsAutoPtr<BufferedOutgoingMsg>> &buffer);

  int SendMsgInternal(OutgoingMsg &msg);

  int SendMsgInternalOrBuffer(nsTArray<nsAutoPtr<BufferedOutgoingMsg>> &buffer, OutgoingMsg &msg,

                              bool &buffered);

  int SendDataMsgInternalOrBuffer(DataChannel &channel, const uint8_t *data, size_t len,

                                  uint32_t ppid);

  int SendDataMsg(DataChannel &channel, const uint8_t *data, size_t len, uint32_t ppidPartial,

                  uint32_t ppidFinal);

  int SendDataMsgCommon(uint16_t stream, const nsACString &aMsg, bool isBinary);

  void DeliverQueuedData(uint16_t stream);

  already_AddRefed<DataChannel> OpenFinish(already_AddRefed<DataChannel>&& aChannel);

  void ProcessQueuedOpens();

  void ClearResets();

  void SendOutgoingStreamReset();

  void ResetOutgoingStream(uint16_t stream);

  void HandleOpenRequestMessage(const struct rtcweb_datachannel_open_request *req,

                                uint32_t length, uint16_t stream);

  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);

  uint8_t BufferMessage(nsACString& recvBuffer, const void *data, uint32_t length, uint32_t ppid,

                        int flags);

  void HandleDataMessage(const void *buffer, size_t length, uint32_t ppid, uint16_t stream,

                         int flags);

  void HandleDCEPMessage(const void *buffer, size_t length, uint32_t ppid, uint16_t stream,

                         int flags);

  void HandleMessage(const void *buffer, size_t length, uint32_t ppid, uint16_t stream, int flags);

  void HandleAssociationChangeEvent(const struct sctp_assoc_change *sac);

  void HandlePeerAddressChangeEvent(const struct sctp_paddr_change *spc);

  void HandleRemoteErrorEvent(const struct sctp_remote_error *sre);

  void HandleShutdownEvent(const struct sctp_shutdown_event *sse);

  void HandleAdaptationIndication(const struct sctp_adaptation_event *sai);

  void HandlePartialDeliveryEvent(const struct sctp_pdapi_event *spde);

  void HandleSendFailedEvent(const struct sctp_send_failed_event *ssfe);

  void HandleStreamResetEvent(const struct sctp_stream_reset_event *strrst);

  void HandleStreamChangeEvent(const struct sctp_stream_change_event *strchg);

  void HandleNotification(const union sctp_notification *notif, size_t n);

#ifdef SCTP_DTLS_SUPPORTED

  bool IsSTSThread() {

    bool on = false;

    if (mSTS) {

      mSTS->IsOnCurrentThread(&on);

    }

    return on;

  }

#endif

  // Exists solely for proxying release of the TransportFlow to the STS thread

  static void ReleaseTransportFlow(const RefPtr<TransportFlow>& aFlow) {}

  bool mSendInterleaved;

  bool mPpidFragmentation;

  bool mMaxMessageSizeSet;

  uint64_t mMaxMessageSize;

  // Data:

  // NOTE: while this array will auto-expand, increases in the number of

  // channels available from the stack must be negotiated!

  bool mAllocateEven;

  AutoTArray<RefPtr<DataChannel>,16> mStreams;

  uint32_t mCurrentStream;

  nsDeque mPending; // Holds addref'ed DataChannel's -- careful!

  // holds data that's come in before a channel is open

  nsTArray<nsAutoPtr<QueuedDataMessage>> mQueuedData;

  // holds outgoing control messages

  nsTArray<nsAutoPtr<BufferedOutgoingMsg>> mBufferedControl; // GUARDED_BY(mConnection->mLock)

  // Streams pending reset

  AutoTArray<uint16_t,4> mStreamsResetting;

  struct socket *mMasterSocket; // accessed from STS thread

  struct socket *mSocket; // cloned from mMasterSocket on successful Connect on STS thread

  uint16_t mState; // Protected with mLock

#ifdef SCTP_DTLS_SUPPORTED

  RefPtr<TransportFlow> mTransportFlow;

  TransportLayerDtls* mDtls;

  nsCOMPtr<nsIEventTarget> mSTS;

#endif

  uint16_t mLocalPort; // Accessed from connect thread

  uint16_t mRemotePort;

  nsCOMPtr<nsIThread> mInternalIOThread;

  uint8_t mPendingType;

  nsCString mRecvBuffer;

#ifdef MOZ_DIAGNOSTIC_ASSERT_ENABLED

  bool mShutdown;

#endif

};

#define ENSURE_DATACONNECTION \

  do { MOZ_ASSERT(mConnection); if (!mConnection) { return; } } while (0)

class DataChannel {

public:

  enum {

    CONNECTING = 0U,

    OPEN = 1U,

    CLOSING = 2U,

    CLOSED = 3U,

    WAITING_TO_OPEN = 4U

  };

  DataChannel(DataChannelConnection *connection,

              uint16_t stream,

              uint16_t state,

              const nsACString& label,

              const nsACString& protocol,

              uint16_t policy, uint32_t value,

              uint32_t flags,

              DataChannelListener *aListener,

              nsISupports *aContext)

    : mListenerLock("netwerk::sctp::DataChannel")

    , mListener(aListener)

    , mContext(aContext)

    , mConnection(connection)

    , mLabel(label)

    , mProtocol(protocol)

    , mState(state)

    , mStream(stream)

    , mPrPolicy(policy)

    , mPrValue(value)

    , mFlags(flags)

    , mId(0)

    , mIsRecvBinary(false)

    , mBufferedThreshold(0) // default from spec

    , mMainThreadEventTarget(connection->GetNeckoTarget())

    {

      NS_ASSERTION(mConnection,"NULL connection");

    }

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, 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 SendBinaryStream(nsIInputStream *aBlob, ErrorResult& aRv);

  uint16_t GetType() { return mPrPolicy; }

  dom::Nullable<uint16_t> GetMaxPacketLifeTime() const;

  dom::Nullable<uint16_t> GetMaxRetransmits() const;

  bool GetOrdered() { return !(mFlags & DATA_CHANNEL_FLAGS_OUT_OF_ORDER_ALLOWED); }

  // Amount of data buffered to send

  uint32_t GetBufferedAmount()

  {

    if (!mConnection) {

      return 0;

    }

    MutexAutoLock lock(mConnection->mLock);

    size_t buffered = GetBufferedAmountLocked();

#if (SIZE_MAX > UINT32_MAX)

    if (buffered > UINT32_MAX) { // paranoia - >4GB buffered is very very unlikely

      buffered = UINT32_MAX;

    }

#endif

    return buffered;

  }

  // Trigger amount for generating BufferedAmountLow events

  uint32_t GetBufferedAmountLowThreshold();

  void SetBufferedAmountLowThreshold(uint32_t aThreshold);

  // Find out state

  uint16_t GetReadyState()

    {

      if (mConnection) {

        MutexAutoLock lock(mConnection->mLock);

        if (mState == WAITING_TO_OPEN)

          return CONNECTING;

        return mState;

      }

      return CLOSED;

    }

  void GetLabel(nsAString& aLabel) { CopyUTF8toUTF16(mLabel, aLabel); }

  void GetProtocol(nsAString& aProtocol) { CopyUTF8toUTF16(mProtocol, aProtocol); }

  uint16_t GetStream() { return mStream; }

  void AppReady();

  void SendOrQueue(DataChannelOnMessageAvailable *aMessage);

protected:

  Mutex mListenerLock; // protects mListener and mContext

  DataChannelListener *mListener;

  nsCOMPtr<nsISupports> mContext;

private:

  friend class DataChannelOnMessageAvailable;

  friend class DataChannelConnection;

  nsresult AddDataToBinaryMsg(const char *data, uint32_t size);

  size_t GetBufferedAmountLocked() const;

  bool EnsureValidStream(ErrorResult& aRv);

  RefPtr<DataChannelConnection> mConnection;

  nsCString mLabel;

  nsCString mProtocol;

  uint16_t mState;

  uint16_t mStream;

  uint16_t mPrPolicy;

  uint32_t mPrValue;

  uint32_t mFlags;

  uint32_t mId;

  bool mIsRecvBinary;

  size_t mBufferedThreshold;

  nsCString mRecvBuffer;

  nsTArray<nsAutoPtr<BufferedOutgoingMsg>> mBufferedData; // GUARDED_BY(mConnection->mLock)

  nsTArray<nsCOMPtr<nsIRunnable>> mQueuedMessages;

  nsCOMPtr<nsIEventTarget> mMainThreadEventTarget;

};

// 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 {

    ON_CONNECTION,

    ON_DISCONNECTED,

    ON_CHANNEL_CREATED,

    ON_CHANNEL_OPEN,

    ON_CHANNEL_CLOSED,

    ON_DATA_STRING,

    ON_DATA_BINARY,

    BUFFER_LOW_THRESHOLD,

    NO_LONGER_BUFFERED,

  };  /* types */

  DataChannelOnMessageAvailable(

    int32_t aType,

    DataChannelConnection* aConnection,

    DataChannel* aChannel,

    nsCString& aData) // XXX this causes inefficiency

    : Runnable("DataChannelOnMessageAvailable")

    , mType(aType)

    , mChannel(aChannel)

    , mConnection(aConnection)

    , mData(aData)

  {

  }

  DataChannelOnMessageAvailable(int32_t 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(int32_t aType,

                                DataChannelConnection* aConnection,

                                DataChannel* aChannel)

    : Runnable("DataChannelOnMessageAvailable")

    , mType(aType)

    , mChannel(aChannel)

    , mConnection(aConnection)

  {

  }

  // for ON_CONNECTION/ON_DISCONNECTED

  DataChannelOnMessageAvailable(int32_t aType,

                                DataChannelConnection* aConnection)

    : Runnable("DataChannelOnMessageAvailable")

    , mType(aType)

    , mConnection(aConnection)

  {

  }

  NS_IMETHOD Run() override

  {

    MOZ_ASSERT(NS_IsMainThread());

    // Note: calling the listeners can indirectly cause the listeners to be

    // made available for GC (by removing event listeners), especially for

    // OnChannelClosed().  We hold a ref to the Channel and the listener

    // while calling this.

    switch (mType) {

      case ON_DATA_STRING:

      case ON_DATA_BINARY:

      case ON_CHANNEL_OPEN:

      case ON_CHANNEL_CLOSED:

      case BUFFER_LOW_THRESHOLD:

      case NO_LONGER_BUFFERED:

        {

          MutexAutoLock lock(mChannel->mListenerLock);

          if (!mChannel->mListener) {

            DATACHANNEL_LOG(("DataChannelOnMessageAvailable (%d) with null Listener!",mType));

            return NS_OK;

          }

          switch (mType) {

            case ON_DATA_STRING:

              mChannel->mListener->OnMessageAvailable(mChannel->mContext, mData);

              break;

            case ON_DATA_BINARY:

              mChannel->mListener->OnBinaryMessageAvailable(mChannel->mContext, mData);

              break;

            case ON_CHANNEL_OPEN:

              mChannel->mListener->OnChannelConnected(mChannel->mContext);

              break;

            case ON_CHANNEL_CLOSED:

              mChannel->mListener->OnChannelClosed(mChannel->mContext);

              break;

            case BUFFER_LOW_THRESHOLD:

              mChannel->mListener->OnBufferLow(mChannel->mContext);

              break;

            case NO_LONGER_BUFFERED:

              mChannel->mListener->NotBuffered(mChannel->mContext);

              break;

          }

          break;

        }

      case ON_DISCONNECTED:

        // If we've disconnected, make sure we close all the streams - from mainthread!

        mConnection->CloseAll();

        MOZ_FALLTHROUGH;

      case ON_CHANNEL_CREATED:

      case ON_CONNECTION:

        // WeakPtr - only used/modified/nulled from MainThread so we can use a WeakPtr here

        if (!mConnection->mListener) {

          DATACHANNEL_LOG(("DataChannelOnMessageAvailable (%d) with null Listener",mType));

          return NS_OK;

        }

        switch (mType) {

          case ON_CHANNEL_CREATED:

            // important to give it an already_AddRefed pointer!

            mConnection->mListener->NotifyDataChannel(mChannel.forget());

            break;

          default:

            break;

        }

        break;

    }

    return NS_OK;

  }

private:

  ~DataChannelOnMessageAvailable() = default;

  int32_t                         mType;

  // XXX should use union

  RefPtr<DataChannel>             mChannel;

  RefPtr<DataChannelConnection>   mConnection;

  nsCString                       mData;

};

}

#endif  // NETWERK_SCTP_DATACHANNEL_DATACHANNEL_H_