aboutsummaryrefslogtreecommitdiffstats
path: root/libtransport
diff options
context:
space:
mode:
Diffstat (limited to 'libtransport')
-rw-r--r--libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc25
-rw-r--r--libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h2
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_producer.h4
3 files changed, 19 insertions, 12 deletions
diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc
index c726dfda8..6a45019a4 100644
--- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc
+++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc
@@ -24,8 +24,9 @@
#define INIT_PACKET_PRODUCTION_RATE 100 // pps random value (almost 1Mbps)
#define STATS_INTERVAL_DURATION 500 // ms
#define INTEREST_LIFETIME_REDUCTION_FACTOR 0.8
-#define INACTIVE_TIME 500 //ms without producing before the socket
- //is considered inactive
+#define INACTIVE_TIME \
+ 500 // ms without producing before the socket
+ // is considered inactive
#define MILLI_IN_A_SEC 1000 // ms in a second
// NACK HEADER
@@ -113,7 +114,9 @@ void RTCProducerSocket::updateStats(uint32_t packet_size, uint64_t now) {
}
}
-void RTCProducerSocket::produce(const uint8_t *buf, size_t buffer_size) {
+void RTCProducerSocket::produce(std::unique_ptr<utils::MemBuf> &&buffer) {
+ auto buffer_size = buffer->length();
+
if (TRANSPORT_EXPECT_FALSE(buffer_size == 0)) {
return;
}
@@ -137,11 +140,11 @@ void RTCProducerSocket::produce(const uint8_t *buf, size_t buffer_size) {
ContentObject content_object(flowName_.setSuffix(currentSeg_));
- auto payload = utils::MemBuf::create(buffer_size + TIMESTAMP_LEN);
+ auto payload = utils::MemBuf::create(TIMESTAMP_LEN);
memcpy(payload->writableData(), &now, TIMESTAMP_LEN);
- memcpy(payload->writableData() + TIMESTAMP_LEN, buf, buffer_size);
- payload->append(buffer_size + TIMESTAMP_LEN);
+ payload->append(TIMESTAMP_LEN);
+ payload->prependChain(std::move(buffer));
content_object.appendPayload(std::move(payload));
content_object.setLifetime(500); // XXX this should be set by the APP
@@ -169,14 +172,14 @@ void RTCProducerSocket::onInterest(Interest::Ptr &&interest) {
{
utils::SpinLock::Acquire locked(lock_);
isActive = active_;
- if(isActive){
+ if (isActive) {
uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
- std::chrono::steady_clock::now().time_since_epoch())
- .count();
+ std::chrono::steady_clock::now().time_since_epoch())
+ .count();
if ((now - lastProduced_) > INACTIVE_TIME) {
- //socket is inactive
+ // socket is inactive
active_ = false;
- isActive = false;
+ isActive = false;
}
}
}
diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h
index 29fd15a4e..5b9a23dd7 100644
--- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h
+++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h
@@ -36,7 +36,7 @@ class RTCProducerSocket : public ProducerSocket {
void registerPrefix(const Prefix &producer_namespace) override;
- void produce(const uint8_t *buffer, size_t buffer_size) override;
+ void produce(std::unique_ptr<utils::MemBuf> &&buffer) override;
void onInterest(Interest::Ptr &&interest) override;
diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h
index cd1c5a374..18adbf4a7 100644
--- a/libtransport/src/hicn/transport/interfaces/socket_producer.h
+++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h
@@ -62,6 +62,10 @@ class ProducerSocket : public Socket<BasePortal>,
void produce(ContentObject &content_object);
virtual void produce(const uint8_t *buffer, size_t buffer_size) {
+ produce(utils::MemBuf::copyBuffer(buffer, buffer_size));
+ }
+
+ virtual void produce(std::unique_ptr<utils::MemBuf> &&buffer) {
// This API is meant to be used just with the RTC producer.
// Here it cannot be used since no name for the content is specified.
throw errors::NotImplementedException();