Added missing functions to C API

README.md

@@ -79,11 +79,11 @@ MY_ON_RECV_CANDIDATE_FROM_REMOTE([pc](string candidate, string mid) {
 ### Observe the PeerConnection state
 
 ```cpp
-pc->onStateChanged([](PeerConnection::State state) {
+pc->onStateChange([](PeerConnection::State state) {
     cout << "State: " << state << endl;
 });
 
-pc->onGatheringStateChanged([](PeerConnection::GatheringState state) {
+pc->onGatheringStateChange([](PeerConnection::GatheringState state) {
     cout << "Gathering state: " << state << endl;
 });
 

include/rtc/channel.hpp

@@ -31,12 +31,10 @@ class Channel {
 public:
 	virtual void close() = 0;
 	virtual bool send(const std::variant<binary, string> &data) = 0; // returns false if buffered
-	virtual std::optional<std::variant<binary, string>> receive() = 0; // only if onMessage unset
 
 	virtual bool isOpen() const = 0;
 	virtual bool isClosed() const = 0;
-
-	virtual size_t availableAmount() const; // total size available to receive
+	virtual size_t maxMessageSize() const; // max message size in a call to send
 	virtual size_t bufferedAmount() const; // total size buffered to send
 
 	void onOpen(std::function<void()> callback);
@@ -47,11 +45,14 @@ public:
 	void onMessage(std::function<void(const binary &data)> binaryCallback,
 	               std::function<void(const string &data)> stringCallback);
 
-	void onAvailable(std::function<void()> callback);
 	void onBufferedAmountLow(std::function<void()> callback);
-
 	void setBufferedAmountLowThreshold(size_t amount);
 
+	// Extended API
+	virtual std::optional<std::variant<binary, string>> receive() = 0; // only if onMessage unset
+	virtual size_t availableAmount() const; // total size available to receive
+	void onAvailable(std::function<void()> callback);
+
 protected:
 	virtual void triggerOpen();
 	virtual void triggerClosed();

include/rtc/datachannel.hpp

@@ -44,26 +44,24 @@ public:
 	            unsigned int stream);
 	~DataChannel();
 
-	void close(void) override;
+	unsigned int stream() const;
+	string label() const;
+	string protocol() const;
+	Reliability reliability() const;
 
+	void close(void) override;
 	bool send(const std::variant<binary, string> &data) override;
 	bool send(const byte *data, size_t size);
-
 	template <typename Buffer> bool sendBuffer(const Buffer &buf);
 	template <typename Iterator> bool sendBuffer(Iterator first, Iterator last);
 
-	std::optional<std::variant<binary, string>> receive() override;
-
 	bool isOpen(void) const override;
 	bool isClosed(void) const override;
-	size_t availableAmount() const override;
-
-	size_t maxMessageSize() const;  // maximum message size in a call to send or sendBuffer
+	size_t maxMessageSize() const override;
 
-	unsigned int stream() const;
-	string label() const;
-	string protocol() const;
-	Reliability reliability() const;
+	// Extended API
+	size_t availableAmount() const override;
+	std::optional<std::variant<binary, string>> receive() override;
 
 private:
 	void remoteClose();

include/rtc/rtc.h

@@ -33,13 +33,13 @@ typedef enum {
 	RTC_FAILED = 4,
 	RTC_CLOSED = 5,
 	RTC_DESTROYING = 6 // internal
-} rtc_state_t;
+} rtcState;
 
 typedef enum {
 	RTC_GATHERING_NEW = 0,
 	RTC_GATHERING_INPROGRESS = 1,
 	RTC_GATHERING_COMPLETE = 2
-} rtc_gathering_state_t;
+} rtcGatheringState;
 
 // Don't change, it must match plog severity
 typedef enum {
@@ -50,31 +50,56 @@ typedef enum {
 	RTC_LOG_INFO = 4,
 	RTC_LOG_DEBUG = 5,
 	RTC_LOG_VERBOSE = 6
-} rtc_log_level_t;
+} rtcLogLevel;
 
-void rtcInitLogger(rtc_log_level_t level);
+typedef void (*dataChannelCallbackFunc)(int dc, void *ptr);
+typedef void (*descriptionCallbackFunc)(const char *sdp, const char *type, void *ptr);
+typedef void (*candidateCallbackFunc)(const char *cand, const char *mid, void *ptr);
+typedef void (*stateChangeCallbackFunc)(rtcState state, void *ptr);
+typedef void (*gatheringStateCallbackFunc)(rtcGatheringState state, void *ptr);
+typedef void (*openCallbackFunc)(void *ptr);
+typedef void (*errorCallbackFunc)(const char *error, void *ptr);
+typedef void (*messageCallbackFunc)(const char *message, int size, void *ptr);
+typedef void (*bufferedAmountLowCallbackFunc)(void *ptr);
+typedef void (*availableCallbackFunc)(void *ptr);
 
+// Log
+void rtcInitLogger(rtcLogLevel level);
+
+// User pointer
+void rtcSetUserPointer(int i, void *ptr);
+
+// PeerConnection
 int rtcCreatePeerConnection(const char **iceServers, int iceServersCount);
-void rtcDeletePeerConnection(int pc);
+int rtcDeletePeerConnection(int pc);
+
+int rtcSetDataChannelCallback(int pc, dataChannelCallbackFunc cb);
+int rtcSetLocalDescriptionCallback(int pc, descriptionCallbackFunc cb);
+int rtcSetLocalCandidateCallback(int pc, candidateCallbackFunc cb);
+int rtcSetStateChangeCallback(int pc, stateChangeCallbackFunc cb);
+int rtcSetGatheringStateChangeCallback(int pc, gatheringStateCallbackFunc cb);
+
+int rtcSetRemoteDescription(int pc, const char *sdp, const char *type);
+int rtcAddRemoteCandidate(int pc, const char *cand, const char *mid);
+
+// DataChannel
 int rtcCreateDataChannel(int pc, const char *label);
-void rtcDeleteDataChannel(int dc);
-void rtcSetDataChannelCallback(int pc, void (*dataChannelCallback)(int, void *));
-void rtcSetLocalDescriptionCallback(int pc, void (*descriptionCallback)(const char *, const char *,
-                                                                        void *));
-void rtcSetLocalCandidateCallback(int pc,
-                                  void (*candidateCallback)(const char *, const char *, void *));
-void rtcSetStateChangeCallback(int pc, void (*stateCallback)(rtc_state_t state, void *));
-void rtcSetGatheringStateChangeCallback(int pc,
-                                        void (*gatheringStateCallback)(rtc_gathering_state_t state,
-                                                                       void *));
-void rtcSetRemoteDescription(int pc, const char *sdp, const char *type);
-void rtcAddRemoteCandidate(int pc, const char *candidate, const char *mid);
-int rtcGetDataChannelLabel(int dc, char *data, int size);
-void rtcSetOpenCallback(int dc, void (*openCallback)(void *));
-void rtcSetErrorCallback(int dc, void (*errorCallback)(const char *, void *));
-void rtcSetMessageCallback(int dc, void (*messageCallback)(const char *, int, void *));
+int rtcDeleteDataChannel(int dc);
+
+int rtcGetDataChannelLabel(int dc, char *buffer, int size);
+int rtcSetOpenCallback(int dc, openCallbackFunc cb);
+int rtcSetErrorCallback(int dc, errorCallbackFunc cb);
+int rtcSetMessageCallback(int dc, messageCallbackFunc cb);
 int rtcSendMessage(int dc, const char *data, int size);
-void rtcSetUserPointer(int i, void *ptr);
+
+int rtcGetBufferedAmount(int dc); // total size buffered to send
+int rtcSetBufferedAmountLowThreshold(int dc, int amount);
+int rtcSetBufferedAmountLowCallback(int dc, bufferedAmountLowCallbackFunc cb);
+
+// DataChannel extended API
+int rtcGetAvailableAmount(int dc); // total size available to receive
+int rtcSetAvailableCallback(int dc, availableCallbackFunc cb);
+int rtcReceiveMessage(int dc, char *buffer, int *size);
 
 #ifdef __cplusplus
 } // extern "C"

src/channel.cpp

@@ -22,6 +22,12 @@ namespace {}
 
 namespace rtc {
 
+size_t Channel::maxMessageSize() const { return DEFAULT_MAX_MESSAGE_SIZE; }
+
+size_t Channel::bufferedAmount() const { return mBufferedAmount; }
+
+size_t Channel::availableAmount() const { return 0; }
+
 void Channel::onOpen(std::function<void()> callback) {
 	mOpenCallback = callback;
 }
@@ -49,20 +55,16 @@ void Channel::onMessage(std::function<void(const binary &data)> binaryCallback,
 	});
 }
 
-void Channel::onAvailable(std::function<void()> callback) {
-	mAvailableCallback = callback;
-}
-
 void Channel::onBufferedAmountLow(std::function<void()> callback) {
 	mBufferedAmountLowCallback = callback;
 }
 
-size_t Channel::availableAmount() const { return 0; }
-
-size_t Channel::bufferedAmount() const { return mBufferedAmount; }
-
 void Channel::setBufferedAmountLowThreshold(size_t amount) { mBufferedAmountLowThreshold = amount; }
 
+void Channel::onAvailable(std::function<void()> callback) {
+	mAvailableCallback = callback;
+}
+
 void Channel::triggerOpen() { mOpenCallback(); }
 
 void Channel::triggerClosed() { mClosedCallback(); }

src/datachannel.cpp

@@ -83,6 +83,14 @@ DataChannel::~DataChannel() {
 	close();
 }
 
+unsigned int DataChannel::stream() const { return mStream; }
+
+string DataChannel::label() const { return mLabel; }
+
+string DataChannel::protocol() const { return mProtocol; }
+
+Reliability DataChannel::reliability() const { return *mReliability; }
+
 void DataChannel::close() {
 	if (mIsOpen.exchange(false) && mSctpTransport)
 		mSctpTransport->reset(mStream);
@@ -137,8 +145,6 @@ bool DataChannel::isOpen(void) const { return mIsOpen; }
 
 bool DataChannel::isClosed(void) const { return mIsClosed; }
 
-size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
-
 size_t DataChannel::maxMessageSize() const {
 	size_t max = DEFAULT_MAX_MESSAGE_SIZE;
 	if (auto description = mPeerConnection->remoteDescription())
@@ -148,13 +154,7 @@ size_t DataChannel::maxMessageSize() const {
 	return std::min(max, LOCAL_MAX_MESSAGE_SIZE);
 }
 
-unsigned int DataChannel::stream() const { return mStream; }
-
-string DataChannel::label() const { return mLabel; }
-
-string DataChannel::protocol() const { return mProtocol; }
-
-Reliability DataChannel::reliability() const { return *mReliability; }
+size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
 
 void DataChannel::open(shared_ptr<SctpTransport> sctpTransport) {
 	mSctpTransport = sctpTransport;

src/rtc.cpp

@@ -22,195 +22,319 @@
 
 #include <rtc.h>
 
+#include <exception>
+#include <mutex>
 #include <unordered_map>
-
-#include <plog/Appenders/ColorConsoleAppender.h>
+#include <utility>
 
 using namespace rtc;
 using std::shared_ptr;
 using std::string;
 
+#define CATCH(statement)                                                                           \
+	try {                                                                                          \
+		statement;                                                                                 \
+	} catch (const std::exception &e) {                                                            \
+		PLOG_ERROR << e.what();                                                                    \
+		return -1;                                                                                 \
+	}
+
 namespace {
 
 std::unordered_map<int, shared_ptr<PeerConnection>> peerConnectionMap;
 std::unordered_map<int, shared_ptr<DataChannel>> dataChannelMap;
 std::unordered_map<int, void *> userPointerMap;
+std::mutex mutex;
 int lastId = 0;
 
 void *getUserPointer(int id) {
+	std::lock_guard lock(mutex);
 	auto it = userPointerMap.find(id);
 	return it != userPointerMap.end() ? it->second : nullptr;
 }
 
+shared_ptr<PeerConnection> getPeerConnection(int id) {
+	std::lock_guard lock(mutex);
+	auto it = peerConnectionMap.find(id);
+	return it != peerConnectionMap.end() ? it->second : nullptr;
+}
+
+shared_ptr<DataChannel> getDataChannel(int id) {
+	std::lock_guard lock(mutex);
+	auto it = dataChannelMap.find(id);
+	return it != dataChannelMap.end() ? it->second : nullptr;
+}
+
+int emplacePeerConnection(shared_ptr<PeerConnection> ptr) {
+	std::lock_guard lock(mutex);
+	int pc = ++lastId;
+	peerConnectionMap.emplace(std::make_pair(pc, ptr));
+	return pc;
+}
+
+int emplaceDataChannel(shared_ptr<DataChannel> ptr) {
+	std::lock_guard lock(mutex);
+	int dc = ++lastId;
+	dataChannelMap.emplace(std::make_pair(dc, ptr));
+	return dc;
+}
+
+bool erasePeerConnection(int pc) {
+	std::lock_guard lock(mutex);
+	if (peerConnectionMap.erase(pc) == 0)
+		return false;
+	userPointerMap.erase(pc);
+	return true;
+}
+
+bool eraseDataChannel(int dc) {
+	std::lock_guard lock(mutex);
+	if (dataChannelMap.erase(dc) == 0)
+		return false;
+	userPointerMap.erase(dc);
+	return true;
+}
+
 } // namespace
 
-void rtcInitLogger(rtc_log_level_t level) { InitLogger(static_cast<LogLevel>(level)); }
+void rtcInitLogger(rtcLogLevel level) { InitLogger(static_cast<LogLevel>(level)); }
+
+void rtcSetUserPointer(int i, void *ptr) {
+	if (ptr)
+		userPointerMap.insert(std::make_pair(i, ptr));
+	else
+		userPointerMap.erase(i);
+}
 
 int rtcCreatePeerConnection(const char **iceServers, int iceServersCount) {
 	Configuration config;
 	for (int i = 0; i < iceServersCount; ++i) {
 		config.iceServers.emplace_back(IceServer(string(iceServers[i])));
 	}
-	int pc = ++lastId;
-	peerConnectionMap.emplace(std::make_pair(pc, std::make_shared<PeerConnection>(config)));
-	return pc;
+	return emplacePeerConnection(std::make_shared<PeerConnection>(config));
 }
 
-void rtcDeletePeerConnection(int pc) { peerConnectionMap.erase(pc); }
+int rtcDeletePeerConnection(int pc) { return erasePeerConnection(pc) ? 0 : -1; }
 
 int rtcCreateDataChannel(int pc, const char *label) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return 0;
-	auto dataChannel = it->second->createDataChannel(string(label));
-	int dc = ++lastId;
-	dataChannelMap.emplace(std::make_pair(dc, dataChannel));
-	return dc;
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	return emplaceDataChannel(peerConnection->createDataChannel(string(label)));
 }
 
-void rtcDeleteDataChannel(int dc) { dataChannelMap.erase(dc); }
+int rtcDeleteDataChannel(int dc) { return eraseDataChannel(dc) ? 0 : -1; }
 
-void rtcSetDataChannelCallback(int pc, void (*dataChannelCallback)(int, void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
+int rtcSetDataChannelCallback(int pc, dataChannelCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
 
-	it->second->onDataChannel([pc, dataChannelCallback](std::shared_ptr<DataChannel> dataChannel) {
-		int dc = ++lastId;
-		dataChannelMap.emplace(std::make_pair(dc, dataChannel));
-		dataChannelCallback(dc, getUserPointer(pc));
+	peerConnection->onDataChannel([pc, cb](std::shared_ptr<DataChannel> dataChannel) {
+		int dc = emplaceDataChannel(dataChannel);
+		cb(dc, getUserPointer(pc));
 	});
+	return 0;
 }
 
-void rtcSetLocalDescriptionCallback(int pc, void (*descriptionCallback)(const char *, const char *,
-                                                                        void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
+int rtcSetLocalDescriptionCallback(int pc, descriptionCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
 
-	it->second->onLocalDescription([pc, descriptionCallback](const Description &description) {
-		descriptionCallback(string(description).c_str(), description.typeString().c_str(),
-		                    getUserPointer(pc));
+	peerConnection->onLocalDescription([pc, cb](const Description &desc) {
+		cb(string(desc).c_str(), desc.typeString().c_str(), getUserPointer(pc));
 	});
+	return 0;
 }
 
-void rtcSetLocalCandidateCallback(int pc,
-                                  void (*candidateCallback)(const char *, const char *, void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
+int rtcSetLocalCandidateCallback(int pc, candidateCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
 
-	it->second->onLocalCandidate([pc, candidateCallback](const Candidate &candidate) {
-		candidateCallback(candidate.candidate().c_str(), candidate.mid().c_str(),
-		                  getUserPointer(pc));
+	peerConnection->onLocalCandidate([pc, cb](const Candidate &cand) {
+		cb(cand.candidate().c_str(), cand.mid().c_str(), getUserPointer(pc));
 	});
+	return 0;
 }
 
-void rtcSetStateChangeCallback(int pc, void (*stateCallback)(rtc_state_t state, void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
+int rtcSetStateChangeCallback(int pc, stateChangeCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
 
-	it->second->onStateChange([pc, stateCallback](PeerConnection::State state) {
-		stateCallback(static_cast<rtc_state_t>(state), getUserPointer(pc));
+	peerConnection->onStateChange([pc, cb](PeerConnection::State state) {
+		cb(static_cast<rtcState>(state), getUserPointer(pc));
 	});
+	return 0;
 }
 
-void rtcSetGatheringStateChangeCallback(int pc,
-                                        void (*gatheringStateCallback)(rtc_gathering_state_t state,
-                                                                       void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
+int rtcSetGatheringStateChangeCallback(int pc, gatheringStateCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
 
-	it->second->onGatheringStateChange(
-	    [pc, gatheringStateCallback](PeerConnection::GatheringState state) {
-		    gatheringStateCallback(static_cast<rtc_gathering_state_t>(state), getUserPointer(pc));
-	    });
+	peerConnection->onGatheringStateChange([pc, cb](PeerConnection::GatheringState state) {
+		cb(static_cast<rtcGatheringState>(state), getUserPointer(pc));
+	});
+	return 0;
 }
 
-void rtcSetRemoteDescription(int pc, const char *sdp, const char *type) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
+int rtcSetRemoteDescription(int pc, const char *sdp, const char *type) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
 
-	it->second->setRemoteDescription(Description(string(sdp), type ? string(type) : ""));
+	CATCH(peerConnection->setRemoteDescription({string(sdp), type ? string(type) : ""}));
+	return 0;
 }
 
-void rtcAddRemoteCandidate(int pc, const char *candidate, const char *mid) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
+int rtcAddRemoteCandidate(int pc, const char *cand, const char *mid) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
 
-	it->second->addRemoteCandidate(Candidate(string(candidate), mid ? string(mid) : ""));
+	CATCH(peerConnection->addRemoteCandidate({string(cand), mid ? string(mid) : ""}))
+	return 0;
 }
 
 int rtcGetDataChannelLabel(int dc, char *buffer, int size) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return 0;
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
 
 	if (!size)
 		return 0;
 
-	string label = it->second->label();
+	string label = dataChannel->label();
 	size = std::min(size_t(size - 1), label.size());
 	std::copy(label.data(), label.data() + size, buffer);
 	buffer[size] = '\0';
 	return size + 1;
 }
 
-void rtcSetOpenCallback(int dc, void (*openCallback)(void *)) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return;
+int rtcSetOpenCallback(int dc, openCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
 
-	it->second->onOpen([dc, openCallback]() { openCallback(getUserPointer(dc)); });
+	dataChannel->onOpen([dc, cb]() { cb(getUserPointer(dc)); });
+	return 0;
 }
 
-void rtcSetErrorCallback(int dc, void (*errorCallback)(const char *, void *)) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return;
+int rtcSetErrorCallback(int dc, errorCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
 
-	it->second->onError([dc, errorCallback](const string &error) {
-		errorCallback(error.c_str(), getUserPointer(dc));
-	});
+	dataChannel->onError([dc, cb](const string &error) { cb(error.c_str(), getUserPointer(dc)); });
+	return 0;
 }
 
-void rtcSetMessageCallback(int dc, void (*messageCallback)(const char *, int, void *)) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return;
+int rtcSetMessageCallback(int dc, messageCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
 
-	it->second->onMessage(
-	    [dc, messageCallback](const binary &b) {
-		    messageCallback(reinterpret_cast<const char *>(b.data()), b.size(), getUserPointer(dc));
+	dataChannel->onMessage(
+	    [dc, cb](const binary &b) {
+		    cb(reinterpret_cast<const char *>(b.data()), b.size(), getUserPointer(dc));
 	    },
-	    [dc, messageCallback](const string &s) {
-		    messageCallback(s.c_str(), -1, getUserPointer(dc));
-	    });
+	    [dc, cb](const string &s) { cb(s.c_str(), -1, getUserPointer(dc)); });
+	return 0;
 }
 
 int rtcSendMessage(int dc, const char *data, int size) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return 0;
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
 
 	if (size >= 0) {
 		auto b = reinterpret_cast<const byte *>(data);
-		it->second->send(b, size);
+		CATCH(dataChannel->send(b, size));
 		return size;
 	} else {
 		string s(data);
-		it->second->send(s);
+		CATCH(dataChannel->send(s));
 		return s.size();
 	}
 }
 
-void rtcSetUserPointer(int i, void *ptr) {
-	if (ptr)
-		userPointerMap.insert(std::make_pair(i, ptr));
-	else
-		userPointerMap.erase(i);
+int rtcGetBufferedAmount(int dc) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	CATCH(return int(dataChannel->bufferedAmount()));
+}
+
+int rtcSetBufferedAmountLowThreshold(int dc, int amount) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	CATCH(dataChannel->setBufferedAmountLowThreshold(size_t(amount)));
+	return 0;
+}
+
+int rtcSetBufferedAmountLowCallback(int dc, bufferedAmountLowCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	dataChannel->onOpen([dc, cb]() { cb(getUserPointer(dc)); });
+	return 0;
+}
+
+int rtcGetAvailableAmount(int dc) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	CATCH(return int(dataChannel->availableAmount()));
+}
+
+int rtcSetAvailableCallback(int dc, availableCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	dataChannel->onOpen([dc, cb]() { cb(getUserPointer(dc)); });
+	return 0;
+}
+
+int rtcReceiveMessage(int dc, char *buffer, int *size) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (!size)
+		return -1;
+
+	CATCH({
+		auto message = dataChannel->receive();
+		if (!message)
+			return 0;
+
+		return std::visit( //
+		    overloaded{    //
+		               [&](const binary &b) {
+			               *size = std::min(*size, int(b.size()));
+			               auto data = reinterpret_cast<const char *>(b.data());
+			               std::copy(data, data + *size, buffer);
+			               return *size;
+		               },
+		               [&](const string &s) {
+			               int len = std::min(*size - 1, int(s.size()));
+			               *size = -1;
+			               if (len >= 0) {
+				               std::copy(s.data(), s.data() + len, buffer);
+				               buffer[len] = '\0';
+			               }
+			               return len + 1;
+		               }},
+		    *message);
+	});
 }