+
+#include "core_v2/payload.h"
+#include "proto/connections/offline_wire_formats.pb.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/base/exception.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+// Defines the operations layered atop a Payload, for use inside the
+// OfflineServiceController.
+//
+// There will be an extension of this abstract base class per type of
+// Payload.
+class InternalPayload {
+ public:
+ explicit InternalPayload(Payload payload);
+ virtual ~InternalPayload() = default;
+
+ Payload ReleasePayload();
+
+ Payload::Id GetId() const;
+
+ // Returns the PayloadType of the Payload to which this object is bound.
+ //
+ //
Note that this is supposed to return the type from the OfflineFrame
+ // proto rather than what is already available via
+ // Payload::getType().
+ //
+ // @return The PayloadType.
+ virtual PayloadTransferFrame::PayloadHeader::PayloadType GetType() const = 0;
+
+ // Deduces the total size of the Payload to which this object is bound.
+ //
+ // @return The total size, or -1 if it cannot be deduced (for example, when
+ // dealing with streaming data).
+ virtual std::int64_t GetTotalSize() const = 0;
+
+ // Breaks off the next chunk from the Payload to which this object is bound.
+ //
+ //
Used when we have a complete Payload that we want to break into smaller
+ // byte blobs for sending across a hard boundary (like the other side of
+ // a Binder, or another device altogether).
+ //
+ // @return The next chunk from the Payload, or null if we've reached the end.
+ virtual ByteArray DetachNextChunk() = 0;
+
+ // Adds the next chunk that comprises the Payload to which this object is
+ // bound.
+ //
+ //
Used when we are trying to reconstruct a Payload that lives on the
+ // other side of a hard boundary (like the other side of a Binder, or another
+ // device altogether), one byte blob at a time.
+ //
+ // @param chunk The next chunk; this being null signals that this is the last
+ // chunk, which will typically be used as a trigger to perform whatever state
+ // cleanup may be required by the concrete implementation.
+ virtual Exception AttachNextChunk(const ByteArray& chunk) = 0;
+
+ // Cleans up any resources used by this Payload. Called when we're stopping
+ // early, e.g. after being cancelled or having no more recipients left.
+ virtual void Close() {}
+
+ protected:
+ Payload payload_;
+ // We're caching the payload ID here because the backing payload will be
+ // released to another owner during the lifetime of an incoming
+ // InternalPayload.
+ Payload::Id payload_id_;
+};
+
+} // namespace connections
+} // namespace nearby
+} // namespace location
+
+#endif // CORE_V2_INTERNAL_INTERNAL_PAYLOAD_H_
diff --git a/cpp/core_v2/internal/internal_payload_factory.cc b/cpp/core_v2/internal/internal_payload_factory.cc
new file mode 100644
index 00000000..240882f2
--- /dev/null
+++ b/cpp/core_v2/internal/internal_payload_factory.cc
@@ -0,0 +1,293 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "core_v2/internal/internal_payload_factory.h"
+
+#include
+#include
+
+#include "core_v2/payload.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/base/exception.h"
+#include "platform_v2/public/condition_variable.h"
+#include "platform_v2/public/file.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/pipe.h"
+#include "absl/memory/memory.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+namespace {
+
+class BytesInternalPayload : public InternalPayload {
+ public:
+ explicit BytesInternalPayload(Payload payload)
+ : InternalPayload(std::move(payload)),
+ total_size_(payload_.AsBytes().size()),
+ detached_only_chunk_(false) {}
+
+ PayloadTransferFrame::PayloadHeader::PayloadType GetType() const override {
+ return PayloadTransferFrame::PayloadHeader::BYTES;
+ }
+
+ std::int64_t GetTotalSize() const override { return total_size_; }
+
+ // Relinquishes ownership of the payload_; retrieves and returns the stored
+ // ByteArray.
+ ByteArray DetachNextChunk() override {
+ if (detached_only_chunk_) {
+ return {};
+ }
+
+ detached_only_chunk_ = true;
+ return std::move(payload_).AsBytes();
+ }
+
+ // Does nothing.
+ Exception AttachNextChunk(const ByteArray& chunk) override {
+ return {Exception::kSuccess};
+ }
+
+ private:
+ // We're caching the total size here because the backing payload will be
+ // moved to another owner during the lifetime of an incoming
+ // InternalPayload.
+ const std::int64_t total_size_;
+ bool detached_only_chunk_;
+};
+
+class OutgoingStreamInternalPayload : public InternalPayload {
+ public:
+ explicit OutgoingStreamInternalPayload(Payload payload)
+ : InternalPayload(std::move(payload)) {}
+
+ PayloadTransferFrame::PayloadHeader::PayloadType GetType() const override {
+ return PayloadTransferFrame::PayloadHeader::STREAM;
+ }
+
+ std::int64_t GetTotalSize() const override { return -1; }
+
+ ByteArray DetachNextChunk() override {
+ InputStream* input_stream = payload_.AsStream();
+ if (!input_stream) return {};
+
+ ExceptionOr bytes_read = input_stream->Read(kChunkSize);
+ if (!bytes_read.ok()) {
+ input_stream->Close();
+ return {};
+ }
+
+ ByteArray scoped_bytes_read = std::move(bytes_read.result());
+
+ if (scoped_bytes_read.Empty()) {
+ // TODO(reznor): logger.atVerbose().log("No more data for outgoing payload
+ // %s, closing InputStream.", this);
+
+ input_stream->Close();
+ return {};
+ }
+
+ return scoped_bytes_read;
+ }
+
+ Exception AttachNextChunk(const ByteArray& chunk) override {
+ return {Exception::kIo};
+ }
+
+ void Close() override {
+ // Ignore the potential Exception returned by close(), as a counterpart
+ // to Java's closeQuietly().
+ InputStream* stream = payload_.AsStream();
+ if (stream) stream->Close();
+ }
+
+ private:
+ static constexpr std::int64_t kChunkSize = Pipe::kChunkSize;
+};
+
+class IncomingStreamInternalPayload : public InternalPayload {
+ public:
+ IncomingStreamInternalPayload(Payload payload, OutputStream& output_stream)
+ : InternalPayload(std::move(payload)), output_stream_(&output_stream) {}
+
+ PayloadTransferFrame::PayloadHeader::PayloadType GetType() const override {
+ return PayloadTransferFrame::PayloadHeader::STREAM;
+ }
+
+ std::int64_t GetTotalSize() const override { return -1; }
+
+ ByteArray DetachNextChunk() override { return {}; }
+
+ Exception AttachNextChunk(const ByteArray& chunk) override {
+ if (chunk.Empty()) {
+ output_stream_->Close();
+ return {Exception::kSuccess};
+ }
+
+ return output_stream_->Write(chunk);
+ }
+
+ void Close() override { output_stream_->Close(); }
+
+ private:
+ OutputStream* output_stream_;
+};
+
+class OutgoingFileInternalPayload : public InternalPayload {
+ public:
+ explicit OutgoingFileInternalPayload(Payload payload)
+ : InternalPayload(std::move(payload)),
+ total_size_{payload_.AsFile()->GetTotalSize()} {}
+
+ PayloadTransferFrame::PayloadHeader::PayloadType GetType() const override {
+ return PayloadTransferFrame::PayloadHeader::FILE;
+ }
+
+ std::int64_t GetTotalSize() const override { return total_size_; }
+
+ ByteArray DetachNextChunk() override {
+ InputFile* file = payload_.AsFile();
+ if (!file) return {};
+
+ ExceptionOr bytes_read = file->Read(kChunkSize);
+ if (!bytes_read.ok()) {
+ return {};
+ }
+
+ ByteArray bytes = std::move(bytes_read.result());
+
+ if (bytes.Empty()) {
+ // No more data for outgoing payload.
+
+ file->Close();
+ return {};
+ }
+
+ return bytes;
+ }
+
+ Exception AttachNextChunk(const ByteArray& chunk) override {
+ return {Exception::kIo};
+ }
+
+ void Close() override {
+ InputFile* file = payload_.AsFile();
+ if (file) file->Close();
+ }
+
+ private:
+ std::int64_t total_size_;
+ static constexpr std::int64_t kChunkSize = 64 * 1024;
+};
+
+class IncomingFileInternalPayload : public InternalPayload {
+ public:
+ IncomingFileInternalPayload(Payload payload, OutputFile output_file,
+ std::int64_t total_size)
+ : InternalPayload(std::move(payload)),
+ output_file_(std::move(output_file)),
+ total_size_(total_size) {}
+
+ PayloadTransferFrame::PayloadHeader::PayloadType GetType() const override {
+ return PayloadTransferFrame::PayloadHeader::FILE;
+ }
+
+ std::int64_t GetTotalSize() const override { return total_size_; }
+
+ ByteArray DetachNextChunk() override { return {}; }
+
+ Exception AttachNextChunk(const ByteArray& chunk) override {
+ if (chunk.Empty()) {
+ // Received null last chunk for incoming payload.
+ output_file_.Close();
+ return {Exception::kSuccess};
+ }
+
+ return output_file_.Write(chunk);
+ }
+
+ void Close() override { output_file_.Close(); }
+
+ private:
+ OutputFile output_file_;
+ const std::int64_t total_size_;
+};
+
+} // namespace
+
+std::unique_ptr CreateOutgoingInternalPayload(
+ Payload payload) {
+ switch (payload.GetType()) {
+ case Payload::Type::kBytes:
+ return absl::make_unique(std::move(payload));
+
+ case Payload::Type::kFile: {
+ InputFile* file = payload.AsFile();
+ const PayloadId file_payload_id = file ? file->GetPayloadId() : 0;
+ const PayloadId payload_id = payload.GetId();
+ CHECK(payload_id == file_payload_id);
+ return absl::make_unique(std::move(payload));
+ }
+
+ case Payload::Type::kStream:
+ return absl::make_unique(
+ std::move(payload));
+
+ default:
+ DCHECK(false); // This should never happen.
+ return {};
+ }
+}
+
+std::unique_ptr CreateIncomingInternalPayload(
+ const PayloadTransferFrame& frame) {
+ if (frame.packet_type() != PayloadTransferFrame::DATA) {
+ return {};
+ }
+
+ const Payload::Id payload_id = frame.payload_header().id();
+ switch (frame.payload_header().type()) {
+ case PayloadTransferFrame::PayloadHeader::BYTES: {
+ return absl::make_unique(
+ Payload(payload_id, ByteArray(frame.payload_chunk().body())));
+ }
+
+ case PayloadTransferFrame::PayloadHeader::STREAM: {
+ auto pipe = std::make_shared();
+
+ return absl::make_unique(
+ Payload(payload_id,
+ [pipe]() -> InputStream& {
+ return pipe->GetInputStream(); // NOLINT
+ }),
+ pipe->GetOutputStream());
+ }
+
+ case PayloadTransferFrame::PayloadHeader::FILE: {
+ std::int64_t total_size = frame.payload_header().total_size();
+ return absl::make_unique(
+ Payload(payload_id, InputFile(payload_id, total_size)),
+ OutputFile(payload_id), total_size);
+ }
+ default:
+ DCHECK(false); // This should never happen.
+ return {};
+ }
+}
+
+} // namespace connections
+} // namespace nearby
+} // namespace location
diff --git a/cpp/core_v2/internal/internal_payload_factory.h b/cpp/core_v2/internal/internal_payload_factory.h
new file mode 100644
index 00000000..3d283a28
--- /dev/null
+++ b/cpp/core_v2/internal/internal_payload_factory.h
@@ -0,0 +1,38 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_V2_INTERNAL_INTERNAL_PAYLOAD_FACTORY_H_
+#define CORE_V2_INTERNAL_INTERNAL_PAYLOAD_FACTORY_H_
+
+#include "core_v2/internal/internal_payload.h"
+#include "core_v2/payload.h"
+#include "proto/connections/offline_wire_formats.pb.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+// Creates an InternalPayload representing an outgoing Payload.
+std::unique_ptr CreateOutgoingInternalPayload(Payload payload);
+
+// Creates an InternalPayload representing an incoming Payload from a remote
+// endpoint.
+std::unique_ptr CreateIncomingInternalPayload(
+ const PayloadTransferFrame& frame);
+
+} // namespace connections
+} // namespace nearby
+} // namespace location
+
+#endif // CORE_V2_INTERNAL_INTERNAL_PAYLOAD_FACTORY_H_
diff --git a/cpp/core_v2/internal/internal_payload_factory_test.cc b/cpp/core_v2/internal/internal_payload_factory_test.cc
new file mode 100644
index 00000000..be98af4d
--- /dev/null
+++ b/cpp/core_v2/internal/internal_payload_factory_test.cc
@@ -0,0 +1,130 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "core_v2/internal/internal_payload_factory.h"
+
+#include "core_v2/internal/offline_frames.h"
+#include "proto/connections/offline_wire_formats.pb.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/public/pipe.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace {
+
+constexpr char kText[] = "data chunk";
+
+TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromBytePayload) {
+ ByteArray data(kText);
+ std::unique_ptr internal_payload =
+ CreateOutgoingInternalPayload(Payload{data});
+ EXPECT_NE(internal_payload, nullptr);
+ Payload payload = internal_payload->ReleasePayload();
+ EXPECT_EQ(payload.AsFile(), nullptr);
+ EXPECT_EQ(payload.AsStream(), nullptr);
+ EXPECT_EQ(payload.AsBytes(), ByteArray(kText));
+}
+
+TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamPayload) {
+ auto pipe = std::make_shared();
+ std::unique_ptr internal_payload =
+ CreateOutgoingInternalPayload(Payload{[pipe]() -> InputStream& {
+ return pipe->GetInputStream(); // NOLINT
+ }});
+ EXPECT_NE(internal_payload, nullptr);
+ Payload payload = internal_payload->ReleasePayload();
+ EXPECT_EQ(payload.AsFile(), nullptr);
+ EXPECT_NE(payload.AsStream(), nullptr);
+ EXPECT_EQ(payload.AsBytes(), ByteArray());
+}
+
+TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromFilePayload) {
+ Payload::Id payload_id = Payload::GenerateId();
+ std::unique_ptr internal_payload =
+ CreateOutgoingInternalPayload(
+ Payload{payload_id, InputFile(payload_id, 512)});
+ EXPECT_NE(internal_payload, nullptr);
+ Payload payload = internal_payload->ReleasePayload();
+ EXPECT_NE(payload.AsFile(), nullptr);
+ EXPECT_EQ(payload.AsStream(), nullptr);
+ EXPECT_EQ(payload.AsBytes(), ByteArray());
+ EXPECT_EQ(payload.GetId(), payload_id);
+ EXPECT_EQ(payload.AsFile()->GetPayloadId(), payload_id);
+}
+
+TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromByteMessage) {
+ PayloadTransferFrame frame;
+ frame.set_packet_type(PayloadTransferFrame::DATA);
+ std::int64_t payload_chunk_offset = 0;
+ ByteArray data(kText);
+ PayloadTransferFrame::PayloadChunk payload_chunk;
+ payload_chunk.set_offset(payload_chunk_offset);
+ payload_chunk.set_body(std::string(std::move(data)));
+ payload_chunk.set_flags(0);
+ auto& header = *frame.mutable_payload_header();
+ header.set_type(PayloadTransferFrame::PayloadHeader::BYTES);
+ header.set_id(12345);
+ header.set_total_size(512);
+ *frame.mutable_payload_chunk() = std::move(payload_chunk);
+ std::unique_ptr internal_payload =
+ CreateIncomingInternalPayload(frame);
+ EXPECT_NE(internal_payload, nullptr);
+ Payload payload = internal_payload->ReleasePayload();
+ EXPECT_EQ(payload.AsFile(), nullptr);
+ EXPECT_EQ(payload.AsStream(), nullptr);
+ EXPECT_EQ(payload.AsBytes(), ByteArray(kText));
+}
+
+TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamMessage) {
+ PayloadTransferFrame frame;
+ frame.set_packet_type(PayloadTransferFrame::DATA);
+ auto& header = *frame.mutable_payload_header();
+ header.set_type(PayloadTransferFrame::PayloadHeader::STREAM);
+ header.set_id(12345);
+ header.set_total_size(0);
+ std::unique_ptr internal_payload =
+ CreateIncomingInternalPayload(frame);
+ EXPECT_NE(internal_payload, nullptr);
+ Payload payload = internal_payload->ReleasePayload();
+ EXPECT_EQ(payload.AsFile(), nullptr);
+ EXPECT_NE(payload.AsStream(), nullptr);
+ EXPECT_EQ(payload.AsBytes(), ByteArray());
+ EXPECT_EQ(payload.GetType(), Payload::Type::kStream);
+}
+
+TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromFileMessage) {
+ PayloadTransferFrame frame;
+ frame.set_packet_type(PayloadTransferFrame::DATA);
+ auto& header = *frame.mutable_payload_header();
+ header.set_type(PayloadTransferFrame::PayloadHeader::FILE);
+ header.set_id(12345);
+ header.set_total_size(512);
+ std::unique_ptr internal_payload =
+ CreateIncomingInternalPayload(frame);
+ EXPECT_NE(internal_payload, nullptr);
+ Payload payload = internal_payload->ReleasePayload();
+ EXPECT_NE(payload.AsFile(), nullptr);
+ EXPECT_EQ(payload.AsStream(), nullptr);
+ EXPECT_EQ(payload.AsBytes(), ByteArray());
+ EXPECT_EQ(payload.GetType(), Payload::Type::kFile);
+ EXPECT_EQ(payload.GetId(), payload.AsFile()->GetPayloadId());
+}
+
+} // namespace
+} // namespace connections
+} // namespace nearby
+} // namespace location
diff --git a/cpp/core_v2/internal/mediums/BUILD b/cpp/core_v2/internal/mediums/BUILD
index afd9116a..939ef43d 100644
--- a/cpp/core_v2/internal/mediums/BUILD
+++ b/cpp/core_v2/internal/mediums/BUILD
@@ -24,6 +24,8 @@ cc_library(
"bluetooth_radio.cc",
"mediums.cc",
"uuid.cc",
+ "webrtc.cc",
+ "wifi_lan.cc",
],
hdrs = [
"advertisement_read_result.h",
@@ -37,22 +39,29 @@ cc_library(
"lost_entity_tracker.h",
"mediums.h",
"uuid.h",
+ "webrtc.h",
+ "wifi_lan.h",
],
visibility = [
"//core_v2/internal:__subpackages__",
],
deps = [
"//core_v2:core_types",
+ "//core_v2/internal/mediums/webrtc",
"//platform_v2/base",
+ "//platform_v2/base:util",
"//platform_v2/public:comm",
"//platform_v2/public:logging",
"//platform_v2/public:types",
+ "//location/nearby/mediums/proto:web_rtc_signaling_frames_cc_proto",
"//absl/container:flat_hash_map",
"//absl/container:flat_hash_set",
"//absl/numeric:int128",
"//absl/strings",
"//absl/time",
"//smhasher:libmurmur3",
+ "//webrtc/api:libjingle_peerconnection_api",
+ "//webrtc/api:scoped_refptr",
],
)
@@ -84,10 +93,13 @@ cc_test(
"bluetooth_radio_test.cc",
"lost_entity_tracker_test.cc",
"uuid_test.cc",
+ "webrtc_test.cc",
+ "wifi_lan_test.cc",
],
shard_count = 16,
deps = [
":mediums",
+ "//core_v2/internal/mediums/webrtc",
"//platform_v2/base",
"//platform_v2/base:test_util",
"//platform_v2/impl/g3", # build_cleaner: keep
diff --git a/cpp/core_v2/internal/mediums/ble_advertisement.cc b/cpp/core_v2/internal/mediums/ble_advertisement.cc
index 1fc8a6b7..238d8385 100644
--- a/cpp/core_v2/internal/mediums/ble_advertisement.cc
+++ b/cpp/core_v2/internal/mediums/ble_advertisement.cc
@@ -16,7 +16,9 @@
#include
+#include "platform_v2/base/base_input_stream.h"
#include "platform_v2/public/logging.h"
+#include "absl/strings/str_cat.h"
namespace location {
namespace nearby {
@@ -56,11 +58,15 @@ BleAdvertisement::BleAdvertisement(const ByteArray &ble_advertisement_bytes) {
return;
}
- // Now, time to read the bytes!
- const auto *read_ptr = ble_advertisement_bytes.data();
+ ByteArray advertisement_bytes{ble_advertisement_bytes};
+ BaseInputStream base_input_stream{advertisement_bytes};
+ // The first 1 byte is supposed to be the version and socket version.
+ auto version_and_socket_version_byte =
+ static_cast(base_input_stream.ReadUint8());
- // 1. Version.
- version_ = static_cast((*read_ptr & kVersionBitmask) >> 5);
+ // Version.
+ version_ = static_cast(
+ (version_and_socket_version_byte & kVersionBitmask) >> 5);
if (!IsSupportedVersion(version_)) {
NEARBY_LOG(INFO,
"Cannot deserialize BleAdvertisement: unsupported Version %u",
@@ -68,49 +74,42 @@ BleAdvertisement::BleAdvertisement(const ByteArray &ble_advertisement_bytes) {
return;
}
- // 2. Socket Version.
- socket_version_ =
- static_cast((*read_ptr & kSocketVersionBitmask) >> 2);
+ // Socket version.
+ socket_version_ = static_cast(
+ (version_and_socket_version_byte & kSocketVersionBitmask) >> 2);
if (!IsSupportedSocketVersion(socket_version_)) {
NEARBY_LOG(
INFO,
- "Cannot deserialize BLEAdvertisement: unsupported SocketVersion %u",
+ "Cannot deserialize BleAdvertisement: unsupported SocketVersion %u",
socket_version_);
version_ = Version::kUndefined;
return;
}
- read_ptr += kVersionLength;
- // 3. Service ID hash.
- service_id_hash_ = ByteArray(read_ptr, kServiceIdHashLength);
- read_ptr += kServiceIdHashLength;
+ // The next 3 bytes are supposed to be the service_id_hash.
+ service_id_hash_ = base_input_stream.ReadBytes(kServiceIdHashLength);
- // 4.1. Data size.
- size_t expected_data_size = DeserializeDataSize(read_ptr);
+ // The next 4 bytes are supposed to be the length of the data.
+ std::uint32_t expected_data_size = base_input_stream.ReadUint32();
if (expected_data_size < 0) {
- NEARBY_LOG(
- INFO,
- "Cannot deserialize BleAdvertisement: negative data size %" PRIu64,
- expected_data_size);
- version_ = Version::kUndefined;
- return;
- }
- read_ptr += kDataSizeLength;
-
- // Check that the stated data size is the same as what we received.
- size_t actual_data_size = ComputeDataSize(ble_advertisement_bytes);
- if (actual_data_size < expected_data_size) {
NEARBY_LOG(INFO,
- "Cannot deserialize BLEAdvertisement: expected data to be %zu "
- "bytes, got %" PRIu64 " bytes",
- expected_data_size, actual_data_size);
+ "Cannot deserialize BleAdvertisement: negative data size %d",
+ expected_data_size);
version_ = Version::kUndefined;
return;
}
- // 4.2. Data.
- data_ = ByteArray(read_ptr, expected_data_size);
- read_ptr += expected_data_size;
+ // The rest bytes are supposed to be the data.
+ // Check that the stated data size is the same as what we received.
+ data_ = base_input_stream.ReadBytes(expected_data_size);
+ if (data_.size() != expected_data_size) {
+ NEARBY_LOG(INFO,
+ "Cannot deserialize BleAdvertisement: expected data to be %u "
+ "bytes, got %" PRIu64 " bytes ",
+ expected_data_size, data_.size());
+ version_ = Version::kUndefined;
+ return;
+ }
}
BleAdvertisement::operator ByteArray() const {
@@ -118,8 +117,6 @@ BleAdvertisement::operator ByteArray() const {
return ByteArray{};
}
- std::string out;
-
// The first 3 bits are the Version.
char version_and_socket_version_byte =
(static_cast(version_) << 5) & kVersionBitmask;
@@ -131,11 +128,13 @@ BleAdvertisement::operator ByteArray() const {
auto *data_size_bytes_write_ptr = data_size_bytes.data();
SerializeDataSize(data_size_bytes_write_ptr, data_.size());
- out.reserve(1 + service_id_hash_.size() + 1 + data_.size());
- out.append(1, version_and_socket_version_byte);
- out.append(std::string(service_id_hash_));
- out.append(std::string(data_size_bytes));
- out.append(std::string(data_));
+ // clang-format off
+ std::string out =
+ absl::StrCat(std::string(1, version_and_socket_version_byte),
+ std::string(service_id_hash_),
+ std::string(data_size_bytes),
+ std::string(data_));
+ // clang-format on
return ByteArray{std::move(out)};
}
@@ -182,33 +181,6 @@ void BleAdvertisement::SerializeDataSize(char *data_size_bytes_write_ptr,
}
}
-size_t BleAdvertisement::DeserializeDataSize(
- const char *data_size_bytes_read_ptr) const {
- // Allocate a chunk of memory to store our deserialized size.
- char data_size_bytes[kDataSizeLength];
-
- // Assign the bits of our size from the given raw bytes, keeping in mind that
- // we need to convert from Big Endian to Little Endian in the process.
- for (int i = 0; i < kDataSizeLength; ++i) {
- data_size_bytes[i] = data_size_bytes_read_ptr[kDataSizeLength - i - 1];
- }
-
- // Interpret the char array as a single int.
- return static_cast(
- *(reinterpret_cast(&data_size_bytes)));
-}
-
-size_t BleAdvertisement::ComputeDataSize(
- const ByteArray &ble_advertisement_bytes) const {
- return ble_advertisement_bytes.size() - kMinAdvertisementLength;
-}
-
-size_t BleAdvertisement::ComputeAdvertisementLength(
- const ByteArray &data) const {
- // The advertisement length is the minimum length + the length of the data.
- return kMinAdvertisementLength + data.size();
-}
-
} // namespace mediums
} // namespace connections
} // namespace nearby
diff --git a/cpp/core_v2/internal/mediums/ble_advertisement.h b/cpp/core_v2/internal/mediums/ble_advertisement.h
index a64de73d..5ead482e 100644
--- a/cpp/core_v2/internal/mediums/ble_advertisement.h
+++ b/cpp/core_v2/internal/mediums/ble_advertisement.h
@@ -81,9 +81,6 @@ class BleAdvertisement {
bool IsSupportedSocketVersion(SocketVersion socket_version) const;
void SerializeDataSize(char *data_size_bytes_write_ptr,
size_t data_size) const;
- size_t DeserializeDataSize(const char *data_size_bytes_read_ptr) const;
- size_t ComputeDataSize(const ByteArray &ble_advertisement_bytes) const;
- size_t ComputeAdvertisementLength(const ByteArray &data) const;
static constexpr int kVersionLength = 1;
// Length of one int. Be sure to re-evaluate how we compute data size in this
diff --git a/cpp/core_v2/internal/mediums/ble_advertisement_header.cc b/cpp/core_v2/internal/mediums/ble_advertisement_header.cc
index 768fd00d..3bca06fc 100644
--- a/cpp/core_v2/internal/mediums/ble_advertisement_header.cc
+++ b/cpp/core_v2/internal/mediums/ble_advertisement_header.cc
@@ -17,7 +17,9 @@
#include
#include "platform_v2/base/base64_utils.h"
+#include "platform_v2/base/base_input_stream.h"
#include "platform_v2/public/logging.h"
+#include "absl/strings/str_cat.h"
namespace location {
namespace nearby {
@@ -27,8 +29,7 @@ namespace mediums {
BleAdvertisementHeader::BleAdvertisementHeader(
Version version, int num_slots, const ByteArray &service_id_bloom_filter,
const ByteArray &advertisement_hash) {
- // TODO(edwinwu): Checks if num_slots needs to be >= 0
- if (version != Version::kV2 ||
+ if (version != Version::kV2 || num_slots <= 0 ||
service_id_bloom_filter.size() != kServiceIdBloomFilterLength ||
advertisement_hash.size() != kAdvertisementHashLength) {
return;
@@ -61,13 +62,12 @@ BleAdvertisementHeader::BleAdvertisementHeader(
return;
}
- // Start reading the bytes.
- auto *ble_advertisement_header_read_ptr =
- ble_advertisement_header_bytes.data();
-
- // The first 3 bits are supposed to be the version.
- version_ = static_cast(
- (*ble_advertisement_header_read_ptr & kVersionBitmask) >> 5);
+ BaseInputStream base_input_stream{ble_advertisement_header_bytes};
+ // The first 1 byte is supposed to be the version and number of slots.
+ auto version_and_pcp_byte = static_cast(base_input_stream.ReadUint8());
+ // The upper 3 bits are supposed to be the version.
+ version_ =
+ static_cast((version_and_pcp_byte & kVersionBitmask) >> 5);
if (version_ != Version::kV2) {
NEARBY_LOG(
ERROR,
@@ -75,20 +75,19 @@ BleAdvertisementHeader::BleAdvertisementHeader(
version_);
return;
}
- // The last 5 bits of the first byte represent the number of slots.
- num_slots_ = static_cast(*ble_advertisement_header_read_ptr &
- kNumSlotsBitmask);
- ble_advertisement_header_read_ptr++;
+ // The lower 5 bits are supposed to be the number of slots.
+ num_slots_ = static_cast(version_and_pcp_byte & kNumSlotsBitmask);
+ if (num_slots_ <= 0) {
+ version_ = Version::kUndefined;
+ return;
+ }
- // Service ID bloom filter.
+ // The next 10 bytes are supposed to be the service_id_bloom_filter.
service_id_bloom_filter_ =
- ByteArray(ble_advertisement_header_read_ptr, kServiceIdBloomFilterLength);
- ble_advertisement_header_read_ptr += kServiceIdBloomFilterLength;
+ base_input_stream.ReadBytes(kServiceIdBloomFilterLength);
- // Advertisement hash.
- advertisement_hash_ =
- ByteArray(ble_advertisement_header_read_ptr, kAdvertisementHashLength);
- ble_advertisement_header_read_ptr += kAdvertisementHashLength;
+ // The next 4 bytes are supposed to be the advertisement_hash.
+ advertisement_hash_ = base_input_stream.ReadBytes(kAdvertisementHashLength);
}
BleAdvertisementHeader::operator std::string() const {
@@ -96,18 +95,18 @@ BleAdvertisementHeader::operator std::string() const {
return "";
}
- std::string out;
-
// The first 3 bits are the Version.
char version_and_num_slots_byte =
(static_cast(version_) << 5) & kVersionBitmask;
// The next 5 bits are the number of slots.
version_and_num_slots_byte |=
static_cast(num_slots_) & kNumSlotsBitmask;
- out.reserve(1 + service_id_bloom_filter_.size() + advertisement_hash_.size());
- out.append(1, version_and_num_slots_byte);
- out.append(std::string(service_id_bloom_filter_));
- out.append(std::string(advertisement_hash_));
+
+ // clang-format off
+ std::string out = absl::StrCat(std::string(1, version_and_num_slots_byte),
+ std::string(service_id_bloom_filter_),
+ std::string(advertisement_hash_));
+ // clang-format on
return Base64Utils::Encode(ByteArray(std::move(out)));
}
diff --git a/cpp/core_v2/internal/mediums/ble_advertisement_header_test.cc b/cpp/core_v2/internal/mediums/ble_advertisement_header_test.cc
index b85be98b..5665eff4 100644
--- a/cpp/core_v2/internal/mediums/ble_advertisement_header_test.cc
+++ b/cpp/core_v2/internal/mediums/ble_advertisement_header_test.cc
@@ -22,16 +22,17 @@ namespace nearby {
namespace connections {
namespace mediums {
namespace {
+
constexpr BleAdvertisementHeader::Version kVersion =
BleAdvertisementHeader::Version::kV2;
constexpr int kNumSlots = 2;
-constexpr char kServiceIDBloomFilter[] =
- "\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a";
-constexpr char kAdvertisementHash[] = "\x0a\x0b\x0c\x0d";
+constexpr absl::string_view kServiceIDBloomFilter{
+ "\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a"};
+constexpr absl::string_view kAdvertisementHash{"\x0a\x0b\x0c\x0d"};
TEST(BleAdvertisementHeaderTest, ConstructionWorks) {
- ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
- ByteArray advertisement_hash{kAdvertisementHash};
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
BleAdvertisementHeader ble_advertisement_header{
kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
@@ -48,8 +49,8 @@ TEST(BleAdvertisementHeaderTest, ConstructionWorks) {
TEST(BleAdvertisementHeaderTest, ConstructionFailsWithBadVersion) {
auto bad_version = static_cast(666);
- ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
- ByteArray advertisement_hash{kAdvertisementHash};
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
BleAdvertisementHeader ble_advertisement_header{
bad_version, kNumSlots, service_id_bloom_filter, advertisement_hash};
@@ -57,12 +58,24 @@ TEST(BleAdvertisementHeaderTest, ConstructionFailsWithBadVersion) {
EXPECT_FALSE(ble_advertisement_header.IsValid());
}
+TEST(BleAdvertisementHeaderTest, ConstructionFailsWitZeroNumSlot) {
+ int num_slot = 0;
+
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
+
+ BleAdvertisementHeader ble_advertisement_header{
+ kVersion, num_slot, service_id_bloom_filter, advertisement_hash};
+
+ EXPECT_FALSE(ble_advertisement_header.IsValid());
+}
+
TEST(BleAdvertisementHeaderTest,
ConstructionFailsWithShortServiceIdBloomFilter) {
char short_service_id_bloom_filter[] = "\x01\x02\x03\x04\x05\x06\x07\x08\x09";
ByteArray short_service_id_bloom_filter_bytes{short_service_id_bloom_filter};
- ByteArray advertisement_hash{kAdvertisementHash};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
BleAdvertisementHeader ble_advertisement_header{
kVersion, kNumSlots, short_service_id_bloom_filter_bytes,
@@ -77,7 +90,7 @@ TEST(BleAdvertisementHeaderTest,
"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b";
ByteArray service_id_bloom_filter{long_service_id_bloom_filter};
- ByteArray advertisement_hash{kAdvertisementHash};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
BleAdvertisementHeader ble_advertisement_header{
kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
@@ -88,7 +101,7 @@ TEST(BleAdvertisementHeaderTest,
TEST(BleAdvertisementHeaderTest, ConstructionFailsWithShortAdvertisementHash) {
char short_advertisement_hash[] = "\x0a\x0b\x0c";
- ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
ByteArray advertisement_hash{short_advertisement_hash};
BleAdvertisementHeader ble_advertisement_header{
@@ -100,7 +113,7 @@ TEST(BleAdvertisementHeaderTest, ConstructionFailsWithShortAdvertisementHash) {
TEST(BleAdvertisementHeaderTest, ConstructionFailsWithLongAdvertisementHash) {
char long_advertisement_hash[] = "\x0a\x0b\x0c\x0d\x0e";
- ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
ByteArray advertisement_hash{long_advertisement_hash};
BleAdvertisementHeader ble_advertisement_header{
kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
@@ -109,8 +122,8 @@ TEST(BleAdvertisementHeaderTest, ConstructionFailsWithLongAdvertisementHash) {
}
TEST(BleAdvertisementHeaderTest, ConstructionFromSerializedStringWorks) {
- ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
- ByteArray advertisement_hash{kAdvertisementHash};
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
BleAdvertisementHeader org_ble_advertisement_header{
kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
@@ -130,8 +143,8 @@ TEST(BleAdvertisementHeaderTest, ConstructionFromSerializedStringWorks) {
}
TEST(BleAdvertisementHeaderTest, ConstructionFromExtraBytesWorks) {
- ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
- ByteArray advertisement_hash{kAdvertisementHash};
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
BleAdvertisementHeader ble_advertisement_header{
kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
@@ -159,8 +172,8 @@ TEST(BleAdvertisementHeaderTest, ConstructionFromExtraBytesWorks) {
}
TEST(BleAdvertisementHeaderTest, ConstructionFromShortLengthFails) {
- ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
- ByteArray advertisement_hash{kAdvertisementHash};
+ ByteArray service_id_bloom_filter{std::string(kServiceIDBloomFilter)};
+ ByteArray advertisement_hash{std::string(kAdvertisementHash)};
BleAdvertisementHeader ble_advertisement_header{
kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
diff --git a/cpp/core_v2/internal/mediums/ble_advertisement_test.cc b/cpp/core_v2/internal/mediums/ble_advertisement_test.cc
index 4d75137f..d8378bff 100644
--- a/cpp/core_v2/internal/mediums/ble_advertisement_test.cc
+++ b/cpp/core_v2/internal/mediums/ble_advertisement_test.cc
@@ -24,20 +24,20 @@ namespace connections {
namespace mediums {
namespace {
-const BleAdvertisement::Version kVersion = BleAdvertisement::Version::kV2;
-const BleAdvertisement::SocketVersion kSocketVersion =
+constexpr BleAdvertisement::Version kVersion = BleAdvertisement::Version::kV2;
+constexpr BleAdvertisement::SocketVersion kSocketVersion =
BleAdvertisement::SocketVersion::kV2;
-const char kServiceIDHashBytes[] = "\x0a\x0b\x0c";
-const char kData[] =
- "How much wood can a woodchuck chuck if a wood chuck would chuck wood?";
+constexpr absl::string_view kServiceIDHashBytes{"\x0a\x0b\x0c"};
+constexpr absl::string_view kData{
+ "How much wood can a woodchuck chuck if a wood chuck would chuck wood?"};
// This corresponds to the length of a specific BleAdvertisement packed with the
// kData given above. Be sure to update this if kData ever changes.
-const size_t kAdvertisementLength = 77;
-const size_t kLongAdvertisementLength = kAdvertisementLength + 1000;
+constexpr size_t kAdvertisementLength = 77;
+constexpr size_t kLongAdvertisementLength = kAdvertisementLength + 1000;
TEST(BleAdvertisementTest, ConstructionWorksV1) {
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
+ ByteArray data{std::string(kData)};
BleAdvertisement ble_advertisement{BleAdvertisement::Version::kV1,
BleAdvertisement::SocketVersion::kV1,
@@ -56,8 +56,8 @@ TEST(BleAdvertisementTest, ConstructionFailsWithBadVersion) {
BleAdvertisement::Version bad_version =
static_cast(666);
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
+ ByteArray data{std::string(kData)};
BleAdvertisement ble_advertisement{bad_version, kSocketVersion,
service_id_hash, data};
@@ -69,8 +69,8 @@ TEST(BleAdvertisementTest, ConstructionFailsWithBadSocketVersion) {
BleAdvertisement::SocketVersion bad_socket_version =
static_cast(666);
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
+ ByteArray data{std::string(kData)};
BleAdvertisement ble_advertisement{kVersion, bad_socket_version,
service_id_hash, data};
@@ -82,7 +82,7 @@ TEST(BleAdvertisementTest, ConstructionFailsWithShortServiceIdHash) {
char short_service_id_hash_bytes[] = "\x0a\x0b";
ByteArray bad_service_id_hash{short_service_id_hash_bytes};
- ByteArray data{kData};
+ ByteArray data{std::string(kData)};
BleAdvertisement ble_advertisement{kVersion, kSocketVersion,
bad_service_id_hash, data};
@@ -94,7 +94,7 @@ TEST(BleAdvertisementTest, ConstructionFailsWithLongServiceIdHash) {
char long_service_id_hash_bytes[] = "\x0a\x0b\x0c\x0d";
ByteArray bad_service_id_hash{long_service_id_hash_bytes};
- ByteArray data{kData};
+ ByteArray data{std::string(kData)};
BleAdvertisement ble_advertisement{kVersion, kSocketVersion,
bad_service_id_hash, data};
@@ -107,7 +107,7 @@ TEST(BleAdvertisementTest, ConstructionFailsWithLongData) {
// attribute length because it needs some room for the preceding fields.
char long_data[512]{};
- ByteArray service_id_hash{kServiceIDHashBytes};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
ByteArray bad_data{long_data, 512};
BleAdvertisement ble_advertisement{kVersion, kSocketVersion, service_id_hash,
@@ -117,8 +117,8 @@ TEST(BleAdvertisementTest, ConstructionFailsWithLongData) {
}
TEST(BleAdvertisementTest, ConstructionFromSerializedBytesWorks) {
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
+ ByteArray data{std::string(kData)};
BleAdvertisement org_ble_advertisement{kVersion, kSocketVersion,
service_id_hash, data};
@@ -134,13 +134,10 @@ TEST(BleAdvertisementTest, ConstructionFromSerializedBytesWorks) {
}
TEST(BleAdvertisementTest, ConstructionFromSerializedBytesWithEmptyDataWorks) {
- char empty_data[0]{};
-
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{empty_data};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
BleAdvertisement org_ble_advertisement{kVersion, kSocketVersion,
- service_id_hash, data};
+ service_id_hash, ByteArray()};
ByteArray ble_advertisement_bytes{org_ble_advertisement};
BleAdvertisement ble_advertisement{ble_advertisement_bytes};
@@ -148,13 +145,12 @@ TEST(BleAdvertisementTest, ConstructionFromSerializedBytesWithEmptyDataWorks) {
EXPECT_EQ(kVersion, ble_advertisement.GetVersion());
EXPECT_EQ(kSocketVersion, ble_advertisement.GetSocketVersion());
EXPECT_EQ(service_id_hash, ble_advertisement.GetServiceIdHash());
- EXPECT_EQ(data.size(), ble_advertisement.GetData().size());
- EXPECT_EQ(data, ble_advertisement.GetData());
+ EXPECT_TRUE(ble_advertisement.GetData().Empty());
}
TEST(BleAdvertisementTest, ConstructionFromExtraSerializedBytesWorks) {
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
+ ByteArray data{std::string(kData)};
BleAdvertisement org_ble_advertisement{kVersion, kSocketVersion,
service_id_hash, data};
@@ -187,8 +183,8 @@ TEST(BleAdvertisementTest, ConstructionFromNullBytesFails) {
}
TEST(BleAdvertisementTest, ConstructionFromShortLengthSerializedBytesFails) {
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
+ ByteArray data{std::string(kData)};
BleAdvertisement org_ble_advertisement{kVersion, kSocketVersion,
service_id_hash, data};
@@ -204,8 +200,8 @@ TEST(BleAdvertisementTest, ConstructionFromShortLengthSerializedBytesFails) {
TEST(BleAdvertisementTest,
ConstructionFromSerializedBytesWithInvalidDataLengthFails) {
- ByteArray service_id_hash{kServiceIDHashBytes};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHashBytes)};
+ ByteArray data{std::string(kData)};
BleAdvertisement org_ble_advertisement{kVersion, kSocketVersion,
service_id_hash, data};
diff --git a/cpp/core_v2/internal/mediums/ble_packet.cc b/cpp/core_v2/internal/mediums/ble_packet.cc
index fce1a855..056e2097 100644
--- a/cpp/core_v2/internal/mediums/ble_packet.cc
+++ b/cpp/core_v2/internal/mediums/ble_packet.cc
@@ -14,7 +14,9 @@
#include "core_v2/internal/mediums/ble_packet.h"
+#include "platform_v2/base/base_input_stream.h"
#include "platform_v2/public/logging.h"
+#include "absl/strings/str_cat.h"
namespace location {
namespace nearby {
@@ -44,13 +46,14 @@ BlePacket::BlePacket(const ByteArray& ble_packet_bytes) {
return;
}
- const char *ble_packet_bytes_read_ptr = ble_packet_bytes.data();
- service_id_hash_ =
- ByteArray(ble_packet_bytes_read_ptr, kServiceIdHashLength);
- ble_packet_bytes_read_ptr += kServiceIdHashLength;
+ ByteArray packet_bytes{ble_packet_bytes};
+ BaseInputStream base_input_stream{packet_bytes};
+ // The first 3 bytes are supposed to be the service_id_hash.
+ service_id_hash_ = base_input_stream.ReadBytes(kServiceIdHashLength);
- data_ = ByteArray(ble_packet_bytes_read_ptr,
- ble_packet_bytes.size() - kServiceIdHashLength);
+ // The rest bytes are supposed to be the data.
+ data_ = base_input_stream.ReadBytes(ble_packet_bytes.size() -
+ kServiceIdHashLength);
}
BlePacket::operator ByteArray() const {
@@ -58,11 +61,8 @@ BlePacket::operator ByteArray() const {
return ByteArray();
}
- std::string out;
-
- out.reserve(service_id_hash_.size() + data_.size());
- out.append(std::string(service_id_hash_));
- out.append(std::string(data_));
+ std::string out =
+ absl::StrCat(std::string(service_id_hash_), std::string(data_));
return ByteArray(std::move(out));
}
diff --git a/cpp/core_v2/internal/mediums/ble_packet_test.cc b/cpp/core_v2/internal/mediums/ble_packet_test.cc
index 514400cf..8e23c6b6 100644
--- a/cpp/core_v2/internal/mediums/ble_packet_test.cc
+++ b/cpp/core_v2/internal/mediums/ble_packet_test.cc
@@ -21,12 +21,12 @@ namespace nearby {
namespace connections {
namespace mediums {
-constexpr char kServiceIDHash[] = "\x0a\x0b\x0c";
-constexpr char kData[] = "\x01\x02\x03\x04\x05";
+constexpr absl::string_view kServiceIDHash{"\x0a\x0b\x0c"};
+constexpr absl::string_view kData{"\x01\x02\x03\x04\x05"};
TEST(BlePacketTest, ConstructionWorks) {
- ByteArray service_id_hash{kServiceIDHash};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHash)};
+ ByteArray data{std::string(kData)};
BlePacket ble_packet{service_id_hash, data};
@@ -38,7 +38,7 @@ TEST(BlePacketTest, ConstructionWorks) {
TEST(BlePacketTest, ConstructionWorksWithEmptyData) {
char empty_data[] = "";
- ByteArray service_id_hash{kServiceIDHash};
+ ByteArray service_id_hash{std::string(kServiceIDHash)};
ByteArray data{empty_data};
BlePacket ble_packet{service_id_hash, data};
@@ -52,7 +52,7 @@ TEST(BlePacketTest, ConstructionFailsWithShortServiceIdHash) {
char short_service_id_hash[] = "\x0a\x0b";
ByteArray service_id_hash{short_service_id_hash};
- ByteArray data{kData};
+ ByteArray data{std::string(kData)};
BlePacket ble_packet(service_id_hash, data);
@@ -63,7 +63,7 @@ TEST(BlePacketTest, ConstructionFailsWithLongServiceIdHash) {
char long_service_id_hash[] = "\x0a\x0b\x0c\x0d";
ByteArray service_id_hash{long_service_id_hash};
- ByteArray data{kData};
+ ByteArray data{std::string(kData)};
BlePacket ble_packet{service_id_hash, data};
@@ -71,8 +71,8 @@ TEST(BlePacketTest, ConstructionFailsWithLongServiceIdHash) {
}
TEST(BlePacketTest, ConstructionFromSerializedBytesWorks) {
- ByteArray service_id_hash{kServiceIDHash};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHash)};
+ ByteArray data{std::string(kData)};
BlePacket org_ble_packet{service_id_hash, data};
ByteArray ble_packet_bytes{org_ble_packet};
@@ -91,8 +91,8 @@ TEST(BlePacketTest, ConstructionFromNullBytesFails) {
}
TEST(BlePacketTest, ConstructionFromShortLengthDataFails) {
- ByteArray service_id_hash{kServiceIDHash};
- ByteArray data{kData};
+ ByteArray service_id_hash{std::string(kServiceIDHash)};
+ ByteArray data{std::string(kData)};
BlePacket org_ble_packet{service_id_hash, data};
ByteArray org_ble_packet_bytes{org_ble_packet};
diff --git a/cpp/core_v2/internal/mediums/ble_peripheral_test.cc b/cpp/core_v2/internal/mediums/ble_peripheral_test.cc
index 66a656d8..d6f1f7f0 100644
--- a/cpp/core_v2/internal/mediums/ble_peripheral_test.cc
+++ b/cpp/core_v2/internal/mediums/ble_peripheral_test.cc
@@ -22,10 +22,10 @@ namespace connections {
namespace mediums {
namespace {
-const char kId[] = "AB12";
+constexpr absl::string_view kId{"AB12"};
TEST(BlePeripheralTest, ConstructionWorks) {
- ByteArray id{kId};
+ ByteArray id{std::string(kId)};
BlePeripheral ble_peripheral{id};
diff --git a/cpp/core_v2/internal/mediums/bloom_filter_test.cc b/cpp/core_v2/internal/mediums/bloom_filter_test.cc
index f81939b2..44467bc1 100644
--- a/cpp/core_v2/internal/mediums/bloom_filter_test.cc
+++ b/cpp/core_v2/internal/mediums/bloom_filter_test.cc
@@ -24,7 +24,7 @@ namespace connections {
namespace mediums {
namespace {
-const size_t kByteArrayLength = 100;
+constexpr size_t kByteArrayLength = 100;
TEST(BloomFilterTest, EmptyFilterReturnsEmptyArray) {
BloomFilter bloom_filter;
diff --git a/cpp/core_v2/internal/mediums/bluetooth_classic_test.cc b/cpp/core_v2/internal/mediums/bluetooth_classic_test.cc
index 8796d277..8204de36 100644
--- a/cpp/core_v2/internal/mediums/bluetooth_classic_test.cc
+++ b/cpp/core_v2/internal/mediums/bluetooth_classic_test.cc
@@ -38,6 +38,7 @@ class BluetoothClassicTest : public ::testing::Test {
using DiscoveryCallback = BluetoothClassicMedium::DiscoveryCallback;
BluetoothClassicTest() {
+ env_.Start();
env_.Reset();
radio_a_ = std::make_unique();
radio_b_ = std::make_unique();
@@ -60,6 +61,7 @@ class BluetoothClassicTest : public ::testing::Test {
radio_a_.reset();
radio_b_.reset();
env_.Reset();
+ env_.Stop();
}
MediumEnvironment& env_{MediumEnvironment::Instance()};
diff --git a/cpp/core_v2/internal/mediums/mediums.cc b/cpp/core_v2/internal/mediums/mediums.cc
index 51661834..8ea5036e 100644
--- a/cpp/core_v2/internal/mediums/mediums.cc
+++ b/cpp/core_v2/internal/mediums/mediums.cc
@@ -26,6 +26,10 @@ BluetoothClassic& Mediums::GetBluetoothClassic() {
return bluetooth_classic_;
}
+WifiLan& Mediums::GetWifiLan() {
+ return wifi_lan_;
+}
+
} // namespace connections
} // namespace nearby
} // namespace location
diff --git a/cpp/core_v2/internal/mediums/mediums.h b/cpp/core_v2/internal/mediums/mediums.h
index 1a41f28b..9018ce02 100644
--- a/cpp/core_v2/internal/mediums/mediums.h
+++ b/cpp/core_v2/internal/mediums/mediums.h
@@ -17,6 +17,8 @@
#include "core_v2/internal/mediums/bluetooth_classic.h"
#include "core_v2/internal/mediums/bluetooth_radio.h"
+#include "core_v2/internal/mediums/wifi_lan.h"
+
namespace location {
namespace nearby {
@@ -34,6 +36,9 @@ class Mediums {
// Returns a handle to the Bluetooth Classic medium.
BluetoothClassic& GetBluetoothClassic();
+ // Returns a handle to the Wifi-Lan medium.
+ WifiLan& GetWifiLan();
+
private:
// The order of declaration is critical for both construction and
// destruction.
@@ -45,6 +50,7 @@ class Mediums {
// corresponding radio.
BluetoothRadio bluetooth_radio_;
BluetoothClassic bluetooth_classic_{bluetooth_radio_};
+ WifiLan wifi_lan_;
};
} // namespace connections
diff --git a/cpp/core_v2/internal/mediums/uuid_test.cc b/cpp/core_v2/internal/mediums/uuid_test.cc
index 993311e9..acf691b2 100644
--- a/cpp/core_v2/internal/mediums/uuid_test.cc
+++ b/cpp/core_v2/internal/mediums/uuid_test.cc
@@ -24,7 +24,7 @@ namespace nearby {
namespace connections {
namespace {
-constexpr char kString[] = "some string";
+constexpr absl::string_view kString{"some string"};
constexpr std::uint64_t kNum1 = 0x123456789abcdef0;
constexpr std::uint64_t kNum2 = 0x21436587a9cbed0f;
diff --git a/cpp/core_v2/internal/mediums/webrtc.cc b/cpp/core_v2/internal/mediums/webrtc.cc
new file mode 100644
index 00000000..94eb891d
--- /dev/null
+++ b/cpp/core_v2/internal/mediums/webrtc.cc
@@ -0,0 +1,462 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "core_v2/internal/mediums/webrtc.h"
+
+#include
+#include
+
+#include "core_v2/internal/mediums/webrtc/session_description_wrapper.h"
+#include "core_v2/internal/mediums/webrtc/signaling_frames.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/base/listeners.h"
+#include "platform_v2/public/future.h"
+#include "platform_v2/public/logging.h"
+#include "platform_v2/public/mutex_lock.h"
+#include "location/nearby/mediums/proto/web_rtc_signaling_frames.pb.h"
+#include "absl/strings/str_cat.h"
+#include "webrtc/api/jsep.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+namespace {
+
+// The maximum amount of time to wait to connect to a data channel via WebRTC.
+// TODO(himanshujaju): Should this be configurable per platform?
+constexpr absl::Duration kDataChannelTimeout = absl::Milliseconds(5000);
+
+} // namespace
+
+WebRtc::WebRtc() = default;
+
+WebRtc::~WebRtc() {
+ single_thread_executor_.Shutdown();
+ {
+ MutexLock lock(&mutex_);
+ Disconnect();
+ }
+}
+
+bool WebRtc::IsAvailable() { return medium_.IsValid(); }
+
+bool WebRtc::IsAcceptingConnections() {
+ MutexLock lock(&mutex_);
+ return role_ == Role::kOfferer;
+}
+
+bool WebRtc::StartAcceptingConnections(const PeerId& self_id,
+ AcceptedConnectionCallback callback) {
+ if (!IsAvailable()) {
+ {
+ MutexLock lock(&mutex_);
+ LogAndDisconnect("WebRTC is not available for data transfer.");
+ }
+ return false;
+ }
+
+ if (IsAcceptingConnections()) {
+ NEARBY_LOG(WARNING, "Already accepting WebRTC connections.");
+ return false;
+ }
+
+ {
+ MutexLock lock(&mutex_);
+ if (role_ != Role::kNone) {
+ NEARBY_LOG(WARNING,
+ "Cannot start accepting WebRTC connections, current role %d",
+ role_);
+ return false;
+ }
+
+ if (!InitWebRtcFlow(Role::kOfferer, self_id)) return false;
+
+ SessionDescriptionWrapper offer = connection_flow_->CreateOffer();
+ pending_local_offer_ = webrtc_frames::EncodeOffer(self_id, offer.GetSdp());
+ if (!SetLocalSessionDescription(std::move(offer))) {
+ return false;
+ }
+
+ // There is no timeout set for the future returned since we do not know how
+ // much time it will take for the two devices to discover each other before
+ // the actual transport can begin.
+ ListenForWebRtcSocketFuture(connection_flow_->GetDataChannel(),
+ std::move(callback));
+ NEARBY_LOG(INFO, "Started listening for WebRtc connections as %s",
+ self_id.GetId().c_str());
+ }
+
+ return true;
+}
+
+WebRtcSocketWrapper WebRtc::Connect(const PeerId& peer_id) {
+ MutexLock lock(&mutex_);
+
+ if (!IsAvailable()) {
+ Disconnect();
+ return WebRtcSocketWrapper();
+ }
+
+ if (role_ != Role::kNone) {
+ NEARBY_LOG(WARNING,
+ "Cannot connect with WebRtc because we are already acting as %d",
+ role_);
+ return WebRtcSocketWrapper();
+ }
+
+ peer_id_ = peer_id;
+ if (!InitWebRtcFlow(Role::kAnswerer, PeerId::FromRandom())) {
+ return WebRtcSocketWrapper();
+ }
+
+ NEARBY_LOG(INFO, "Attempting to make a WebRTC connection to %s.",
+ peer_id.GetId().c_str());
+
+ std::shared_ptr> socket_future =
+ ListenForWebRtcSocketFuture(connection_flow_->GetDataChannel(),
+ AcceptedConnectionCallback());
+
+ // The two devices have discovered each other, hence we have a timeout for
+ // establishing the transport channel.
+ ExceptionOr result =
+ socket_future->Get(kDataChannelTimeout);
+ if (result.ok()) return result.result();
+
+ Disconnect();
+ return WebRtcSocketWrapper();
+}
+
+bool WebRtc::SetLocalSessionDescription(SessionDescriptionWrapper sdp) {
+ if (!connection_flow_->SetLocalSessionDescription(std::move(sdp))) {
+ LogAndDisconnect("Unable to set local session description");
+ return false;
+ }
+
+ return true;
+}
+
+void WebRtc::StopAcceptingConnections() {
+ if (!IsAcceptingConnections()) {
+ NEARBY_LOG(INFO,
+ "Skipped StopAcceptingConnections since we are not currently "
+ "accepting WebRTC connections");
+ return;
+ }
+
+ {
+ MutexLock lock(&mutex_);
+ ShutdownSignaling();
+ }
+ NEARBY_LOG(INFO, "Stopped accepting WebRTC connections");
+}
+
+std::shared_ptr>
+WebRtc::ListenForWebRtcSocketFuture(
+ Future>*
+ data_channel_future,
+ AcceptedConnectionCallback callback) {
+ auto socket_future = std::make_shared>();
+ auto data_channel_runnable = [this, socket_future, data_channel_future,
+ callback{std::move(callback)}]() {
+ // The overall timeout of creating the socket and data channel is controlled
+ // by the caller of this function.
+ ExceptionOr> res =
+ data_channel_future->Get();
+ if (res.ok()) {
+ WebRtcSocketWrapper wrapper = CreateWebRtcSocketWrapper(res.result());
+ callback.accepted_cb(wrapper);
+ {
+ MutexLock lock(&mutex_);
+ socket_ = wrapper;
+ }
+ socket_future->Set(wrapper);
+ } else {
+ NEARBY_LOG(WARNING, "Failed to get WebRtcSocket.");
+ socket_future->Set(WebRtcSocketWrapper());
+ }
+ };
+
+ data_channel_future->AddListener(std::move(data_channel_runnable),
+ &single_thread_executor_);
+
+ return socket_future;
+}
+
+WebRtcSocketWrapper WebRtc::CreateWebRtcSocketWrapper(
+ rtc::scoped_refptr data_channel) {
+ if (data_channel == nullptr) {
+ return WebRtcSocketWrapper();
+ }
+
+ auto socket = std::make_unique("WebRtcSocket", data_channel);
+ socket->SetOnSocketClosedListener({std::bind(&WebRtc::Disconnect, this)});
+ return WebRtcSocketWrapper(std::move(socket));
+}
+
+bool WebRtc::InitWebRtcFlow(Role role, const PeerId& self_id) {
+ role_ = role;
+ self_id_ = self_id;
+
+ if (connection_flow_) {
+ LogAndShutdownSignaling(
+ "Tried to initialize WebRTC without shutting down the previous "
+ "connection");
+ return false;
+ }
+
+ if (signaling_messenger_) {
+ LogAndShutdownSignaling(
+ "Tried to initialize WebRTC without shutting down signaling messenger");
+ return false;
+ }
+
+ signaling_messenger_ = medium_.GetSignalingMessenger(self_id_.GetId());
+ auto signaling_message_callback = [this](ByteArray message) {
+ OffloadFromSignalingThread([this, message{std::move(message)}]() {
+ ProcessSignalingMessage(message);
+ });
+ };
+
+ if (!signaling_messenger_->IsValid() ||
+ !signaling_messenger_->StartReceivingMessages(
+ signaling_message_callback)) {
+ Disconnect();
+ return false;
+ }
+
+ if (role_ == Role::kAnswerer &&
+ !signaling_messenger_->SendMessage(
+ peer_id_.GetId(),
+ webrtc_frames::EncodeReadyForSignalingPoke(self_id))) {
+ LogAndDisconnect(absl::StrCat("Could not send signaling poke to peer ",
+ peer_id_.GetId()));
+ return false;
+ }
+
+ connection_flow_ = ConnectionFlow::Create(GetLocalIceCandidateListener(),
+ GetDataChannelListener(), medium_);
+ return true;
+}
+
+void WebRtc::OnLocalIceCandidate(
+ const webrtc::IceCandidateInterface* local_ice_candidate) {
+ ::location::nearby::mediums::IceCandidate ice_candidate =
+ webrtc_frames::EncodeIceCandidate(*local_ice_candidate);
+
+ OffloadFromSignalingThread([this, ice_candidate{std::move(ice_candidate)}]() {
+ MutexLock lock(&mutex_);
+ if (IsSignaling()) {
+ signaling_messenger_->SendMessage(
+ peer_id_.GetId(), webrtc_frames::EncodeIceCandidates(
+ self_id_, {std::move(ice_candidate)}));
+ } else {
+ pending_local_ice_candidates_.push_back(std::move(ice_candidate));
+ }
+ });
+}
+
+LocalIceCandidateListener WebRtc::GetLocalIceCandidateListener() {
+ return {std::bind(&WebRtc::OnLocalIceCandidate, this, std::placeholders::_1)};
+}
+
+void WebRtc::OnDataChannelClosed() {
+ OffloadFromSignalingThread([this]() {
+ MutexLock lock(&mutex_);
+ LogAndDisconnect("WebRTC data channel closed");
+ });
+}
+
+void WebRtc::OnDataChannelMessageReceived(const ByteArray& message) {
+ OffloadFromSignalingThread([this, message]() {
+ MutexLock lock(&mutex_);
+ if (!socket_.IsValid()) {
+ LogAndDisconnect("Received a data channel message without a socket");
+ return;
+ }
+
+ socket_.NotifyDataChannelMsgReceived(message);
+ });
+}
+
+void WebRtc::OnDataChannelBufferedAmountChanged() {
+ OffloadFromSignalingThread([this]() {
+ MutexLock lock(&mutex_);
+ if (!socket_.IsValid()) {
+ LogAndDisconnect("Data channel buffer changed without a socket");
+ return;
+ }
+
+ socket_.NotifyDataChannelBufferedAmountChanged();
+ });
+}
+
+DataChannelListener WebRtc::GetDataChannelListener() {
+ return {
+ .data_channel_closed_cb = std::bind(&WebRtc::OnDataChannelClosed, this),
+ .data_channel_message_received_cb = std::bind(
+ &WebRtc::OnDataChannelMessageReceived, this, std::placeholders::_1),
+ .data_channel_buffered_amount_changed_cb =
+ std::bind(&WebRtc::OnDataChannelBufferedAmountChanged, this),
+ };
+}
+
+bool WebRtc::IsSignaling() {
+ return (role_ != Role::kNone && self_id_.IsValid() && peer_id_.IsValid());
+}
+
+void WebRtc::ProcessSignalingMessage(const ByteArray& message) {
+ MutexLock lock(&mutex_);
+
+ if (!connection_flow_) {
+ LogAndDisconnect("Received WebRTC frame before signaling was started");
+ return;
+ }
+
+ location::nearby::mediums::WebRtcSignalingFrame frame;
+ if (!frame.ParseFromString(std::string(message))) {
+ LogAndDisconnect("Failed to parse signaling message");
+ return;
+ }
+
+ if (!frame.has_sender_id()) {
+ LogAndDisconnect("Invalid WebRTC frame: Sender ID is missing");
+ return;
+ }
+
+ if (frame.has_ready_for_signaling_poke() && !peer_id_.IsValid()) {
+ peer_id_ = PeerId(frame.sender_id().id());
+ NEARBY_LOG(INFO, "Peer %s is ready for signaling",
+ peer_id_.GetId().c_str());
+ }
+
+ if (!IsSignaling()) {
+ NEARBY_LOG(INFO,
+ "Ignoring WebRTC frame: we are not currently listening for "
+ "signaling messages");
+ return;
+ }
+
+ if (frame.sender_id().id() != peer_id_.GetId()) {
+ NEARBY_LOG(
+ INFO, "Ignoring WebRTC frame: we are only listening for another peer.");
+ return;
+ }
+
+ if (frame.has_ready_for_signaling_poke()) {
+ SendOfferAndIceCandidatesToPeer();
+ } else if (frame.has_offer()) {
+ connection_flow_->OnOfferReceived(
+ SessionDescriptionWrapper(webrtc_frames::DecodeOffer(frame).release()));
+ SendAnswerToPeer();
+ } else if (frame.has_answer()) {
+ connection_flow_->OnAnswerReceived(SessionDescriptionWrapper(
+ webrtc_frames::DecodeAnswer(frame).release()));
+ } else if (frame.has_ice_candidates()) {
+ if (!connection_flow_->OnRemoteIceCandidatesReceived(
+ webrtc_frames::DecodeIceCandidates(frame))) {
+ LogAndDisconnect("Could not add remote ice candidates.");
+ }
+ }
+}
+
+void WebRtc::SendOfferAndIceCandidatesToPeer() {
+ if (pending_local_offer_.Empty()) {
+ LogAndDisconnect(
+ "Unable to send pending offer to remote peer: local offer not set");
+ return;
+ }
+
+ if (!signaling_messenger_->SendMessage(peer_id_.GetId(),
+ pending_local_offer_)) {
+ LogAndDisconnect("Failed to send local offer via signaling messenger");
+ return;
+ }
+ pending_local_offer_ = ByteArray();
+
+ if (!pending_local_ice_candidates_.empty()) {
+ signaling_messenger_->SendMessage(
+ peer_id_.GetId(),
+ webrtc_frames::EncodeIceCandidates(
+ self_id_, std::move(pending_local_ice_candidates_)));
+ }
+}
+
+void WebRtc::SendAnswerToPeer() {
+ SessionDescriptionWrapper answer = connection_flow_->CreateAnswer();
+ ByteArray answer_message(
+ webrtc_frames::EncodeAnswer(self_id_, answer.GetSdp()));
+
+ if (!SetLocalSessionDescription(std::move(answer))) return;
+
+ if (!signaling_messenger_->SendMessage(peer_id_.GetId(), answer_message)) {
+ LogAndDisconnect("Failed to send local answer via signaling messenger");
+ return;
+ }
+}
+
+void WebRtc::LogAndDisconnect(const std::string& error_message) {
+ NEARBY_LOG(WARNING, "Disconnecting WebRTC : %s", error_message.c_str());
+ Disconnect();
+}
+
+void WebRtc::LogAndShutdownSignaling(const std::string& error_message) {
+ NEARBY_LOG(WARNING, "Stopping WebRTC signaling : %s", error_message.c_str());
+ ShutdownSignaling();
+}
+
+void WebRtc::ShutdownSignaling() {
+ role_ = Role::kNone;
+ self_id_ = PeerId();
+ peer_id_ = PeerId();
+ pending_local_offer_ = ByteArray();
+ pending_local_ice_candidates_.clear();
+
+ if (signaling_messenger_) {
+ signaling_messenger_->StopReceivingMessages();
+ signaling_messenger_.reset();
+ }
+
+ if (!socket_.IsValid()) ShutdownIceCandidateCollection();
+}
+
+void WebRtc::Disconnect() {
+ ShutdownSignaling();
+ ShutdownWebRtcSocket();
+ ShutdownIceCandidateCollection();
+}
+
+void WebRtc::ShutdownWebRtcSocket() {
+ if (socket_.IsValid()) {
+ socket_.Close();
+ socket_ = WebRtcSocketWrapper();
+ }
+}
+
+void WebRtc::ShutdownIceCandidateCollection() {
+ if (connection_flow_) {
+ connection_flow_->Close();
+ connection_flow_.reset();
+ }
+}
+
+void WebRtc::OffloadFromSignalingThread(Runnable runnable) {
+ single_thread_executor_.Execute(std::move(runnable));
+}
+
+} // namespace mediums
+} // namespace connections
+} // namespace nearby
+} // namespace location
diff --git a/cpp/core_v2/internal/mediums/webrtc.h b/cpp/core_v2/internal/mediums/webrtc.h
new file mode 100644
index 00000000..6a4e7d7d
--- /dev/null
+++ b/cpp/core_v2/internal/mediums/webrtc.h
@@ -0,0 +1,169 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_
+
+#include
+#include
+
+#include "core_v2/internal/mediums/webrtc/connection_flow.h"
+#include "core_v2/internal/mediums/webrtc/data_channel_listener.h"
+#include "core_v2/internal/mediums/webrtc/local_ice_candidate_listener.h"
+#include "core_v2/internal/mediums/webrtc/peer_id.h"
+#include "core_v2/internal/mediums/webrtc/webrtc_socket.h"
+#include "core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/base/listeners.h"
+#include "platform_v2/base/runnable.h"
+#include "platform_v2/public/future.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/single_thread_executor.h"
+#include "platform_v2/public/webrtc.h"
+#include "location/nearby/mediums/proto/web_rtc_signaling_frames.pb.h"
+#include "webrtc/api/data_channel_interface.h"
+#include "webrtc/api/jsep.h"
+#include "webrtc/api/scoped_refptr.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+// Callback that is invoked when a new connection is accepted.
+struct AcceptedConnectionCallback {
+ std::function accepted_cb =
+ DefaultCallback();
+};
+
+// Entry point for connecting a data channel between two devices via WebRtc.
+class WebRtc {
+ public:
+ WebRtc();
+ ~WebRtc();
+
+ // Returns if WebRtc is available as a medium for nearby to transport data.
+ // Runs on @MainThread.
+ bool IsAvailable();
+
+ // Returns if the device is ready to accept connections from remote devices.
+ // Runs on @MainThread.
+ bool IsAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ // Prepares the device to accept incoming WebRtc connections. Returns a
+ // boolean value indicating if the device has started accepting connections.
+ // Runs on @MainThread.
+ bool StartAcceptingConnections(const PeerId& self_id,
+ AcceptedConnectionCallback callback)
+ ABSL_LOCKS_EXCLUDED(mutex_);
+
+ // Prevents device from accepting future connections until
+ // StartAcceptingConnections() is called.
+ // Runs on @MainThread.
+ void StopAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ // Initiates a WebRtc connection with peer device identified by |peer_id|.
+ // Runs on @MainThread.
+ WebRtcSocketWrapper Connect(const PeerId& peer_id)
+ ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+ enum class Role {
+ kNone = 0,
+ kOfferer = 1,
+ kAnswerer = 2,
+ };
+
+ bool InitWebRtcFlow(Role role, const PeerId& self_id)
+ ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ std::shared_ptr> ListenForWebRtcSocketFuture(
+ Future>*
+ data_channel_future,
+ AcceptedConnectionCallback callback);
+
+ WebRtcSocketWrapper CreateWebRtcSocketWrapper(
+ rtc::scoped_refptr data_channel);
+
+ LocalIceCandidateListener GetLocalIceCandidateListener();
+ void OnLocalIceCandidate(
+ const webrtc::IceCandidateInterface* local_ice_candidate);
+
+ DataChannelListener GetDataChannelListener();
+ void OnDataChannelClosed();
+ void OnDataChannelMessageReceived(const ByteArray& message);
+ void OnDataChannelBufferedAmountChanged();
+
+ // Runs on @MainThread and |single_thread_executor_|.
+ bool SetLocalSessionDescription(SessionDescriptionWrapper sdp)
+ ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on |single_thread_executor_|.
+ bool IsSignaling() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on |single_thread_executor_|.
+ void ProcessSignalingMessage(const ByteArray& message)
+ ABSL_LOCKS_EXCLUDED(mutex_);
+
+ // Runs on |single_thread_executor_|.
+ void SendOfferAndIceCandidatesToPeer() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on |single_thread_executor_|.
+ void SendAnswerToPeer() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on @MainThread and |single_thread_executor_|.
+ void LogAndDisconnect(const std::string& error_message)
+ ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on @MainThread and |single_thread_executor_|.
+ void Disconnect() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ void LogAndShutdownSignaling(const std::string& error_message)
+ ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on @MainThread and |single_thread_executor_|.
+ void ShutdownSignaling() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on @MainThread and |single_thread_executor_|.
+ void ShutdownWebRtcSocket() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ // Runs on @MainThread and |single_thread_executor_|.
+ void ShutdownIceCandidateCollection();
+
+ void OffloadFromSignalingThread(Runnable runnable);
+
+ Mutex mutex_;
+
+ Role role_ ABSL_GUARDED_BY(mutex_) = Role::kNone;
+ PeerId self_id_ ABSL_GUARDED_BY(mutex_);
+ PeerId peer_id_ ABSL_GUARDED_BY(mutex_);
+ ByteArray pending_local_offer_ ABSL_GUARDED_BY(mutex_);
+ std::vector<::location::nearby::mediums::IceCandidate>
+ pending_local_ice_candidates_ ABSL_GUARDED_BY(mutex_);
+
+ std::unique_ptr connection_flow_;
+ std::unique_ptr signaling_messenger_
+ ABSL_GUARDED_BY(mutex_);
+ WebRtcSocketWrapper socket_ ABSL_GUARDED_BY(mutex_);
+ WebRtcMedium medium_;
+
+ SingleThreadExecutor single_thread_executor_;
+};
+
+} // namespace mediums
+} // namespace connections
+} // namespace nearby
+} // namespace location
+
+#endif // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_
diff --git a/cpp/core_v2/internal/mediums/webrtc/BUILD b/cpp/core_v2/internal/mediums/webrtc/BUILD
index 61e96c60..699da15f 100644
--- a/cpp/core_v2/internal/mediums/webrtc/BUILD
+++ b/cpp/core_v2/internal/mediums/webrtc/BUILD
@@ -16,23 +16,38 @@ cc_library(
name = "webrtc",
srcs = [
"connection_flow.cc",
+ "data_channel_observer_impl.cc",
"peer_connection_observer_impl.cc",
+ "peer_id.cc",
+ "signaling_frames.cc",
"webrtc_socket.cc",
],
hdrs = [
"connection_flow.h",
"data_channel_listener.h",
+ "data_channel_observer_impl.h",
"local_ice_candidate_listener.h",
"peer_connection_observer_impl.h",
+ "peer_id.h",
+ "session_description_wrapper.h",
+ "signaling_frames.h",
"webrtc_socket.h",
+ "webrtc_socket_wrapper.h",
+ ],
+ visibility = [
+ "//core_v2/internal:__subpackages__",
],
deps = [
"//core_v2:core_types",
+ "//core_v2/internal/mediums:utils",
"//platform_v2/base",
"//platform_v2/public:comm",
"//platform_v2/public:logging",
"//platform_v2/public:types",
+ "//location/nearby/mediums/proto:web_rtc_signaling_frames_cc_proto",
"//absl/memory",
+ "//absl/strings",
+ "//absl/time",
"//webrtc/api:libjingle_peerconnection_api",
],
)
@@ -41,6 +56,8 @@ cc_test(
name = "webrtc_test",
srcs = [
"connection_flow_test.cc",
+ "peer_id_test.cc",
+ "signaling_frames_test.cc",
"webrtc_socket_test.cc",
],
deps = [
@@ -48,56 +65,12 @@ cc_test(
"//platform_v2/base",
"//platform_v2/impl/g3", # buildcleaner: keep
"//platform_v2/public:comm",
- "//testing/base/public:gunit_main",
- "//webrtc/api:libjingle_peerconnection_api",
- ],
-)
-
-cc_test(
- name = "peer_id_test",
- srcs = ["peer_id_test.cc"],
- deps = [
- ":peer_id",
- "//platform_v2/base",
- "//platform_v2/impl/g3", #buildcleaner: keep
- "//platform_v2/public:comm",
"//platform_v2/public:types",
- "//testing/base/public:gunit_main",
- ],
-)
-
-cc_test(
- name = "signaling_frames_test",
- srcs = ["signaling_frames_test.cc"],
- deps = [
- ":peer_id",
- ":signaling_frames",
- "//platform_v2/impl/g3", # buildcleaner: keep
"//net/proto2/public:proto2",
"//testing/base/public:gunit_main",
- "//webrtc/pc:peerconnection", # buildcleaner: keep
- ],
-)
-
-cc_library(
- name = "peer_id",
- srcs = ["peer_id.cc"],
- hdrs = ["peer_id.h"],
- deps = [
- "//core_v2/internal/mediums:utils",
- "//platform_v2/base",
- "//absl/strings",
- ],
-)
-
-cc_library(
- name = "signaling_frames",
- srcs = ["signaling_frames.cc"],
- hdrs = ["signaling_frames.h"],
- deps = [
- ":peer_id",
- "//platform_v2/base",
- "//location/nearby/mediums/proto:web_rtc_signaling_frames_cc_proto",
+ "//absl/time",
"//webrtc/api:libjingle_peerconnection_api",
+ "//webrtc/api:rtc_error",
+ "//webrtc/api:scoped_refptr",
],
)
diff --git a/cpp/core_v2/internal/mediums/webrtc/connection_flow.cc b/cpp/core_v2/internal/mediums/webrtc/connection_flow.cc
index c34ee3e9..1712b46d 100644
--- a/cpp/core_v2/internal/mediums/webrtc/connection_flow.cc
+++ b/cpp/core_v2/internal/mediums/webrtc/connection_flow.cc
@@ -14,25 +14,79 @@
#include "core_v2/internal/mediums/webrtc/connection_flow.h"
+#include
#include
+#include "core_v2/internal/mediums/webrtc/session_description_wrapper.h"
+#include "platform_v2/public/logging.h"
#include "platform_v2/public/mutex_lock.h"
#include "platform_v2/public/webrtc.h"
#include "absl/memory/memory.h"
+#include "absl/time/time.h"
+#include "webrtc/api/data_channel_interface.h"
+#include "webrtc/api/jsep.h"
namespace location {
namespace nearby {
namespace connections {
namespace mediums {
+namespace {
+// This is the same as the nearby data channel name.
+const char kDataChannelName[] = "dataChannel";
+
+class CreateSessionDescriptionObserverImpl
+ : public webrtc::CreateSessionDescriptionObserver {
+ public:
+ explicit CreateSessionDescriptionObserverImpl(
+ Future* settable_future)
+ : settable_future_(settable_future) {}
+ ~CreateSessionDescriptionObserverImpl() override = default;
+
+ // webrtc::CreateSessionDescriptionObserver
+ void OnSuccess(webrtc::SessionDescriptionInterface* desc) override {
+ settable_future_->Set(SessionDescriptionWrapper{desc});
+ }
+
+ void OnFailure(webrtc::RTCError error) override {
+ NEARBY_LOG(ERROR, "Error when creating session description: %s",
+ error.message());
+ settable_future_->SetException({Exception::kFailed});
+ }
+
+ private:
+ std::unique_ptr> settable_future_;
+};
+
+class SetSessionDescriptionObserverImpl
+ : public webrtc::SetSessionDescriptionObserver {
+ public:
+ explicit SetSessionDescriptionObserverImpl(Future* settable_future)
+ : settable_future_(settable_future) {}
+
+ void OnSuccess() override { settable_future_->Set(true); }
+
+ void OnFailure(webrtc::RTCError error) override {
+ NEARBY_LOG(ERROR, "Error when setting session description: %s",
+ error.message());
+ settable_future_->SetException({Exception::kFailed});
+ }
+
+ private:
+ std::unique_ptr> settable_future_;
+};
+
+using PeerConnectionState =
+ webrtc::PeerConnectionInterface::PeerConnectionState;
+
+} // namespace
+
std::unique_ptr ConnectionFlow::Create(
LocalIceCandidateListener local_ice_candidate_listener,
- DataChannelListener data_channel_listener,
- SingleThreadExecutor* single_threaded_executor,
- WebRtcMedium& webrtc_medium) {
- auto connection_flow = absl::WrapUnique(new ConnectionFlow(
- std::move(local_ice_candidate_listener), std::move(data_channel_listener),
- single_threaded_executor));
+ DataChannelListener data_channel_listener, WebRtcMedium& webrtc_medium) {
+ auto connection_flow = absl::WrapUnique(
+ new ConnectionFlow(std::move(local_ice_candidate_listener),
+ std::move(data_channel_listener)));
if (connection_flow->InitPeerConnection(webrtc_medium)) {
return connection_flow;
}
@@ -42,75 +96,149 @@ std::unique_ptr ConnectionFlow::Create(
ConnectionFlow::ConnectionFlow(
LocalIceCandidateListener local_ice_candidate_listener,
- DataChannelListener data_channel_listener,
- SingleThreadExecutor* single_threaded_executor)
+ DataChannelListener data_channel_listener)
: data_channel_listener_(std::move(data_channel_listener)),
- peer_connection_observer_(this, std::move(local_ice_candidate_listener),
- single_threaded_executor) {}
-
-std::unique_ptr
-ConnectionFlow::CreateOffer() {
- MutexLock lock(&mutex_);
-
- // TODO(bfranz): Implement
-
- return std::unique_ptr();
+ peer_connection_observer_(this, std::move(local_ice_candidate_listener)) {
}
-std::unique_ptr
-ConnectionFlow::CreateAnswer() {
+ConnectionFlow::~ConnectionFlow() { Close(); }
+
+SessionDescriptionWrapper ConnectionFlow::CreateOffer() {
MutexLock lock(&mutex_);
- // TODO(bfranz): Implement
+ if (!TransitionState(State::kInitialized, State::kCreatingOffer)) {
+ return SessionDescriptionWrapper();
+ }
- return std::unique_ptr();
+ webrtc::DataChannelInit data_channel_init;
+ data_channel_init.reliable = true;
+ rtc::scoped_refptr data_channel =
+ peer_connection_->CreateDataChannel(kDataChannelName, &data_channel_init);
+ data_channel->RegisterObserver(CreateDataChannelObserver(data_channel));
+
+ auto success_future = new Future();
+ webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options;
+ rtc::scoped_refptr observer =
+ new rtc::RefCountedObject(
+ success_future);
+ peer_connection_->CreateOffer(observer, options);
+
+ ExceptionOr result = success_future->Get(kTimeout);
+ if (result.ok() &&
+ TransitionState(State::kCreatingOffer, State::kWaitingForAnswer)) {
+ return std::move(result.result());
+ }
+
+ return SessionDescriptionWrapper();
}
-bool ConnectionFlow::SetLocalSessionDescription(
- std::unique_ptr sdp) {
+SessionDescriptionWrapper ConnectionFlow::CreateAnswer() {
MutexLock lock(&mutex_);
- // TODO(bfranz): Implement
+ if (!TransitionState(State::kReceivedOffer, State::kCreatingAnswer)) {
+ return SessionDescriptionWrapper();
+ }
- return false;
+ auto success_future = new Future();
+ webrtc::PeerConnectionInterface::RTCOfferAnswerOptions options;
+ rtc::scoped_refptr observer =
+ new rtc::RefCountedObject(
+ success_future);
+ peer_connection_->CreateAnswer(observer, options);
+
+ ExceptionOr result = success_future->Get(kTimeout);
+ if (result.ok() &&
+ TransitionState(State::kCreatingAnswer, State::kWaitingToConnect)) {
+ return std::move(result.result());
+ }
+
+ return SessionDescriptionWrapper();
}
-void ConnectionFlow::OnOfferReceived(
- std::unique_ptr offer) {
+bool ConnectionFlow::SetLocalSessionDescription(SessionDescriptionWrapper sdp) {
MutexLock lock(&mutex_);
- // TODO(bfranz): Implement
+ if (!sdp.IsValid()) return false;
+
+ auto success_future = new Future();
+ rtc::scoped_refptr observer =
+ new rtc::RefCountedObject(
+ success_future);
+
+ peer_connection_->SetLocalDescription(observer, sdp.Release());
+
+ ExceptionOr result = success_future->Get(kTimeout);
+ return result.ok() && result.result();
}
-void ConnectionFlow::OnAnswerReceived(
- std::unique_ptr answer) {
+bool ConnectionFlow::SetRemoteSessionDescription(
+ SessionDescriptionWrapper sdp) {
+ if (!sdp.IsValid()) return false;
+
+ auto success_future = new Future();
+ rtc::scoped_refptr observer =
+ new rtc::RefCountedObject(
+ success_future);
+
+ peer_connection_->SetRemoteDescription(observer, sdp.Release());
+
+ ExceptionOr result = success_future->Get(kTimeout);
+ return result.ok() && result.result();
+}
+
+bool ConnectionFlow::OnOfferReceived(SessionDescriptionWrapper offer) {
MutexLock lock(&mutex_);
- // TODO(bfranz): Implement
+ if (!TransitionState(State::kInitialized, State::kReceivedOffer)) {
+ return false;
+ }
+ return SetRemoteSessionDescription(std::move(offer));
+}
+
+bool ConnectionFlow::OnAnswerReceived(SessionDescriptionWrapper answer) {
+ MutexLock lock(&mutex_);
+
+ if (!TransitionState(State::kWaitingForAnswer, State::kWaitingToConnect)) {
+ return false;
+ }
+ return SetRemoteSessionDescription(std::move(answer));
}
bool ConnectionFlow::OnRemoteIceCandidatesReceived(
- std::vector ice_candidates) {
+ std::vector>
+ ice_candidates) {
MutexLock lock(&mutex_);
- // TODO(bfranz): Implement
+ if (state_ == State::kEnded) {
+ NEARBY_LOG(WARNING,
+ "You cannot add ice candidates to a disconnected session.");
+ return false;
+ }
- return false;
+ if (state_ != State::kWaitingToConnect && state_ != State::kConnected) {
+ cached_remote_ice_candidates_.insert(
+ cached_remote_ice_candidates_.end(),
+ std::make_move_iterator(ice_candidates.begin()),
+ std::make_move_iterator(ice_candidates.end()));
+ return true;
+ }
+
+ for (auto&& ice_candidate : ice_candidates) {
+ if (!peer_connection_->AddIceCandidate(ice_candidate.get())) {
+ NEARBY_LOG(WARNING, "Unable to add remote ice candidate.");
+ }
+ }
+ return true;
}
-api::ListenableFuture>*
+Future>*
ConnectionFlow::GetDataChannel() {
- return static_cast<
- api::ListenableFuture>*>(
- &data_channel_future_);
+ return &data_channel_future_;
}
bool ConnectionFlow::Close() {
MutexLock lock(&mutex_);
-
- // TODO(bfranz): Implement
-
- return false;
+ return CloseLocked();
}
bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) {
@@ -128,20 +256,96 @@ bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) {
}
void ConnectionFlow::OnSignalingStable() {
- // TODO(bfranz): Implement
+ MutexLock lock(&mutex_);
+
+ if (state_ != State::kWaitingToConnect && state_ != State::kConnected) return;
+
+ for (auto&& ice_candidate : cached_remote_ice_candidates_) {
+ if (!peer_connection_->AddIceCandidate(ice_candidate.get())) {
+ NEARBY_LOG(WARNING, "Unable to add remote ice candidate.");
+ }
+ }
+ cached_remote_ice_candidates_.clear();
}
void ConnectionFlow::ProcessOnPeerConnectionChange(
webrtc::PeerConnectionInterface::PeerConnectionState new_state) {
- // TODO(bfranz): Implement
+ if (new_state == PeerConnectionState::kClosed ||
+ new_state == PeerConnectionState::kFailed ||
+ new_state == PeerConnectionState::kDisconnected) {
+ MutexLock lock(&mutex_);
+ CloseAndNotifyLocked();
+ }
+}
+
+void ConnectionFlow::ProcessDataChannelConnected() {
+ MutexLock lock(&mutex_);
+ NEARBY_LOG(INFO, "Data channel state changed to connected.");
+ if (!TransitionState(State::kWaitingToConnect, State::kConnected))
+ CloseAndNotifyLocked();
}
webrtc::DataChannelObserver* ConnectionFlow::CreateDataChannelObserver(
rtc::scoped_refptr data_channel) {
- // TODO(bfranz): Implement
+ if (!data_channel_observer_) {
+ auto state_change_callback = [this,
+ data_channel{std::move(data_channel)}]() {
+ if (data_channel->state() ==
+ webrtc::DataChannelInterface::DataState::kOpen) {
+ data_channel_future_.Set(std::move(data_channel));
+ OffloadFromSignalingThread([this]() { ProcessDataChannelConnected(); });
+ } else if (data_channel->state() ==
+ webrtc::DataChannelInterface::DataState::kClosed) {
+ data_channel->UnregisterObserver();
+ OffloadFromSignalingThread([this]() {
+ MutexLock lock(&mutex_);
+ CloseAndNotifyLocked();
+ });
+ }
+ };
+ data_channel_observer_ = absl::make_unique(
+ &data_channel_listener_, std::move(state_change_callback));
+ }
- return nullptr;
+ return reinterpret_cast(
+ data_channel_observer_.get());
}
+
+bool ConnectionFlow::TransitionState(State current_state, State new_state) {
+ if (current_state != state_) {
+ NEARBY_LOG(
+ WARNING,
+ "Invalid state transition to %d: current state is %d but expected %d.",
+ new_state, state_, current_state);
+ return false;
+ }
+ state_ = new_state;
+ return true;
+}
+
+void ConnectionFlow::CloseAndNotifyLocked() {
+ if (CloseLocked()) {
+ data_channel_listener_.data_channel_closed_cb();
+ }
+}
+
+bool ConnectionFlow::CloseLocked() {
+ if (state_ == State::kEnded) {
+ return false;
+ }
+ state_ = State::kEnded;
+
+ data_channel_future_.SetException({Exception::kInterrupted});
+ peer_connection_->Close();
+ data_channel_observer_.reset();
+ NEARBY_LOG(INFO, "Closed WebRTC connection.");
+ return true;
+}
+
+void ConnectionFlow::OffloadFromSignalingThread(Runnable runnable) {
+ single_threaded_signaling_offloader_.Execute(std::move(runnable));
+}
+
} // namespace mediums
} // namespace connections
} // namespace nearby
diff --git a/cpp/core_v2/internal/mediums/webrtc/connection_flow.h b/cpp/core_v2/internal/mediums/webrtc/connection_flow.h
index 661fd1bc..97cd0efc 100644
--- a/cpp/core_v2/internal/mediums/webrtc/connection_flow.h
+++ b/cpp/core_v2/internal/mediums/webrtc/connection_flow.h
@@ -18,8 +18,10 @@
#include
#include "core_v2/internal/mediums/webrtc/data_channel_listener.h"
+#include "core_v2/internal/mediums/webrtc/data_channel_observer_impl.h"
#include "core_v2/internal/mediums/webrtc/local_ice_candidate_listener.h"
#include "core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h"
+#include "core_v2/internal/mediums/webrtc/session_description_wrapper.h"
#include "platform_v2/base/runnable.h"
#include "platform_v2/public/future.h"
#include "platform_v2/public/single_thread_executor.h"
@@ -70,73 +72,98 @@ class ConnectionFlow {
// This method blocks on the creation of the peer connection object.
static std::unique_ptr Create(
LocalIceCandidateListener local_ice_candidate_listener,
- DataChannelListener data_channel_listener,
- SingleThreadExecutor* single_threaded_executor,
- WebRtcMedium& webrtc_medium);
- ~ConnectionFlow() = default;
+ DataChannelListener data_channel_listener, WebRtcMedium& webrtc_medium);
+ ~ConnectionFlow();
// Create the offer that will be sent to the remote. Mirrors the behaviour of
// PeerConnectionInterface::CreateOffer.
- std::unique_ptr CreateOffer()
- ABSL_LOCKS_EXCLUDED(mutex_);
+ SessionDescriptionWrapper CreateOffer() ABSL_LOCKS_EXCLUDED(mutex_);
// Create the answer that will be sent to the remote. Mirrors the behaviour of
// PeerConnectionInterface::CreateAnswer.
- std::unique_ptr CreateAnswer()
- ABSL_LOCKS_EXCLUDED(mutex_);
+ SessionDescriptionWrapper CreateAnswer() ABSL_LOCKS_EXCLUDED(mutex_);
// Set the local session description. |sdp| was created via CreateOffer()
// or CreateAnswer().
- bool SetLocalSessionDescription(
- std::unique_ptr sdp)
+ bool SetLocalSessionDescription(SessionDescriptionWrapper sdp)
ABSL_LOCKS_EXCLUDED(mutex_);
// Invoked when an offer was received from a remote; this will set the remote
- // session description on the peer connection.
- void OnOfferReceived(
- std::unique_ptr offer)
+ // session description on the peer connection. Returns true if the offer was
+ // successfully set as remote session description.
+ bool OnOfferReceived(SessionDescriptionWrapper offer)
ABSL_LOCKS_EXCLUDED(mutex_);
// Invoked when an answer was received from a remote; this will set the remote
- // session description on the peer connection.
- void OnAnswerReceived(
- std::unique_ptr answer)
+ // session description on the peer connection. Returns true if the offer was
+ // successfully set as remote session description.
+ bool OnAnswerReceived(SessionDescriptionWrapper answer)
ABSL_LOCKS_EXCLUDED(mutex_);
// Invoked when an ice candidate was received from a remote; this will add the
// ice candidate to the peer connection if ready or cache it otherwise.
bool OnRemoteIceCandidatesReceived(
- std::vector ice_candidates)
- ABSL_LOCKS_EXCLUDED(mutex_);
+ std::vector>
+ ice_candidates) ABSL_LOCKS_EXCLUDED(mutex_);
// Get a future for the data channel.
- api::ListenableFuture>*
- GetDataChannel();
+ Future>* GetDataChannel();
// Close the peer connection and data channel.
bool Close() ABSL_LOCKS_EXCLUDED(mutex_);
// Invoked when the peer connection indicates that signaling is stable.
- void OnSignalingStable();
+ void OnSignalingStable() ABSL_LOCKS_EXCLUDED(mutex_);
webrtc::DataChannelObserver* CreateDataChannelObserver(
rtc::scoped_refptr data_channel);
// Invoked upon changes in the state of peer connection, e.g. react to
// disconnect.
void ProcessOnPeerConnectionChange(
- webrtc::PeerConnectionInterface::PeerConnectionState new_state);
+ webrtc::PeerConnectionInterface::PeerConnectionState new_state)
+ ABSL_LOCKS_EXCLUDED(mutex_);
private:
+ enum class State {
+ kInitialized,
+ kCreatingOffer,
+ kWaitingForAnswer,
+ kReceivedOffer,
+ kCreatingAnswer,
+ kWaitingToConnect,
+ kConnected,
+ kEnded,
+ };
+
ConnectionFlow(LocalIceCandidateListener local_ice_candidate_listener,
- DataChannelListener data_channel_listener,
- SingleThreadExecutor* single_threaded_executor);
+ DataChannelListener data_channel_listener);
// TODO(bfranz): Consider whether this needs to be configurable per platform
static constexpr absl::Duration kTimeout = absl::Milliseconds(250);
bool InitPeerConnection(WebRtcMedium& webrtc_medium);
+ bool TransitionState(State current_state, State new_state)
+ ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ bool SetRemoteSessionDescription(SessionDescriptionWrapper sdp);
+
+ void ProcessDataChannelConnected() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ void CloseAndNotifyLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+ bool CloseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+ void OffloadFromSignalingThread(Runnable runnable);
+
+ Mutex mutex_;
+
+ State state_ ABSL_GUARDED_BY(mutex_) = State::kInitialized;
DataChannelListener data_channel_listener_;
+ std::unique_ptr data_channel_observer_;
+
Future> data_channel_future_;
PeerConnectionObserverImpl peer_connection_observer_;
rtc::scoped_refptr peer_connection_;
- Mutex mutex_;
+ std::vector>
+ cached_remote_ice_candidates_ ABSL_GUARDED_BY(mutex_);
+
+ SingleThreadExecutor single_threaded_signaling_offloader_;
};
} // namespace mediums
diff --git a/cpp/core_v2/internal/mediums/webrtc/connection_flow_test.cc b/cpp/core_v2/internal/mediums/webrtc/connection_flow_test.cc
index 8daf6356..8abaf197 100644
--- a/cpp/core_v2/internal/mediums/webrtc/connection_flow_test.cc
+++ b/cpp/core_v2/internal/mediums/webrtc/connection_flow_test.cc
@@ -15,10 +15,18 @@
#include "core_v2/internal/mediums/webrtc/connection_flow.h"
#include
+#include
+#include "core_v2/internal/mediums/webrtc/session_description_wrapper.h"
+#include "platform_v2/base/byte_array.h"
#include "platform_v2/public/webrtc.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
+#include "absl/time/time.h"
+#include "webrtc/api/data_channel_interface.h"
+#include "webrtc/api/jsep.h"
+#include "webrtc/api/rtc_error.h"
+#include "webrtc/api/scoped_refptr.h"
namespace location {
namespace nearby {
@@ -26,17 +34,159 @@ namespace connections {
namespace mediums {
namespace {
-TEST(ConnectionFlowTest, Create) {
- LocalIceCandidateListener local_ice_candidate_listener;
- DataChannelListener data_channel_listener;
- SingleThreadExecutor executor;
+std::unique_ptr CopyCandidate(
+ const webrtc::IceCandidateInterface* candidate) {
+ return webrtc::CreateIceCandidate(candidate->sdp_mid(),
+ candidate->sdp_mline_index(),
+ candidate->candidate());
+}
+
+// TODO(bfranz) - Add test that deterministically sends answerer_ice_candidates
+// before answer is sent.
+TEST(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
+ WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;
+
+ Future message_received_future;
+
+ std::unique_ptr offerer, answerer;
+
+ // Send Ice Candidates immediately when you retrieve them
+ offerer = ConnectionFlow::Create(
+ {.local_ice_candidate_found_cb =
+ [&answerer](const webrtc::IceCandidateInterface* candidate) {
+ std::vector> vec;
+ vec.push_back(CopyCandidate(candidate));
+ // The callback might be alive while the objects in test are
+ // destroyed.
+ if (answerer)
+ answerer->OnRemoteIceCandidatesReceived(std::move(vec));
+ }},
+ DataChannelListener(), webrtc_medium_offerer);
+ ASSERT_NE(offerer, nullptr);
+ answerer = ConnectionFlow::Create(
+ {.local_ice_candidate_found_cb =
+ [&offerer](const webrtc::IceCandidateInterface* candidate) {
+ std::vector> vec;
+ vec.push_back(CopyCandidate(candidate));
+ // The callback might be alive while the objects in test are
+ // destroyed.
+ if (offerer)
+ offerer->OnRemoteIceCandidatesReceived(std::move(vec));
+ }},
+ {.data_channel_message_received_cb =
+ [&message_received_future](ByteArray bytes) {
+ message_received_future.Set(std::move(bytes));
+ }},
+ webrtc_medium_answerer);
+ ASSERT_NE(answerer, nullptr);
+
+ // Create and send offer
+ SessionDescriptionWrapper offer = offerer->CreateOffer();
+ EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
+ EXPECT_TRUE(answerer->OnOfferReceived(offer));
+ EXPECT_TRUE(offerer->SetLocalSessionDescription(std::move(offer)));
+
+ // Create and send answer
+ SessionDescriptionWrapper answer = answerer->CreateAnswer();
+ EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer);
+ EXPECT_TRUE(offerer->OnAnswerReceived(answer));
+ EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer)));
+
+ // Retrieve Data Channels
+ ExceptionOr>
+ offerer_channel = offerer->GetDataChannel()->Get(absl::Seconds(1));
+ EXPECT_TRUE(offerer_channel.ok());
+ ExceptionOr>
+ answerer_channel = answerer->GetDataChannel()->Get(absl::Seconds(1));
+ EXPECT_TRUE(answerer_channel.ok());
+
+ // Send message on data channel
+ const char message[] = "Test";
+ offerer_channel.result()->Send(webrtc::DataBuffer(message));
+ ExceptionOr received_message =
+ message_received_future.Get(absl::Seconds(1));
+ EXPECT_TRUE(received_message.ok());
+ EXPECT_EQ(received_message.result(), ByteArray{message});
+}
+
+TEST(ConnectionFlowTest, CreateAnswerBeforeOfferReceived) {
WebRtcMedium webrtc_medium;
- std::unique_ptr connection_flow = ConnectionFlow::Create(
- std::move(local_ice_candidate_listener), std::move(data_channel_listener),
- &executor, webrtc_medium);
+ std::unique_ptr answerer = ConnectionFlow::Create(
+ LocalIceCandidateListener(), DataChannelListener(), webrtc_medium);
+ ASSERT_NE(answerer, nullptr);
- EXPECT_NE(connection_flow, nullptr);
+ SessionDescriptionWrapper answer = answerer->CreateAnswer();
+ EXPECT_FALSE(answer.IsValid());
+}
+
+TEST(ConnectionFlowTest, SetAnswerBeforeOffer) {
+ WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;
+
+ std::unique_ptr offerer =
+ ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
+ webrtc_medium_offerer);
+ ASSERT_NE(offerer, nullptr);
+ std::unique_ptr answerer =
+ ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
+ webrtc_medium_answerer);
+ ASSERT_NE(answerer, nullptr);
+
+ SessionDescriptionWrapper offer = offerer->CreateOffer();
+ EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
+ // Did not set offer as local session description
+ EXPECT_TRUE(answerer->OnOfferReceived(offer));
+
+ SessionDescriptionWrapper answer = answerer->CreateAnswer();
+ EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer);
+ EXPECT_FALSE(offerer->OnAnswerReceived(answer));
+}
+
+TEST(ConnectionFlowTest, CannotCreateOfferAfterClose) {
+ WebRtcMedium webrtc_medium;
+
+ std::unique_ptr offerer = ConnectionFlow::Create(
+ LocalIceCandidateListener(), DataChannelListener(), webrtc_medium);
+ ASSERT_NE(offerer, nullptr);
+
+ EXPECT_TRUE(offerer->Close());
+
+ EXPECT_FALSE(offerer->CreateOffer().IsValid());
+}
+
+TEST(ConnectionFlowTest, CannotSetSessionDescriptionAfterClose) {
+ WebRtcMedium webrtc_medium;
+
+ std::unique_ptr offerer = ConnectionFlow::Create(
+ LocalIceCandidateListener(), DataChannelListener(), webrtc_medium);
+ ASSERT_NE(offerer, nullptr);
+
+ SessionDescriptionWrapper offer = offerer->CreateOffer();
+ EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
+
+ EXPECT_TRUE(offerer->Close());
+
+ EXPECT_FALSE(offerer->SetLocalSessionDescription(offer));
+}
+
+TEST(ConnectionFlowTest, CannotReceiveOfferAfterClose) {
+ WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;
+
+ std::unique_ptr offerer =
+ ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
+ webrtc_medium_offerer);
+ ASSERT_NE(offerer, nullptr);
+ std::unique_ptr answerer =
+ ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
+ webrtc_medium_answerer);
+ ASSERT_NE(answerer, nullptr);
+
+ EXPECT_TRUE(answerer->Close());
+
+ SessionDescriptionWrapper offer = offerer->CreateOffer();
+ EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
+
+ EXPECT_FALSE(answerer->OnOfferReceived(offer));
}
} // namespace
diff --git a/cpp/core_v2/internal/mediums/webrtc/data_channel_listener.h b/cpp/core_v2/internal/mediums/webrtc/data_channel_listener.h
index fab89933..58bcb7a9 100644
--- a/cpp/core_v2/internal/mediums/webrtc/data_channel_listener.h
+++ b/cpp/core_v2/internal/mediums/webrtc/data_channel_listener.h
@@ -28,8 +28,8 @@ struct DataChannelListener {
std::function data_channel_closed_cb = DefaultCallback<>();
// Called when a new message was received on the data channel.
- std::function data_channel_message_received_cb =
- DefaultCallback();
+ std::function data_channel_message_received_cb =
+ DefaultCallback();
// Called when the data channel indicates that the buffered amount has
// changed.
diff --git a/cpp/core_v2/internal/mediums/webrtc/data_channel_observer_impl.cc b/cpp/core_v2/internal/mediums/webrtc/data_channel_observer_impl.cc
new file mode 100644
index 00000000..0781c501
--- /dev/null
+++ b/cpp/core_v2/internal/mediums/webrtc/data_channel_observer_impl.cc
@@ -0,0 +1,42 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "core_v2/internal/mediums/webrtc/data_channel_observer_impl.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+DataChannelObserverImpl::DataChannelObserverImpl(
+ DataChannelListener* data_channel_listener,
+ DataChannelStateChangeCallback callback)
+ : data_channel_listener_(data_channel_listener),
+ state_change_callback_(std::move(callback)) {}
+
+void DataChannelObserverImpl::OnStateChange() { state_change_callback_(); }
+
+void DataChannelObserverImpl::OnMessage(const webrtc::DataBuffer& buffer) {
+ data_channel_listener_->data_channel_message_received_cb(
+ ByteArray(buffer.data.data(), buffer.size()));
+}
+
+void DataChannelObserverImpl::OnBufferedAmountChange(uint64_t sent_data_size) {
+ data_channel_listener_->data_channel_buffered_amount_changed_cb();
+}
+
+} // namespace mediums
+} // namespace connections
+} // namespace nearby
+} // namespace location
diff --git a/cpp/core_v2/internal/mediums/webrtc/data_channel_observer_impl.h b/cpp/core_v2/internal/mediums/webrtc/data_channel_observer_impl.h
new file mode 100644
index 00000000..9c7ac1b3
--- /dev/null
+++ b/cpp/core_v2/internal/mediums/webrtc/data_channel_observer_impl.h
@@ -0,0 +1,49 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_OBSERVER_IMPL_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_OBSERVER_IMPL_H_
+
+#include "core_v2/internal/mediums/webrtc/data_channel_listener.h"
+#include "webrtc/api/data_channel_interface.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+class DataChannelObserverImpl : public webrtc::DataChannelObserver {
+ public:
+ using DataChannelStateChangeCallback = std::function;
+
+ ~DataChannelObserverImpl() override = default;
+ DataChannelObserverImpl(DataChannelListener* data_channel_listener,
+ DataChannelStateChangeCallback callback);
+
+ // webrtc::DataChannelObserver:
+ void OnStateChange() override;
+ void OnMessage(const webrtc::DataBuffer& buffer) override;
+ void OnBufferedAmountChange(uint64_t sent_data_size) override;
+
+ private:
+ DataChannelListener* data_channel_listener_;
+ DataChannelStateChangeCallback state_change_callback_;
+};
+
+} // namespace mediums
+} // namespace connections
+} // namespace nearby
+} // namespace location
+
+#endif // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_OBSERVER_IMPL_H_
diff --git a/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.cc b/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.cc
index 1be9c968..73d9f868 100644
--- a/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.cc
+++ b/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.cc
@@ -24,11 +24,9 @@ namespace mediums {
PeerConnectionObserverImpl::PeerConnectionObserverImpl(
ConnectionFlow* connection_flow,
- LocalIceCandidateListener local_ice_candidate_listener,
- SingleThreadExecutor* executor)
+ LocalIceCandidateListener local_ice_candidate_listener)
: connection_flow_(connection_flow),
- local_ice_candidate_listener_(std::move(local_ice_candidate_listener)),
- single_threaded_signaling_offloader_(executor) {}
+ local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {}
void PeerConnectionObserverImpl::OnIceCandidate(
const webrtc::IceCandidateInterface* candidate) {
@@ -73,7 +71,7 @@ void PeerConnectionObserverImpl ::OnRenegotiationNeeded() {
}
void PeerConnectionObserverImpl::OffloadFromSignalingThread(Runnable runnable) {
- single_threaded_signaling_offloader_->Execute(std::move(runnable));
+ single_threaded_signaling_offloader_.Execute(std::move(runnable));
}
} // namespace mediums
diff --git a/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h b/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h
index 6265bab1..6093e062 100644
--- a/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h
+++ b/cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h
@@ -31,8 +31,7 @@ class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
~PeerConnectionObserverImpl() override = default;
PeerConnectionObserverImpl(
ConnectionFlow* connection_flow,
- LocalIceCandidateListener local_ice_candidate_listener,
- SingleThreadExecutor* executor);
+ LocalIceCandidateListener local_ice_candidate_listener);
// webrtc::PeerConnectionObserver:
void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override;
@@ -51,7 +50,7 @@ class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
ConnectionFlow* connection_flow_;
LocalIceCandidateListener local_ice_candidate_listener_;
- SingleThreadExecutor* single_threaded_signaling_offloader_;
+ SingleThreadExecutor single_threaded_signaling_offloader_;
};
} // namespace mediums
diff --git a/cpp/core_v2/internal/mediums/webrtc/peer_id.cc b/cpp/core_v2/internal/mediums/webrtc/peer_id.cc
index 6a830627..59ab04f2 100644
--- a/cpp/core_v2/internal/mediums/webrtc/peer_id.cc
+++ b/cpp/core_v2/internal/mediums/webrtc/peer_id.cc
@@ -46,6 +46,8 @@ PeerId PeerId::FromSeed(const ByteArray& seed) {
return PeerId(BytesToStringUppercase(hashed_seed));
}
+bool PeerId::IsValid() const { return !id_.empty(); }
+
} // namespace mediums
} // namespace connections
} // namespace nearby
diff --git a/cpp/core_v2/internal/mediums/webrtc/peer_id.h b/cpp/core_v2/internal/mediums/webrtc/peer_id.h
index 307724af..c182f20f 100644
--- a/cpp/core_v2/internal/mediums/webrtc/peer_id.h
+++ b/cpp/core_v2/internal/mediums/webrtc/peer_id.h
@@ -26,19 +26,22 @@ namespace connections {
namespace mediums {
// PeerId is used as an identifier to exchange SDP messages to establish WebRTC
-// p2p connection.
+// p2p connection. An empty PeerId is considered to be invalid.
class PeerId {
public:
- explicit PeerId(const string& id) : id_(id) {}
+ PeerId() = default;
+ explicit PeerId(const std::string& id) : id_(id) {}
~PeerId() = default;
static PeerId FromRandom();
static PeerId FromSeed(const ByteArray& seed);
- const string& GetId() const { return id_; }
+ bool IsValid() const;
+
+ const std::string& GetId() const { return id_; }
private:
- const string id_;
+ std::string id_;
};
} // namespace mediums
diff --git a/cpp/core_v2/internal/mediums/webrtc/session_description_wrapper.h b/cpp/core_v2/internal/mediums/webrtc/session_description_wrapper.h
new file mode 100644
index 00000000..e68c816e
--- /dev/null
+++ b/cpp/core_v2/internal/mediums/webrtc/session_description_wrapper.h
@@ -0,0 +1,64 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_SESSION_DESCRIPTION_WRAPPER_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_SESSION_DESCRIPTION_WRAPPER_H_
+
+#include "webrtc/api/peer_connection_interface.h"
+
+// Wrapper object around SessionDescriptionInterface*.
+// This object owns the SessionDescriptionInterface* unless Release() has been
+// called.
+class SessionDescriptionWrapper {
+ public:
+ SessionDescriptionWrapper() = default;
+ explicit SessionDescriptionWrapper(webrtc::SessionDescriptionInterface* sdp)
+ : impl_(sdp) {}
+
+ // Copy constructor that performs a deep copy, i.e. creates a new
+ // SessionDescriptionInterface.
+ SessionDescriptionWrapper(const SessionDescriptionWrapper& sdp) {
+ if (sdp.IsValid()) {
+ impl_ = webrtc::CreateSessionDescription(sdp.GetType(), sdp.ToString());
+ }
+ }
+
+ SessionDescriptionWrapper(SessionDescriptionWrapper&&) = default;
+ SessionDescriptionWrapper& operator=(SessionDescriptionWrapper&&) = default;
+
+ // Release the ownership of the SessionDescriptionInterface*.
+ webrtc::SessionDescriptionInterface* Release() { return impl_.release(); }
+
+ // Returns a string representation of the sdp. Only call this, if IsValid() is
+ // true.
+ std::string ToString() const {
+ std::string str;
+ impl_->ToString(&str);
+ return str;
+ }
+
+ // Returns the SdpType of the SessionDescriptionInterface. Only call this, if
+ // IsValid() is true.
+ webrtc::SdpType GetType() const { return impl_->GetType(); }
+
+ const webrtc::SessionDescriptionInterface& GetSdp() { return *impl_; }
+
+ // Return whether this object currently holds a SessionDescriptionInterface.
+ bool IsValid() const { return impl_ != nullptr; }
+
+ private:
+ std::unique_ptr impl_;
+};
+
+#endif // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_SESSION_DESCRIPTION_WRAPPER_H_
diff --git a/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.cc b/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.cc
index 11f89236..fa02d005 100644
--- a/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.cc
+++ b/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.cc
@@ -54,7 +54,7 @@ Exception WebRtcSocket::OutputStreamImpl::Close() {
// WebRtcSocket
WebRtcSocket::WebRtcSocket(
- const string& name,
+ const std::string& name,
rtc::scoped_refptr data_channel)
: name_(name), data_channel_(std::move(data_channel)) {}
diff --git a/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.h b/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.h
index 91bff312..ef3836f0 100644
--- a/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.h
+++ b/cpp/core_v2/internal/mediums/webrtc/webrtc_socket.h
@@ -41,7 +41,7 @@ constexpr int kMaxDataSize = 1 * 1024 * 1024;
// which could lead to data loss.
class WebRtcSocket : public Socket {
public:
- WebRtcSocket(const string& name,
+ WebRtcSocket(const std::string& name,
rtc::scoped_refptr data_channel);
~WebRtcSocket() override = default;
@@ -92,7 +92,7 @@ class WebRtcSocket : public Socket {
bool SendMessage(const ByteArray& data);
void BlockUntilSufficientSpaceInBuffer(int length);
- string name_;
+ std::string name_;
rtc::scoped_refptr data_channel_;
Pipe pipe_;
diff --git a/cpp/core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h b/cpp/core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h
new file mode 100644
index 00000000..2aa66d8d
--- /dev/null
+++ b/cpp/core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h
@@ -0,0 +1,63 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_WRAPPER_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_WRAPPER_H_
+
+#include
+
+#include "core_v2/internal/mediums/webrtc/webrtc_socket.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+class WebRtcSocketWrapper final {
+ public:
+ WebRtcSocketWrapper() = default;
+ WebRtcSocketWrapper(const WebRtcSocketWrapper&) = default;
+ WebRtcSocketWrapper& operator=(const WebRtcSocketWrapper&) = default;
+ explicit WebRtcSocketWrapper(std::unique_ptr socket)
+ : impl_(socket.release()) {}
+ ~WebRtcSocketWrapper() = default;
+
+ InputStream& GetInputStream() { return impl_->GetInputStream(); }
+
+ OutputStream& GetOutputStream() { return impl_->GetOutputStream(); }
+
+ void NotifyDataChannelMsgReceived(const ByteArray& message) {
+ impl_->NotifyDataChannelMsgReceived(message);
+ }
+
+ void NotifyDataChannelBufferedAmountChanged() {
+ impl_->NotifyDataChannelBufferedAmountChanged();
+ }
+
+ void Close() { return impl_->Close(); }
+
+ bool IsValid() const { return impl_ != nullptr; }
+
+ WebRtcSocket& GetImpl() { return *impl_; }
+
+ private:
+ std::shared_ptr impl_;
+};
+
+} // namespace mediums
+} // namespace connections
+} // namespace nearby
+} // namespace location
+
+#endif // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_WRAPPER_H_
diff --git a/cpp/core_v2/internal/mediums/webrtc_test.cc b/cpp/core_v2/internal/mediums/webrtc_test.cc
new file mode 100644
index 00000000..a375f60d
--- /dev/null
+++ b/cpp/core_v2/internal/mediums/webrtc_test.cc
@@ -0,0 +1,135 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "core_v2/internal/mediums/webrtc.h"
+
+#include "core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h"
+#include "platform_v2/base/listeners.h"
+#include "platform_v2/public/mutex_lock.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+namespace {
+
+// Basic test to check that device is accepting connections when initialized.
+TEST(WebRtcTest, NotAcceptingConnections) {
+ WebRtc webrtc;
+ ASSERT_TRUE(webrtc.IsAvailable());
+ EXPECT_FALSE(webrtc.IsAcceptingConnections());
+}
+
+// Tests the flow when the device tries to accept connections twice. In this
+// case, only the first call is successful and subsequent calls fail.
+TEST(WebRtcTest, StartAcceptingConnectionTwice) {
+ using MockAcceptedCallback =
+ testing::MockFunction