Move TachyonExpressSignallingMessenger into mediums/webrtc.

PiperOrigin-RevId: 919882997

internal/platform/BUILD

@@ -284,38 +284,6 @@ cc_library(
     ],
 )
 
-cc_library(
-    name = "tachyon_express_signaling_messenger",
-    srcs = ["tachyon_express_signaling_messenger.cc"],
-    hdrs = ["tachyon_express_signaling_messenger.h"],
-    visibility = [
-        "//connections:__subpackages__",
-        "//internal/platform/implementation:__subpackages__",
-        "//internal/test:__subpackages__",
-        "//third_party/nearby/presence:__subpackages__",
-    ],
-    deps = [
-        ":base",
-        ":logging",
-        ":types",
-        "//internal/account",
-        "//internal/platform/implementation:webrtc_platform",
-        "//internal/proto:messaging_cc_grpc_proto",
-        "//internal/proto:tachyon_cc_proto",
-        "//internal/rpc:utils",
-        "//location/nearby/sharing/lib/account:account_manager",
-        "//third_party/gloop/util/random:mt_random",
-        "//third_party/grpc:gpr",
-        "//third_party/grpc:grpc++",
-        "//util/random:util",
-        "@com_google_absl//absl/base:core_headers",
-        "@com_google_absl//absl/functional:any_invocable",
-        "@com_google_absl//absl/strings",
-        "@com_google_absl//absl/synchronization",
-        "@com_google_absl//absl/time",
-    ],
-)
-
 cc_library(
     name = "comm",
     srcs = [

internal/platform/implementation/apple/BUILD

@@ -61,8 +61,8 @@ objc_library(
         "webrtc.h",
     ],
     deps = [
+        "//connections/implementation/mediums/webrtc:tachyon_express_signaling_messenger",
         "//internal/platform:logging",
-        "//internal/platform:tachyon_express_signaling_messenger",
         "//internal/platform:types",
         "//internal/platform/implementation:webrtc_platform",
         "//internal/proto:tachyon_cc_proto",
@@ -130,7 +130,6 @@ objc_library(
         "//internal/crypto_cros",
         "//internal/platform:comm",
         "//internal/platform:logging",
-        "//internal/platform:tachyon_express_signaling_messenger",
         "//internal/platform:types",
         "//internal/proto:tachyon_cc_proto",
         "//internal/platform:base",

internal/platform/implementation/apple/webrtc.mm

@@ -25,10 +25,10 @@
 
 #include "absl/status/status.h"
 #include "absl/strings/string_view.h"
+#include "connections/implementation/mediums/webrtc/tachyon_express_signaling_messenger.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/crypto.h"
 #include "internal/platform/logging.h"
-#include "internal/platform/tachyon_express_signaling_messenger.h"
 #include "internal/proto/tachyon.pb.h"
 #include "internal/proto/tachyon_enums.proto.h"
 #include "webrtc/api/create_modular_peer_connection_factory.h"
@@ -91,7 +91,8 @@ void WebRtcMedium::CreatePeerConnection(
 
 std::unique_ptr<api::WebRtcSignalingMessenger> WebRtcMedium::GetSignalingMessenger(
     absl::string_view self_id, const location::nearby::connections::LocationHint& location_hint) {
-  return std::make_unique<TachyonExpressSignalingMessenger>(self_id, location_hint);
+  return std::make_unique<
+      nearby::connections::mediums::TachyonExpressSignalingMessenger>(self_id, location_hint);
 }
 
 }  // namespace nearby::apple

internal/platform/implementation/windows/BUILD

@@ -238,9 +238,9 @@ cc_library(
     hdrs = ["webrtc.h"],
     tags = ["windows"],
     deps = [
+        "//connections/implementation/mediums/webrtc:tachyon_express_signaling_messenger",
         "//internal/platform:logging",
-        "//internal/platform:tachyon_express_signaling_messenger",
-        "//internal/platform/implementation:comm",
+        "//internal/platform/implementation:webrtc_platform",
         "//third_party/webrtc/files/stable/webrtc/api:create_modular_peer_connection_factory",
         "//third_party/webrtc/files/stable/webrtc/api:peer_connection_interface",
         "//third_party/webrtc/files/stable/webrtc/api:rtc_error",

internal/platform/implementation/windows/webrtc.cc

@@ -22,9 +22,9 @@
 #include <utility>
 
 #include "absl/strings/string_view.h"
+#include "connections/implementation/mediums/webrtc/tachyon_express_signaling_messenger.h"
 #include "internal/platform/implementation/webrtc.h"
 #include "internal/platform/logging.h"
-#include "internal/platform/tachyon_express_signaling_messenger.h"
 #include "webrtc/api/create_modular_peer_connection_factory.h"
 #include "webrtc/api/peer_connection_interface.h"
 #include "webrtc/api/rtc_error.h"
@@ -33,6 +33,8 @@
 
 namespace nearby::windows {
 
+using ::nearby::connections::mediums::TachyonExpressSignalingMessenger;
+
 std::string WebRtcMedium::GetDefaultCountryCode() {
   wchar_t systemGeoName[LOCALE_NAME_MAX_LENGTH];
 

internal/platform/tachyon_express_signaling_messenger.cc

@@ -1,345 +0,0 @@
-// Copyright 2025 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 NO_WEBRTC
-
-#include "internal/platform/tachyon_express_signaling_messenger.h"
-
-#include <ctime>
-#include <memory>
-#include <string>
-#include <utility>
-
-#include "absl/functional/any_invocable.h"
-#include "absl/strings/string_view.h"
-#include "absl/synchronization/mutex.h"
-#include "absl/time/time.h"
-#include "third_party/gloop/util/random/mt_random.h"
-#include "third_party/grpc/include/grpc/support/time.h"
-#include "third_party/grpc/include/grpcpp/channel.h"
-#include "third_party/grpc/include/grpcpp/client_context.h"
-#include "third_party/grpc/include/grpcpp/create_channel.h"
-#include "third_party/grpc/include/grpcpp/security/credentials.h"
-#include "third_party/grpc/include/grpcpp/support/client_callback.h"
-#include "third_party/grpc/include/grpcpp/support/status.h"
-#include "internal/account/account_manager_impl.h"
-#include "internal/platform/byte_array.h"
-#include "internal/platform/count_down_latch.h"
-#include "internal/platform/implementation/webrtc.h"
-#include "internal/platform/logging.h"
-#include "internal/proto/messaging.grpc.pb.h"
-#include "internal/proto/tachyon.proto.h"
-#include "internal/proto/tachyon_common.proto.h"
-#include "internal/proto/tachyon_enums.proto.h"
-#include "internal/rpc/utils.h"
-#include "util/random/util.h"
-
-namespace nearby {
-
-namespace {
-using ::google::internal::communications::instantmessaging::v1::ClientInfo;
-using ::google::internal::communications::instantmessaging::v1::Id;
-using ::google::internal::communications::instantmessaging::v1::
-    LocationStandard;
-using ::google::internal::communications::instantmessaging::v1::
-    ReceiveMessagesResponse;
-using ::google::internal::communications::instantmessaging::v1::RequestHeader;
-using ::google::internal::communications::instantmessaging::v1::
-    SendMessageExpressRequest;
-using ::google::internal::communications::instantmessaging::v1::
-    SendMessageExpressResponse;
-
-constexpr absl::string_view kApp = "Nearby";
-constexpr absl::string_view kTachyonAddress =
-    "instantmessaging-pa.googleapis.com:443";
-
-// It is unclear to me where these magic numbers are from but they are used
-// across both the Android and CrOS implementations.
-// See:
-// https://source.chromium.org/chromium/chromium/src/+/main:chrome/browser/nearby_sharing/tachyon_ice_config_fetcher.cc;l=53
-constexpr int kMajorVersion = 1;
-constexpr int kMinorVersion = 24;
-constexpr int kPointVersion = 0;
-
-void InitId(Id& id, absl::string_view id_str,
-            const location::nearby::connections::LocationHint& location_hint) {
-  id.set_id(id_str);
-  id.set_app(kApp);
-  id.set_type(google::internal::communications::instantmessaging::v1::IdType::
-                  NEARBY_ID);
-  auto* request_location_hint = id.mutable_location_hint();
-  request_location_hint->set_location(location_hint.location());
-  if (location_hint.format() ==
-      location::nearby::connections::LocationStandard::E164_CALLING) {
-    request_location_hint->set_format(LocationStandard::E164_CALLING);
-  } else if (location_hint.format() ==
-             location::nearby::connections::LocationStandard::
-                 ISO_3166_1_ALPHA_2) {
-    request_location_hint->set_format(LocationStandard::ISO_3166_1_ALPHA_2);
-  } else {
-    request_location_hint->set_format(LocationStandard::UNKNOWN);
-  }
-}
-
-void InitRequestHeader(
-    RequestHeader& header, absl::string_view self_id,
-    const location::nearby::connections::LocationHint& location_hint) {
-  ClientInfo* client_info = header.mutable_client_info();
-  client_info->set_platform_type(google::internal::communications::
-                                     instantmessaging::v1::Platform::DESKTOP);
-
-  client_info->set_major(kMajorVersion);
-  client_info->set_minor(kMinorVersion);
-  client_info->set_point(kPointVersion);
-
-  client_info->set_api_version(
-      google::internal::communications::instantmessaging::v1::ApiVersion::V4);
-
-  // Generate a random message identifier.
-  MTRandom rand;
-  header.set_request_id(util_random::RandomString(&rand, /*length=*/13,
-                                                  util::random::kWebsafe64));
-  header.set_app(kApp);
-
-  InitId(*header.mutable_requester_id(), self_id, location_hint);
-}
-
-}  // namespace
-
-bool TachyonExpressSignalingMessenger::ReceiveMessagesReader::Start(
-    google::internal::communications::instantmessaging::v1::grpc::Messaging::
-        StubInterface* stub,
-    absl::string_view self_id,
-    const location::nearby::connections::LocationHint& location_hint,
-    absl::string_view access_token,
-    absl::AnyInvocable<void()> on_fast_path_ready_callback,
-    absl::AnyInvocable<void(const ByteArray&)> on_inbox_message_callback,
-    absl::AnyInvocable<void(bool)> on_complete_callback) {
-  {
-    absl::MutexLock lock(mutex_);
-    if (is_receiving_messages_) {
-      return false;
-    }
-    is_receiving_messages_ = true;
-  }
-
-  on_fast_path_ready_callback_ = std::move(on_fast_path_ready_callback);
-  on_inbox_message_callback_ = std::move(on_inbox_message_callback);
-  on_complete_callback_ = std::move(on_complete_callback);
-  const std::shared_ptr<grpc::CallCredentials> call_creds =
-      grpc::AccessTokenCredentials(std::string(access_token));
-  context_.set_credentials(call_creds);
-  gpr_timespec deadline = gpr_now(GPR_CLOCK_MONOTONIC);
-  timespec timespec = absl::ToTimespec(absl::Seconds(30));
-  deadline.tv_sec += timespec.tv_sec;
-  deadline.tv_nsec += timespec.tv_nsec;
-  context_.set_deadline(deadline);
-
-  InitRequestHeader(*request_.mutable_header(), self_id, location_hint);
-  stub->async()->ReceiveMessagesExpress(&context_, &request_, this);
-  StartRead(&response_);
-  StartCall();
-  return true;
-}
-
-void TachyonExpressSignalingMessenger::ReceiveMessagesReader::OnReadDone(
-    bool ok) {
-  {
-    absl::MutexLock lock(mutex_);
-    if (!is_receiving_messages_) {
-      return;
-    }
-  }
-  if (ok) {
-    switch (response_.body_case()) {
-      case ReceiveMessagesResponse::kFastPathReady:
-        on_fast_path_ready_callback_();
-        break;
-      case ReceiveMessagesResponse::kInboxMessage:
-        on_inbox_message_callback_(
-            ByteArray(response_.inbox_message().message()));
-        break;
-      default:
-        break;
-    }
-    StartRead(&response_);
-  }
-}
-
-void TachyonExpressSignalingMessenger::ReceiveMessagesReader::OnDone(
-    const grpc::Status& s) {
-  {
-    absl::MutexLock lock(mutex_);
-    if (!is_receiving_messages_) {
-      return;
-    }
-  }
-  if (!s.ok()) {
-    LOG(ERROR) << "ReceiveMessagesExpress failed: "
-               << rpc::GrpcStatusToAbslStatus(s);
-  }
-  on_complete_callback_(s.ok());
-}
-
-void TachyonExpressSignalingMessenger::ReceiveMessagesReader::TryCancel() {
-  {
-    absl::MutexLock lock(mutex_);
-    if (!is_receiving_messages_) {
-      return;
-    }
-    is_receiving_messages_ = false;
-  }
-  context_.TryCancel();
-}
-
-TachyonExpressSignalingMessenger::TachyonExpressSignalingMessenger(
-    absl::string_view self_id,
-    const location::nearby::connections::LocationHint& location_hint)
-    : self_id_(self_id),
-      location_hint_(location_hint),
-      account_manager_(AccountManagerImpl::Factory::instance()) {
-  std::shared_ptr<grpc::Channel> channel =
-      grpc::CreateChannel(std::string(kTachyonAddress),
-                          grpc::SslCredentials(grpc::SslCredentialsOptions()));
-  messaging_stub_ = google::internal::communications::instantmessaging::v1::
-      grpc::Messaging::NewStub(channel);
-}
-
-struct StartState {
-  CountDownLatch latch{1};
-  bool success = false;
-};
-
-bool TachyonExpressSignalingMessenger::StartReceivingMessages(
-    OnSignalingMessageCallback on_message_callback,
-    OnSignalingCompleteCallback on_complete_callback) {
-  auto state = std::make_shared<StartState>();
-
-  account_manager_->GetAccessToken(
-      [this, state, on_message_callback = std::move(on_message_callback),
-       on_complete_callback = std::move(on_complete_callback)](
-          absl::StatusOr<std::string> token) mutable {
-        if (!token.ok()) {
-          state->success = false;
-          state->latch.CountDown();
-          return;
-        }
-        auto reader = std::make_shared<ReceiveMessagesReader>();
-
-        reader_ = reader;
-
-        std::weak_ptr<StartState> weak_state = state;
-
-        bool started = reader->Start(
-            messaging_stub_.get(), self_id_, location_hint_, token.value(),
-            /*on_fast_path_ready_callback=*/
-            [state] {
-              LOG(INFO) << "Received fast path ready message from tachyon.";
-              state->success = true;
-              state->latch.CountDown();
-            },
-            std::move(on_message_callback),
-            [reader, weak_state,
-             cb = std::move(on_complete_callback)](bool s) mutable {
-              LOG(INFO) << "Finished receiving messages from tachyon.";
-              cb(s);
-              if (auto state = weak_state.lock()) {
-                state->success = false;
-                state->latch.CountDown();
-              }
-            });
-
-        if (!started) {
-          state->success = false;
-          state->latch.CountDown();
-        }
-      });
-  state->latch.Await();
-  if (state->success) {
-    LOG(INFO) << "Receiving messages from tachyon.";
-  } else {
-    LOG(ERROR) << "Failed to start receiving messages from tachyon.";
-    reader_.reset();
-  }
-  return state->success;
-}
-
-void TachyonExpressSignalingMessenger::StopReceivingMessages() {
-  if (reader_) {
-    reader_->TryCancel();
-    reader_.reset();
-  }
-}
-
-bool TachyonExpressSignalingMessenger::SendMessage(absl::string_view peer_id,
-                                                   const ByteArray& message) {
-  auto rpc_state =
-      std::make_shared<rpc::AsyncRpcArgs<SendMessageExpressRequest,
-                                         SendMessageExpressResponse>>();
-
-  InitRequestHeader(*rpc_state->request.mutable_header(), self_id_,
-                    location_hint_);
-  InitId(*rpc_state->request.mutable_dest_id(), peer_id, location_hint_);
-
-  auto* request_message = rpc_state->request.mutable_message();
-  request_message->set_message(message.string_data());
-  request_message->set_message_type(
-      google::internal::communications::instantmessaging::v1::InboxMessage::
-          BASIC);
-  request_message->set_message_class(
-      google::internal::communications::instantmessaging::v1::InboxMessage::
-          EPHEMERAL);
-  MTRandom rand;
-  request_message->set_message_id(util_random::RandomString(
-      &rand, /*length=*/13, util::random::kWebsafe64));
-
-  CountDownLatch latch(1);
-  bool success = false;
-  account_manager_->GetAccessToken(
-      [this, &latch, &success, rpc_state](absl::StatusOr<std::string> token) {
-        if (!token.ok()) {
-          success = false;
-          latch.CountDown();
-          return;
-        }
-
-        const std::shared_ptr<grpc::CallCredentials> call_creds =
-            grpc::AccessTokenCredentials(token.value());
-        rpc_state->context.set_credentials(call_creds);
-        gpr_timespec deadline = gpr_now(GPR_CLOCK_MONOTONIC);
-        timespec timespec = absl::ToTimespec(absl::Seconds(30));
-        deadline.tv_sec += timespec.tv_sec;
-        deadline.tv_nsec += timespec.tv_nsec;
-        rpc_state->context.set_deadline(deadline);
-
-        // `rpc_state` is captured to ensure its lifetime is valid until the
-        // callback is executed.
-        messaging_stub_->async()->SendMessageExpress(
-            &rpc_state->context, &rpc_state->request, &rpc_state->response,
-            [&success, &latch, rpc_state](grpc::Status status) {
-              if (!status.ok()) {
-                LOG(ERROR) << "SendMessageExpress failed: "
-                           << rpc::GrpcStatusToAbslStatus(status);
-              }
-              success = status.ok();
-              latch.CountDown();
-            });
-      });
-  latch.Await();
-  return success;
-}
-
-}  // namespace nearby
-
-#endif  // #ifndef NO_WEBRTC

internal/platform/tachyon_express_signaling_messenger.h

@@ -1,105 +0,0 @@
-// Copyright 2025 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 THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_TACHYON_MESSAGING_CLIENT_H_
-#define THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_TACHYON_MESSAGING_CLIENT_H_
-
-#ifndef NO_WEBRTC
-
-#include <grpcpp/security/credentials.h>
-
-#include <memory>
-#include <string>
-
-#include "location/nearby/sharing/lib/account/account_manager.h"
-#include "absl/base/thread_annotations.h"
-#include "absl/functional/any_invocable.h"
-#include "absl/strings/string_view.h"
-#include "absl/synchronization/mutex.h"
-#include "third_party/grpc/include/grpcpp/client_context.h"
-#include "third_party/grpc/include/grpcpp/support/client_callback.h"
-#include "third_party/grpc/include/grpcpp/support/status.h"
-#include "internal/platform/byte_array.h"
-#include "internal/platform/implementation/webrtc.h"
-#include "internal/proto/messaging.grpc.pb.h"
-
-namespace nearby {
-
-// Interface for the messaging Tachyon service. See
-// third_party/nearby/internal/proto/messaging.proto
-class TachyonExpressSignalingMessenger : public api::WebRtcSignalingMessenger {
- public:
-  explicit TachyonExpressSignalingMessenger(
-      absl::string_view self_id,
-      const location::nearby::connections::LocationHint& location_hint);
-
-  class ReceiveMessagesReader
-      : public grpc::ClientReadReactor<
-            google::internal::communications::instantmessaging::v1::
-                ReceiveMessagesResponse> {
-   public:
-    ReceiveMessagesReader() = default;
-
-    void OnReadDone(bool ok) override;
-    void OnDone(const grpc::Status& s) override;
-
-    bool Start(
-        google::internal::communications::instantmessaging::v1::grpc::
-            Messaging::StubInterface* stub,
-        absl::string_view self_id,
-        const location::nearby::connections::LocationHint& location_hint,
-        absl::string_view access_token,
-        absl::AnyInvocable<void()> on_fast_path_ready_callback,
-        absl::AnyInvocable<void(const ByteArray&)> on_inbox_message_callback,
-        absl::AnyInvocable<void(bool)> on_complete_callback);
-    void TryCancel();
-
-   private:
-    grpc::ClientContext context_;
-    google::internal::communications::instantmessaging::v1::
-        ReceiveMessagesExpressRequest request_;
-    google::internal::communications::instantmessaging::v1::
-        ReceiveMessagesResponse response_;
-    absl::AnyInvocable<void()> on_fast_path_ready_callback_;
-    absl::AnyInvocable<void(const ByteArray&)> on_inbox_message_callback_;
-    absl::AnyInvocable<void(bool)> on_complete_callback_;
-
-    absl::Mutex mutex_;
-    bool is_receiving_messages_ ABSL_GUARDED_BY(mutex_) = false;
-  };
-
-  bool SendMessage(absl::string_view peer_id,
-                   const ByteArray& message) override;
-
-  bool StartReceivingMessages(
-      OnSignalingMessageCallback on_message_callback,
-      OnSignalingCompleteCallback on_complete_callback) override;
-
-  void StopReceivingMessages() override;
-
- private:
-  std::string self_id_;
-  location::nearby::connections::LocationHint location_hint_;
-  std::unique_ptr<google::internal::communications::instantmessaging::v1::grpc::
-                      Messaging::StubInterface>
-      messaging_stub_;
-  nearby::sharing::AccountManager* const account_manager_;
-  std::shared_ptr<ReceiveMessagesReader> reader_ = nullptr;
-};
-
-}  // namespace nearby
-
-#endif  // #ifndef NO_WEBRTC
-
-#endif  // THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_TACHYON_MESSAGING_CLIENT_H_