diff --git a/connections/implementation/mediums/ble/ble_socket.cc b/connections/implementation/mediums/ble/ble_socket.cc index 1b82eec3..51c0bd37 100644 --- a/connections/implementation/mediums/ble/ble_socket.cc +++ b/connections/implementation/mediums/ble/ble_socket.cc @@ -23,16 +23,19 @@ #include "absl/strings/escaping.h" #include "absl/strings/str_cat.h" #include "connections/implementation/flags/nearby_connections_feature_flags.h" +#include "connections/implementation/mediums/ble/ble_l2cap_packet.h" #include "connections/implementation/mediums/ble/ble_packet.h" #include "internal/flags/nearby_flags.h" #include "internal/platform/ble.h" #include "internal/platform/byte_array.h" #include "internal/platform/byte_utils.h" +#include "internal/platform/count_down_latch.h" #include "internal/platform/exception.h" #include "internal/platform/input_stream.h" #include "internal/platform/logging.h" #include "internal/platform/mutex_lock.h" #include "internal/platform/output_stream.h" +#include "internal/platform/runnable.h" namespace nearby { namespace connections { @@ -76,6 +79,10 @@ Exception BleOutputStream::Flush() { return source_.Flush(); } Exception BleOutputStream::Close() { return source_.Close(); } +Exception BleOutputStream::WriteControlPacket(const ByteArray& data) { + return source_.Write(data); +} + Exception BleOutputStream::WritePayloadLength(int payload_length) { if (payload_length_ != 0) { return {Exception::kFailed}; @@ -150,6 +157,7 @@ Exception BleSocket::CloseLocked() { LOG(FATAL) << "Socket close on unknown medium."; break; } + serial_executor_.Shutdown(); return {Exception::kIo}; } @@ -228,18 +236,25 @@ ExceptionOr BleSocket::DispatchPacket() { } ExceptionOr BleSocket::ReadPayloadLength() { - MutexLock lock(&mutex_); - if (!ble_input_stream_) { - return {Exception::kIo}; - } + int payload_length = 0; + { + MutexLock lock(&mutex_); + if (!ble_input_stream_) { + return {Exception::kIo}; + } - ExceptionOr read_bytes = - ble_input_stream_->Read(sizeof(std::int32_t)); - if (!read_bytes.ok()) { - return read_bytes.exception(); - } + ExceptionOr read_bytes = + ble_input_stream_->Read(sizeof(std::int32_t)); + if (!read_bytes.ok()) { + return read_bytes.exception(); + } - int payload_length = byte_utils::BytesToInt(std::move(read_bytes.result())); + payload_length = byte_utils::BytesToInt(std::move(read_bytes.result())); + } + Exception send_ack_result = SendPacketAcknowledgement(payload_length); + if (!send_ack_result.Ok()) { + LOG(WARNING) << "Failed to send packet acknowledgement."; + } return ExceptionOr(payload_length); } @@ -301,28 +316,186 @@ ExceptionOr BleSocket::ProcessBleControlPacketLocked() { } Exception BleSocket::SendIntroduction() { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + Exception result = {Exception::kFailed}; + CountDownLatch latch(1); + RunOnSocketThread([this, &latch, &result]() { + MutexLock lock(&mutex_); + if (!ble_output_stream_) { + latch.CountDown(); + return; + } + + absl::StatusOr ble_packet_status_or = + BlePacket::CreateControlIntroductionPacket(service_id_hash_); + if (!ble_packet_status_or.ok()) { + LOG(WARNING) << "Failed to create BLE introduction packet: " + << ble_packet_status_or.status(); + latch.CountDown(); + return; + } + result = ble_output_stream_->WriteControlPacket( + ByteArray(ble_packet_status_or.value())); + if (!result.Ok()) { + LOG(WARNING) << "Failed to write BLE introduction packet: " + << result.value; + } + latch.CountDown(); + }); + latch.Await(); + return result; } Exception BleSocket::SendDisconnection() { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + Exception result = {Exception::kFailed}; + CountDownLatch latch(1); + RunOnSocketThread([this, &latch, &result]() { + MutexLock lock(&mutex_); + if (!ble_output_stream_) { + latch.CountDown(); + return; + } + + absl::StatusOr ble_packet_status_or = + BlePacket::CreateControlDisconnectionPacket(service_id_hash_); + if (!ble_packet_status_or.ok()) { + LOG(WARNING) << "Failed to create BLE control disconnection packet: " + << ble_packet_status_or.status(); + latch.CountDown(); + return; + } + result = ble_output_stream_->WriteControlPacket( + ByteArray(ble_packet_status_or.value())); + if (!result.Ok()) { + LOG(WARNING) << "Failed to write BLE control disconnection packet: " + << result.value; + } + latch.CountDown(); + }); + latch.Await(); + return result; } Exception BleSocket::SendPacketAcknowledgement(int received_size) { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + Exception result = {Exception::kFailed}; + CountDownLatch latch(1); + RunOnSocketThread([this, &latch, &result, &received_size]() { + MutexLock lock(&mutex_); + if (!ble_output_stream_) { + latch.CountDown(); + return; + } + + absl::StatusOr ble_packet_status_or = + BlePacket::CreateControlPacketAcknowledgementPacket(service_id_hash_, + received_size); + if (!ble_packet_status_or.ok()) { + LOG(WARNING) + << "Failed to create BLE control packet acknowledgement packet: " + << ble_packet_status_or.status(); + latch.CountDown(); + return; + } + result = ble_output_stream_->WriteControlPacket( + ByteArray(ble_packet_status_or.value())); + if (!result.Ok()) { + LOG(WARNING) + << "Failed to write BLE control packet acknowledgement packet: " + << result.value; + } + latch.CountDown(); + }); + latch.Await(); + return result; } Exception BleSocket::ProcessIncomingL2capPacketValidation() { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + MutexLock lock(&mutex_); + if (!ble_input_stream_) { + return {Exception::kFailed}; + } + + absl::StatusOr ble_l2cap_packet_status_or = + BleL2capPacket::CreateFromStream(*ble_input_stream_); + if (!ble_l2cap_packet_status_or.ok()) { + LOG(WARNING) << "Failed to create BleL2capPacket: " + << ble_l2cap_packet_status_or.status(); + return {Exception::kFailed}; + } + + // Make sure the packet is a data connection request. + BleL2capPacket ble_l2cap_packet = ble_l2cap_packet_status_or.value(); + if (!ble_l2cap_packet.IsDataConnectionRequest()) { + LOG(WARNING) + << "Received an L2CAP packet that is not a data connection request."; + return {Exception::kFailed}; + } + + // Send out the Command::kResponseDataConnectionReady packet. + Exception result = + SendL2capPacketLocked(BleL2capPacket::ByteArrayForDataConnectionReady()); + if (!result.Ok()) { + LOG(WARNING) + << "Failed to send L2CAP data connection ready response packet: " + << result.value; + return result; + } + + return result; } Exception BleSocket::ProcessOutgoingL2capPacketValidation() { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + MutexLock lock(&mutex_); + if (!ble_input_stream_) { + return {Exception::kFailed}; + } + + // Send out the Command::kRequestDataConnection packet. + Exception result = SendL2capPacketLocked( + BleL2capPacket::ByteArrayForRequestDataConnection()); + if (!result.Ok()) { + LOG(WARNING) << "Failed to send L2CAP request data connection packet: " + << result.value; + return result; + } + + // Wait here for the Command::kResponseDataConnectionReady packet. + absl::StatusOr ble_l2cap_packet_status_or = + BleL2capPacket::CreateFromStream(*ble_input_stream_); + if (!ble_l2cap_packet_status_or.ok()) { + LOG(WARNING) << "Failed to create BleL2capPacket: " + << ble_l2cap_packet_status_or.status(); + return {Exception::kFailed}; + } + BleL2capPacket ble_l2cap_packet = ble_l2cap_packet_status_or.value(); + if (!ble_l2cap_packet.IsDataConnectionReadyResponse()) { + LOG(WARNING) << "Unexpected L2CAP packet received."; + return {Exception::kFailed}; + } + return {Exception::kSuccess}; +} + +Exception BleSocket::SendL2capPacketLocked(const ByteArray& packet_byte) { + if (!ble_output_stream_) { + return {Exception::kFailed}; + } + + Exception result = {Exception::kFailed}; + CountDownLatch latch(1); + RunOnSocketThread([this, &packet_byte, &latch, &result]() { + mutex_.AssertHeld(); + + result = ble_output_stream_->WriteControlPacket(packet_byte); + if (!result.Ok()) { + LOG(WARNING) << "Failed to write L2CAP packet: " << result.value; + } + latch.CountDown(); + }); + latch.Await(); + return result; +} + +void BleSocket::RunOnSocketThread(Runnable runnable) { + serial_executor_.Execute(std::move(runnable)); } } // namespace mediums diff --git a/connections/implementation/mediums/ble/ble_socket.h b/connections/implementation/mediums/ble/ble_socket.h index 38616e87..728b4f28 100644 --- a/connections/implementation/mediums/ble/ble_socket.h +++ b/connections/implementation/mediums/ble/ble_socket.h @@ -26,6 +26,8 @@ #include "internal/platform/input_stream.h" #include "internal/platform/mutex.h" #include "internal/platform/output_stream.h" +#include "internal/platform/runnable.h" +#include "internal/platform/single_thread_executor.h" namespace nearby { namespace connections { @@ -105,6 +107,25 @@ class BleOutputStream : public OutputStream { Exception Flush() override; Exception Close() override; + /** + * Sends a protocol-level control packet over the BLE socket. + * + * Control packets are used for managing the connection state, such as sending + * introduction frames, acknowledgements, or disconnection messages, as + * distinct from bulk payload data transfer. + * + * This method is a decorated API on the `BleOutputStream`. It ensures the + * packet is formatted according to the Nearby Connections protocol before + * being written to the underlying stream. This includes automatically + * prepending the 3-byte `service_id_hash` to the packet data. + * + * @param data The `ByteArray` containing the raw content of the control + * packet to be sent. + * @return An `Exception` object indicating the status of the write + * operation. `{Exception::kSuccess}` on success. + */ + Exception WriteControlPacket(const ByteArray& data); + /** * Sends the length of a data payload to the remote endpoint. * @@ -356,11 +377,34 @@ class BleSocket final { ExceptionOr ProcessBleControlPacketLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + /** + * Sends a raw L2CAP packet over the socket. + * + * This function provides a direct, low-level interface for transmitting a + * pre-formatted L2CAP packet. It is intended for use when the caller has + * already constructed a complete L2CAP packet (e.g., for a specific command + * or data payload) and needs to send it directly. + * + * This method bypasses the higher-level handshake logic encapsulated in + * `ProcessIncomingL2capPacketValidation` and + * `ProcessOutgoingL2capPacketValidation`. + * + * @param packet_byte A `ByteArray` containing the raw, complete L2CAP + * packet to be sent over the socket. + * @return An `Exception` object indicating the status of the write + * operation. `{Exception::kSuccess}` on success. + */ + Exception SendL2capPacketLocked(const ByteArray& packet_byte) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + void RunOnSocketThread(Runnable runnable); ::location::nearby::proto::connections::Medium GetMediumLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); Exception CloseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); mutable Mutex mutex_; + SingleThreadExecutor serial_executor_; + const ByteArray service_id_hash_; std::unique_ptr ble_input_stream_ ABSL_GUARDED_BY(mutex_) = nullptr; diff --git a/connections/implementation/mediums/ble/ble_socket_test.cc b/connections/implementation/mediums/ble/ble_socket_test.cc index da6ccb45..754a59b0 100644 --- a/connections/implementation/mediums/ble/ble_socket_test.cc +++ b/connections/implementation/mediums/ble/ble_socket_test.cc @@ -25,6 +25,7 @@ #include "absl/strings/str_cat.h" #include "absl/strings/string_view.h" #include "connections/implementation/flags/nearby_connections_feature_flags.h" +#include "connections/implementation/mediums/ble/ble_l2cap_packet.h" #include "connections/implementation/mediums/ble/ble_packet.h" #include "internal/flags/nearby_flags.h" #include "internal/platform/ble.h" @@ -77,6 +78,9 @@ class FakeInputStream : public InputStream { class FakeOutputStream : public OutputStream { public: Exception Write(const ByteArray& data) override { + if (exception_on_write_) { + return {Exception::kIo}; + } absl::StrAppend(&buffer_, std::string(data)); return {Exception::kSuccess}; } @@ -84,9 +88,13 @@ class FakeOutputStream : public OutputStream { Exception Close() override { return {Exception::kSuccess}; } std::string GetPayload() { return buffer_; } void Clear() { buffer_.clear(); } + void SetExceptionOnWrite(bool exception_on_write) { + exception_on_write_ = exception_on_write; + } private: std::string buffer_; + bool exception_on_write_ = false; }; class FakeBleSocketImpl : public api::ble::BleSocket { @@ -287,10 +295,70 @@ TEST_F(BleSocketBleMediumTest, IsValid) { EXPECT_TRUE(socket_->IsValid()); } +TEST_F(BleSocketBleMediumTest, SendIntroductionSuccess) { + EXPECT_TRUE(socket_->SendIntroduction().Ok()); + auto intro_packet = BlePacket::CreateControlIntroductionPacket( + ByteArray(std::string(kServiceIdHash))); + ASSERT_TRUE(intro_packet.ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(ByteArray(*intro_packet))); +} + +TEST_F(BleSocketBleMediumTest, SendIntroductionAfterClose) { + socket_->Close(); + EXPECT_FALSE(socket_->SendIntroduction().Ok()); +} + +TEST_F(BleSocketBleMediumTest, SendIntroductionWriteFails) { + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->SendIntroduction().Ok()); +} + TEST_F(BleSocketBleMediumTest, GetMedium) { EXPECT_EQ(socket_->GetMedium(), Medium::BLE); } +TEST_F(BleSocketBleMediumTest, SendDisconnectionSuccess) { + EXPECT_TRUE(socket_->SendDisconnection().Ok()); + auto disconnection_packet = BlePacket::CreateControlDisconnectionPacket( + ByteArray(std::string(kServiceIdHash))); + ASSERT_TRUE(disconnection_packet.ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(ByteArray(*disconnection_packet))); +} + +TEST_F(BleSocketBleMediumTest, SendDisconnectionAfterClose) { + socket_->Close(); + EXPECT_FALSE(socket_->SendDisconnection().Ok()); +} + +TEST_F(BleSocketBleMediumTest, SendDisconnectionWriteFails) { + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->SendDisconnection().Ok()); +} + +TEST_F(BleSocketBleMediumTest, SendPacketAcknowledgementSuccess) { + constexpr int kReceivedSize = 100; + EXPECT_TRUE(socket_->SendPacketAcknowledgement(kReceivedSize).Ok()); + auto ack_packet = BlePacket::CreateControlPacketAcknowledgementPacket( + ByteArray(std::string(kServiceIdHash)), kReceivedSize); + ASSERT_TRUE(ack_packet.ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(ByteArray(*ack_packet))); +} + +TEST_F(BleSocketBleMediumTest, SendPacketAcknowledgementAfterClose) { + constexpr int kReceivedSize = 100; + socket_->Close(); + EXPECT_FALSE(socket_->SendPacketAcknowledgement(kReceivedSize).Ok()); +} + +TEST_F(BleSocketBleMediumTest, SendPacketAcknowledgementWriteFails) { + constexpr int kReceivedSize = 100; + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->SendPacketAcknowledgement(kReceivedSize).Ok()); +} + class BleL2capSocketBleMediumTest : public ::testing::Test { protected: void SetUp() override { @@ -442,10 +510,138 @@ TEST_F(BleL2capSocketBleMediumTest, IsValid) { EXPECT_TRUE(socket_->IsValid()); } +TEST_F(BleL2capSocketBleMediumTest, SendIntroductionSuccess) { + EXPECT_TRUE(socket_->SendIntroduction().Ok()); + auto intro_packet = BlePacket::CreateControlIntroductionPacket( + ByteArray(std::string(kServiceIdHash))); + ASSERT_TRUE(intro_packet.ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(ByteArray(*intro_packet))); +} + +TEST_F(BleL2capSocketBleMediumTest, SendIntroductionAfterClose) { + socket_->Close(); + EXPECT_FALSE(socket_->SendIntroduction().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, SendIntroductionWriteFails) { + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->SendIntroduction().Ok()); +} + TEST_F(BleL2capSocketBleMediumTest, GetMedium) { EXPECT_EQ(socket_->GetMedium(), Medium::BLE_L2CAP); } +TEST_F(BleL2capSocketBleMediumTest, SendDisconnectionSuccess) { + EXPECT_TRUE(socket_->SendDisconnection().Ok()); + auto disconnection_packet = BlePacket::CreateControlDisconnectionPacket( + ByteArray(std::string(kServiceIdHash))); + ASSERT_TRUE(disconnection_packet.ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(ByteArray(*disconnection_packet))); +} + +TEST_F(BleL2capSocketBleMediumTest, SendDisconnectionAfterClose) { + socket_->Close(); + EXPECT_FALSE(socket_->SendDisconnection().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, SendDisconnectionWriteFails) { + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->SendDisconnection().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, SendPacketAcknowledgementSuccess) { + constexpr int kReceivedSize = 100; + EXPECT_TRUE(socket_->SendPacketAcknowledgement(kReceivedSize).Ok()); + auto ack_packet = BlePacket::CreateControlPacketAcknowledgementPacket( + ByteArray(std::string(kServiceIdHash)), kReceivedSize); + ASSERT_TRUE(ack_packet.ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(ByteArray(*ack_packet))); +} + +TEST_F(BleL2capSocketBleMediumTest, SendPacketAcknowledgementAfterClose) { + constexpr int kReceivedSize = 100; + socket_->Close(); + EXPECT_FALSE(socket_->SendPacketAcknowledgement(kReceivedSize).Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, SendPacketAcknowledgementWriteFails) { + constexpr int kReceivedSize = 100; + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->SendPacketAcknowledgement(kReceivedSize).Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessIncomingL2capPacketValidationSuccess) { + fake_input_stream_.Append( + BleL2capPacket::ByteArrayForRequestDataConnection()); + EXPECT_TRUE(socket_->ProcessIncomingL2capPacketValidation().Ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(BleL2capPacket::ByteArrayForDataConnectionReady())); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessIncomingL2capPacketValidationReadFails) { + fake_input_stream_.SetExceptionOnRead(true); + EXPECT_FALSE(socket_->ProcessIncomingL2capPacketValidation().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessIncomingL2capPacketValidationWrongPacket) { + fake_input_stream_.Append(BleL2capPacket::ByteArrayForDataConnectionReady()); + EXPECT_FALSE(socket_->ProcessIncomingL2capPacketValidation().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessIncomingL2capPacketValidationWriteFails) { + fake_input_stream_.Append( + BleL2capPacket::ByteArrayForRequestDataConnection()); + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->ProcessIncomingL2capPacketValidation().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessIncomingL2capPacketValidationAfterClose) { + socket_->Close(); + EXPECT_FALSE(socket_->ProcessIncomingL2capPacketValidation().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessOutgoingL2capPacketValidationSuccess) { + fake_input_stream_.Append(BleL2capPacket::ByteArrayForDataConnectionReady()); + EXPECT_TRUE(socket_->ProcessOutgoingL2capPacketValidation().Ok()); + EXPECT_EQ(fake_output_stream_.GetPayload(), + std::string(BleL2capPacket::ByteArrayForRequestDataConnection())); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessOutgoingL2capPacketValidationWriteFails) { + fake_output_stream_.SetExceptionOnWrite(true); + EXPECT_FALSE(socket_->ProcessOutgoingL2capPacketValidation().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessOutgoingL2capPacketValidationReadFails) { + fake_input_stream_.SetExceptionOnRead(true); + EXPECT_FALSE(socket_->ProcessOutgoingL2capPacketValidation().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessOutgoingL2capPacketValidationWrongPacket) { + fake_input_stream_.Append( + BleL2capPacket::ByteArrayForRequestDataConnection()); + EXPECT_FALSE(socket_->ProcessOutgoingL2capPacketValidation().Ok()); +} + +TEST_F(BleL2capSocketBleMediumTest, + ProcessOutgoingL2capPacketValidationAfterClose) { + socket_->Close(); + EXPECT_FALSE(socket_->ProcessOutgoingL2capPacketValidation().Ok()); +} + } // namespace } // namespace mediums } // namespace connections diff --git a/connections/implementation/mediums/ble_test.cc b/connections/implementation/mediums/ble_test.cc index 13f80d64..fd935e7b 100644 --- a/connections/implementation/mediums/ble_test.cc +++ b/connections/implementation/mediums/ble_test.cc @@ -190,8 +190,11 @@ TEST_P(BleTest, CanConnect2) { EXPECT_TRUE(ble_server.StopAdvertising(service_id)); ASSERT_NE(socket_for_server, nullptr); EXPECT_TRUE(socket_for_server.get()->IsValid()); - ASSERT_FALSE(socket_for_client_result.has_value()); - // TODO: edwinwu - add more tests. + ASSERT_TRUE(socket_for_client_result.has_value()); + EXPECT_TRUE(socket_for_client_result.value()->IsValid()); + EXPECT_TRUE(socket_for_server->GetRemotePeripheral().IsValid()); + EXPECT_TRUE( + socket_for_client_result.value()->GetRemotePeripheral().IsValid()); env_.Stop(); } @@ -318,8 +321,10 @@ TEST_P(BleTest, CanCancelConnect2) { EXPECT_TRUE(ble_server.StopAdvertising(service_id)); ASSERT_NE(socket_for_server, nullptr); EXPECT_TRUE(socket_for_server.get()->IsValid()); - // TODO: edwinwu - add more tests. + EXPECT_TRUE(socket_for_client_result.has_value()); EXPECT_TRUE(socket_for_server.get()->GetRemotePeripheral().IsValid()); + EXPECT_TRUE( + socket_for_client_result.value()->GetRemotePeripheral().IsValid()); } else { EXPECT_FALSE(accept_latch.Await(kWaitDuration).result()); EXPECT_TRUE(ble_server.StopAcceptingConnections(service_id));