Make the emulated VHAL always listen for socket connections.
Previously, when running under the emulator, it would only connect to
the emulator's qemu pipe and you couldn't script the emulated VHAL.
Also changed the logic of SocketComm so that it supports connections
from more than one client at a time.
Bug: 87643732
Test: tested locally
Change-Id: I504cd806f0a95799a68b75c3515c1f230109f1d0
(cherry picked from commit 8dad67a0148114cb60f3567268bc2cf40ff18f82)
diff --git a/automotive/vehicle/2.0/default/Android.bp b/automotive/vehicle/2.0/default/Android.bp
index 65e9133..7802ef0 100644
--- a/automotive/vehicle/2.0/default/Android.bp
+++ b/automotive/vehicle/2.0/default/Android.bp
@@ -58,6 +58,7 @@
vendor: true,
defaults: ["vhal_v2_0_defaults"],
srcs: [
+ "impl/vhal_v2_0/CommConn.cpp",
"impl/vhal_v2_0/EmulatedVehicleHal.cpp",
"impl/vhal_v2_0/VehicleEmulator.cpp",
"impl/vhal_v2_0/PipeComm.cpp",
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommBase.h b/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommBase.h
deleted file mode 100644
index 6832ad3..0000000
--- a/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommBase.h
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Copyright (C) 2017 The Android Open Source Project
- *
- * 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
- *
- * http://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 android_hardware_automotive_vehicle_V2_0_impl_CommBase_H_
-#define android_hardware_automotive_vehicle_V2_0_impl_CommBase_H_
-
-#include <string>
-#include <vector>
-
-namespace android {
-namespace hardware {
-namespace automotive {
-namespace vehicle {
-namespace V2_0 {
-
-namespace impl {
-
-/**
- * This is the communications base class. It defines the interface used in DefaultVehicleHal to
- * send and receive data to and from the emulator.
- */
-class CommBase {
-public:
- virtual ~CommBase() = default;
-
- /**
- * Closes a connection if it is open.
- */
- virtual void stop() {}
-
- /**
- * Creates a connection to the other side.
- *
- * @return int Returns fd or socket number if connection is successful.
- * Otherwise, returns -1 if no connection is availble.
- */
- virtual int connect() { return 0; }
-
- /**
- * Opens the communications channel.
- *
- * @return int Returns 0 if channel is opened, else -errno if failed.
- */
- virtual int open() = 0;
-
- /**
- * Blocking call to read data from the connection.
- *
- * @return std::vector<uint8_t> Serialized protobuf data received from emulator. This will be
- * an empty vector if the connection was closed or some other error occurred.
- */
- virtual std::vector<uint8_t> read() = 0;
-
- /**
- * Transmits a string of data to the emulator.
- *
- * @param data Serialized protobuf data to transmit.
- *
- * @return int Number of bytes transmitted, or -1 if failed.
- */
- virtual int write(const std::vector<uint8_t>& data) = 0;
-};
-
-} // impl
-
-} // namespace V2_0
-} // namespace vehicle
-} // namespace automotive
-} // namespace hardware
-} // namespace android
-
-
-#endif // android_hardware_automotive_vehicle_V2_0_impl_CommBase_H_
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommConn.cpp b/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommConn.cpp
new file mode 100644
index 0000000..bf1de81
--- /dev/null
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommConn.cpp
@@ -0,0 +1,80 @@
+/*
+ * Copyright (C) 2018 The Android Open Source Project
+ *
+ * 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
+ *
+ * http://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.
+ */
+
+#define LOG_TAG "CommConn"
+
+#include <thread>
+
+#include <android/hardware/automotive/vehicle/2.0/IVehicle.h>
+#include <log/log.h>
+
+#include "CommConn.h"
+
+namespace android {
+namespace hardware {
+namespace automotive {
+namespace vehicle {
+namespace V2_0 {
+
+namespace impl {
+
+void CommConn::start() {
+ mReadThread = std::make_unique<std::thread>(std::bind(&CommConn::readThread, this));
+}
+
+void CommConn::stop() {
+ if (mReadThread->joinable()) {
+ mReadThread->join();
+ }
+}
+
+void CommConn::sendMessage(emulator::EmulatorMessage const& msg) {
+ int numBytes = msg.ByteSize();
+ std::vector<uint8_t> buffer(static_cast<size_t>(numBytes));
+ if (!msg.SerializeToArray(buffer.data(), numBytes)) {
+ ALOGE("%s: SerializeToString failed!", __func__);
+ return;
+ }
+
+ write(buffer);
+}
+
+void CommConn::readThread() {
+ std::vector<uint8_t> buffer;
+ while (isOpen()) {
+ buffer = read();
+ if (buffer.size() == 0) {
+ ALOGI("%s: Read returned empty message, exiting read loop.", __func__);
+ break;
+ }
+
+ emulator::EmulatorMessage rxMsg;
+ if (rxMsg.ParseFromArray(buffer.data(), static_cast<int32_t>(buffer.size()))) {
+ emulator::EmulatorMessage respMsg;
+ mMessageProcessor->processMessage(rxMsg, respMsg);
+
+ sendMessage(respMsg);
+ }
+ }
+}
+
+} // namespace impl
+
+} // namespace V2_0
+} // namespace vehicle
+} // namespace automotive
+} // namespace hardware
+} // namespace android
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommConn.h b/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommConn.h
new file mode 100644
index 0000000..87b0dfc
--- /dev/null
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/CommConn.h
@@ -0,0 +1,117 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * 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
+ *
+ * http://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 android_hardware_automotive_vehicle_V2_0_impl_CommBase_H_
+#define android_hardware_automotive_vehicle_V2_0_impl_CommBase_H_
+
+#include <android/hardware/automotive/vehicle/2.0/IVehicle.h>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "VehicleHalProto.pb.h"
+
+namespace android {
+namespace hardware {
+namespace automotive {
+namespace vehicle {
+namespace V2_0 {
+
+namespace impl {
+
+/**
+ * MessageProcess is an interface implemented by VehicleEmulator to process messages received
+ * over a CommConn.
+ */
+class MessageProcessor {
+ public:
+ virtual ~MessageProcessor() = default;
+
+ /**
+ * Process a single message received over a CommConn. Populate the given respMsg with the reply
+ * message we should send.
+ */
+ virtual void processMessage(emulator::EmulatorMessage const& rxMsg,
+ emulator::EmulatorMessage& respMsg) = 0;
+};
+
+/**
+ * This is the interface that both PipeComm and SocketComm use to represent a connection. The
+ * connection will listen for commands on a separate 'read' thread.
+ */
+class CommConn {
+ public:
+ CommConn(MessageProcessor* messageProcessor) : mMessageProcessor(messageProcessor) {}
+
+ virtual ~CommConn() {}
+
+ /**
+ * Start the read thread reading messages from this connection.
+ */
+ virtual void start();
+
+ /**
+ * Closes a connection if it is open.
+ */
+ virtual void stop();
+
+ /**
+ * Returns true if the connection is open and available to send/receive.
+ */
+ virtual bool isOpen() = 0;
+
+ /**
+ * Blocking call to read data from the connection.
+ *
+ * @return std::vector<uint8_t> Serialized protobuf data received from emulator. This will be
+ * an empty vector if the connection was closed or some other error occurred.
+ */
+ virtual std::vector<uint8_t> read() = 0;
+
+ /**
+ * Transmits a string of data to the emulator.
+ *
+ * @param data Serialized protobuf data to transmit.
+ *
+ * @return int Number of bytes transmitted, or -1 if failed.
+ */
+ virtual int write(const std::vector<uint8_t>& data) = 0;
+
+ /**
+ * Serialized and send the given message to the other side.
+ */
+ void sendMessage(emulator::EmulatorMessage const& msg);
+
+ protected:
+ std::unique_ptr<std::thread> mReadThread;
+ MessageProcessor* mMessageProcessor;
+
+ /**
+ * A thread that reads messages in a loop, and responds. You can stop this thread by calling
+ * stop().
+ */
+ void readThread();
+};
+
+} // namespace impl
+
+} // namespace V2_0
+} // namespace vehicle
+} // namespace automotive
+} // namespace hardware
+} // namespace android
+
+#endif // android_hardware_automotive_vehicle_V2_0_impl_CommBase_H_
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.cpp b/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.cpp
index 5a9b254..f024287 100644
--- a/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.cpp
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.cpp
@@ -33,23 +33,28 @@
namespace impl {
-PipeComm::PipeComm() {
- // Initialize member vars
- mPipeFd = -1;
-}
+PipeComm::PipeComm(MessageProcessor* messageProcessor) : CommConn(messageProcessor), mPipeFd(-1) {}
-
-int PipeComm::open() {
+void PipeComm::start() {
int fd = qemu_pipe_open(CAR_SERVICE_NAME);
if (fd < 0) {
ALOGE("%s: Could not open connection to service: %s %d", __FUNCTION__, strerror(errno), fd);
- return -errno;
+ return;
}
- ALOGI("%s: OPENED PIPE, fd=%d", __FUNCTION__, fd);
+ ALOGI("%s: Starting pipe connection, fd=%d", __FUNCTION__, fd);
mPipeFd = fd;
- return 0;
+
+ CommConn::start();
+}
+
+void PipeComm::stop() {
+ if (mPipeFd > 0) {
+ ::close(mPipeFd);
+ mPipeFd = -1;
+ }
+ CommConn::stop();
}
std::vector<uint8_t> PipeComm::read() {
@@ -60,16 +65,13 @@
numBytes = qemu_pipe_frame_recv(mPipeFd, msg.data(), msg.size());
if (numBytes == MAX_RX_MSG_SZ) {
- ALOGE("%s: Received max size = %d", __FUNCTION__, MAX_RX_MSG_SZ);
+ ALOGE("%s: Received max size = %d", __FUNCTION__, MAX_RX_MSG_SZ);
} else if (numBytes > 0) {
msg.resize(numBytes);
return msg;
} else {
ALOGD("%s: Connection terminated on pipe %d, numBytes=%d", __FUNCTION__, mPipeFd, numBytes);
- {
- std::lock_guard<std::mutex> lock(mMutex);
- mPipeFd = -1;
- }
+ mPipeFd = -1;
}
return std::vector<uint8_t>();
@@ -78,11 +80,8 @@
int PipeComm::write(const std::vector<uint8_t>& data) {
int retVal = 0;
- {
- std::lock_guard<std::mutex> lock(mMutex);
- if (mPipeFd != -1) {
- retVal = qemu_pipe_frame_send(mPipeFd, data.data(), data.size());
- }
+ if (mPipeFd != -1) {
+ retVal = qemu_pipe_frame_send(mPipeFd, data.data(), data.size());
}
if (retVal < 0) {
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.h b/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.h
index bcd32d0..c8eabb8 100644
--- a/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.h
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/PipeComm.h
@@ -19,7 +19,7 @@
#include <mutex>
#include <vector>
-#include "CommBase.h"
+#include "CommConn.h"
namespace android {
namespace hardware {
@@ -30,38 +30,25 @@
namespace impl {
/**
- * PipeComm uses a qemu pipe interface to connect to the Goldfish Emulator.
+ * PipeComm opens a qemu pipe to connect to the emulator, allowing the emulator UI to access the
+ * Vehicle HAL and simulate changing properties.
+ *
+ * Since the pipe is a client, it directly implements CommConn, and only one PipeComm can be open
+ * at a time.
*/
-class PipeComm : public CommBase {
-public:
- PipeComm();
+class PipeComm : public CommConn {
+ public:
+ PipeComm(MessageProcessor* messageProcessor);
- /**
- * Opens a pipe and begins listening.
- *
- * @return int Returns 0 on success.
- */
- int open() override;
+ void start() override;
+ void stop() override;
- /**
- * Blocking call to read data from the connection.
- *
- * @return std::vector<uint8_t> Serialized protobuf data received from emulator. This will be
- * an empty vector if the connection was closed or some other error occurred.
- */
std::vector<uint8_t> read() override;
-
- /**
- * Transmits a string of data to the emulator.
- *
- * @param data Serialized protobuf data to transmit.
- *
- * @return int Number of bytes transmitted, or -1 if failed.
- */
int write(const std::vector<uint8_t>& data) override;
-private:
- std::mutex mMutex;
+ inline bool isOpen() override { return mPipeFd > 0; }
+
+ private:
int mPipeFd;
};
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.cpp b/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.cpp
index 42c1c78..9eb8894 100644
--- a/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.cpp
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.cpp
@@ -18,6 +18,7 @@
#include <android/hardware/automotive/vehicle/2.0/IVehicle.h>
#include <android/log.h>
+#include <arpa/inet.h>
#include <log/log.h>
#include <netinet/in.h>
#include <sys/socket.h>
@@ -35,45 +36,46 @@
namespace impl {
-SocketComm::SocketComm() {
- // Initialize member vars
- mCurSockFd = -1;
- mExit = 0;
- mSockFd = -1;
-}
-
+SocketComm::SocketComm(MessageProcessor* messageProcessor)
+ : mListenFd(-1), mMessageProcessor(messageProcessor) {}
SocketComm::~SocketComm() {
- stop();
}
-int SocketComm::connect() {
- sockaddr_in cliAddr;
- socklen_t cliLen = sizeof(cliAddr);
- int cSockFd = accept(mSockFd, reinterpret_cast<struct sockaddr*>(&cliAddr), &cliLen);
-
- if (cSockFd >= 0) {
- {
- std::lock_guard<std::mutex> lock(mMutex);
- mCurSockFd = cSockFd;
- }
- ALOGD("%s: Incoming connection received on socket %d", __FUNCTION__, cSockFd);
- } else {
- cSockFd = -1;
+void SocketComm::start() {
+ if (!listen()) {
+ return;
}
- return cSockFd;
+ mListenThread = std::make_unique<std::thread>(std::bind(&SocketComm::listenThread, this));
}
-int SocketComm::open() {
+void SocketComm::stop() {
+ if (mListenFd > 0) {
+ ::close(mListenFd);
+ if (mListenThread->joinable()) {
+ mListenThread->join();
+ }
+ mListenFd = -1;
+ }
+}
+
+void SocketComm::sendMessage(emulator::EmulatorMessage const& msg) {
+ std::lock_guard<std::mutex> lock(mMutex);
+ for (std::unique_ptr<SocketConn> const& conn : mOpenConnections) {
+ conn->sendMessage(msg);
+ }
+}
+
+bool SocketComm::listen() {
int retVal;
struct sockaddr_in servAddr;
- mSockFd = socket(AF_INET, SOCK_STREAM, 0);
- if (mSockFd < 0) {
- ALOGE("%s: socket() failed, mSockFd=%d, errno=%d", __FUNCTION__, mSockFd, errno);
- mSockFd = -1;
- return -errno;
+ mListenFd = socket(AF_INET, SOCK_STREAM, 0);
+ if (mListenFd < 0) {
+ ALOGE("%s: socket() failed, mSockFd=%d, errno=%d", __FUNCTION__, mListenFd, errno);
+ mListenFd = -1;
+ return false;
}
memset(&servAddr, 0, sizeof(servAddr));
@@ -81,82 +83,114 @@
servAddr.sin_addr.s_addr = INADDR_ANY;
servAddr.sin_port = htons(DEBUG_SOCKET);
- retVal = bind(mSockFd, reinterpret_cast<struct sockaddr*>(&servAddr), sizeof(servAddr));
+ retVal = bind(mListenFd, reinterpret_cast<struct sockaddr*>(&servAddr), sizeof(servAddr));
if(retVal < 0) {
ALOGE("%s: Error on binding: retVal=%d, errno=%d", __FUNCTION__, retVal, errno);
+ close(mListenFd);
+ mListenFd = -1;
+ return false;
+ }
+
+ ALOGI("%s: Listening for connections on port %d", __FUNCTION__, DEBUG_SOCKET);
+ ::listen(mListenFd, 1);
+ return true;
+}
+
+SocketConn* SocketComm::accept() {
+ sockaddr_in cliAddr;
+ socklen_t cliLen = sizeof(cliAddr);
+ int sfd = ::accept(mListenFd, reinterpret_cast<struct sockaddr*>(&cliAddr), &cliLen);
+
+ if (sfd > 0) {
+ char addr[INET_ADDRSTRLEN];
+ inet_ntop(AF_INET, &cliAddr.sin_addr, addr, INET_ADDRSTRLEN);
+
+ ALOGD("%s: Incoming connection received from %s:%d", __FUNCTION__, addr, cliAddr.sin_port);
+ return new SocketConn(mMessageProcessor, sfd);
+ }
+
+ return nullptr;
+}
+
+void SocketComm::listenThread() {
+ while (true) {
+ SocketConn* conn = accept();
+ if (conn == nullptr) {
+ return;
+ }
+
+ conn->start();
+ {
+ std::lock_guard<std::mutex> lock(mMutex);
+ mOpenConnections.push_back(std::unique_ptr<SocketConn>(conn));
+ }
+ }
+}
+
+/**
+ * Called occasionally to clean up connections that have been closed.
+ */
+void SocketComm::removeClosedConnections() {
+ std::lock_guard<std::mutex> lock(mMutex);
+ std::remove_if(mOpenConnections.begin(), mOpenConnections.end(),
+ [](std::unique_ptr<SocketConn> const& c) { return !c->isOpen(); });
+}
+
+SocketConn::SocketConn(MessageProcessor* messageProcessor, int sfd)
+ : CommConn(messageProcessor), mSockFd(sfd) {}
+
+/**
+ * Reads, in a loop, exactly numBytes from the given fd. If the connection is closed, returns
+ * an empty buffer, otherwise will return exactly the given number of bytes.
+ */
+std::vector<uint8_t> readExactly(int fd, int numBytes) {
+ std::vector<uint8_t> buffer(numBytes);
+ int totalRead = 0;
+ int offset = 0;
+ while (totalRead < numBytes) {
+ int numRead = ::read(fd, &buffer.data()[offset], numBytes - offset);
+ if (numRead == 0) {
+ buffer.resize(0);
+ return buffer;
+ }
+
+ totalRead += numRead;
+ }
+ return buffer;
+}
+
+/**
+ * Reads an int, guaranteed to be non-zero, from the given fd. If the connection is closed, returns
+ * -1.
+ */
+int32_t readInt(int fd) {
+ std::vector<uint8_t> buffer = readExactly(fd, sizeof(int32_t));
+ if (buffer.size() == 0) {
+ return -1;
+ }
+
+ int32_t value = *reinterpret_cast<int32_t*>(buffer.data());
+ return ntohl(value);
+}
+
+std::vector<uint8_t> SocketConn::read() {
+ int32_t msgSize = readInt(mSockFd);
+ if (msgSize <= 0) {
+ ALOGD("%s: Connection terminated on socket %d", __FUNCTION__, mSockFd);
+ return std::vector<uint8_t>();
+ }
+
+ return readExactly(mSockFd, msgSize);
+}
+
+void SocketConn::stop() {
+ if (mSockFd > 0) {
close(mSockFd);
mSockFd = -1;
- return -errno;
- }
-
- listen(mSockFd, 1);
-
- // Set the socket to be non-blocking so we can poll it continouously
- fcntl(mSockFd, F_SETFL, O_NONBLOCK);
-
- return 0;
-}
-
-std::vector<uint8_t> SocketComm::read() {
- int32_t msgSize;
- int numBytes = 0;
-
- // This is a variable length message.
- // Read the number of bytes to rx over the socket
- numBytes = ::read(mCurSockFd, &msgSize, sizeof(msgSize));
- msgSize = ntohl(msgSize);
-
- if (numBytes != sizeof(msgSize)) {
- // This happens when connection is closed
- ALOGD("%s: numBytes=%d, expected=4", __FUNCTION__, numBytes);
- ALOGD("%s: Connection terminated on socket %d", __FUNCTION__, mCurSockFd);
- {
- std::lock_guard<std::mutex> lock(mMutex);
- mCurSockFd = -1;
- }
-
- return std::vector<uint8_t>();
- }
-
- std::vector<uint8_t> msg = std::vector<uint8_t>(msgSize);
-
- numBytes = ::read(mCurSockFd, msg.data(), msgSize);
-
- if ((numBytes == msgSize) && (msgSize > 0)) {
- // Received a message.
- return msg;
- } else {
- // This happens when connection is closed
- ALOGD("%s: numBytes=%d, msgSize=%d", __FUNCTION__, numBytes, msgSize);
- ALOGD("%s: Connection terminated on socket %d", __FUNCTION__, mCurSockFd);
- {
- std::lock_guard<std::mutex> lock(mMutex);
- mCurSockFd = -1;
- }
-
- return std::vector<uint8_t>();
}
}
-void SocketComm::stop() {
- if (mExit == 0) {
- std::lock_guard<std::mutex> lock(mMutex);
- mExit = 1;
-
- // Close emulator socket if it is open
- if (mCurSockFd != -1) {
- close(mCurSockFd);
- mCurSockFd = -1;
- }
-
- if (mSockFd != -1) {
- close(mSockFd);
- mSockFd = -1;
- }
- }
-}
-
-int SocketComm::write(const std::vector<uint8_t>& data) {
+int SocketConn::write(const std::vector<uint8_t>& data) {
static constexpr int MSG_HEADER_LEN = 4;
int retVal = 0;
union {
@@ -168,19 +202,17 @@
msgLen = static_cast<uint32_t>(data.size());
msgLen = htonl(msgLen);
- std::lock_guard<std::mutex> lock(mMutex);
- if (mCurSockFd != -1) {
- retVal = ::write(mCurSockFd, msgLenBytes, MSG_HEADER_LEN);
+ if (mSockFd > 0) {
+ retVal = ::write(mSockFd, msgLenBytes, MSG_HEADER_LEN);
if (retVal == MSG_HEADER_LEN) {
- retVal = ::write(mCurSockFd, data.data(), data.size());
+ retVal = ::write(mSockFd, data.data(), data.size());
}
}
return retVal;
}
-
} // impl
} // namespace V2_0
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.h b/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.h
index 12cfb29..88b852b 100644
--- a/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.h
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/SocketComm.h
@@ -18,8 +18,9 @@
#define android_hardware_automotive_vehicle_V2_0_impl_SocketComm_H_
#include <mutex>
+#include <thread>
#include <vector>
-#include "CommBase.h"
+#include "CommConn.h"
namespace android {
namespace hardware {
@@ -29,29 +30,60 @@
namespace impl {
+class SocketConn;
+
/**
- * SocketComm opens a socket via adb's TCP port forwarding to enable a Host PC to connect to
- * the VehicleHAL.
+ * SocketComm opens a socket, and listens for connections from clients. Typically the client will be
+ * adb's TCP port-forwarding to enable a host PC to connect to the VehicleHAL.
*/
-class SocketComm : public CommBase {
-public:
- SocketComm();
+class SocketComm {
+ public:
+ SocketComm(MessageProcessor* messageProcessor);
virtual ~SocketComm();
+ void start();
+ void stop();
+
/**
- * Creates a connection to the other side.
+ * Serialized and send the given message to all connected clients.
+ */
+ void sendMessage(emulator::EmulatorMessage const& msg);
+
+ private:
+ int mListenFd;
+ std::unique_ptr<std::thread> mListenThread;
+ std::vector<std::unique_ptr<SocketConn>> mOpenConnections;
+ MessageProcessor* mMessageProcessor;
+ std::mutex mMutex;
+
+ /**
+ * Opens the socket and begins listening.
+ *
+ * @return bool Returns true on success.
+ */
+ bool listen();
+
+ /**
+ * Blocks and waits for a connection from a client, returns a new SocketConn with the connection
+ * or null, if the connection has been closed.
*
* @return int Returns fd or socket number if connection is successful.
* Otherwise, returns -1 if no connection is availble.
*/
- int connect() override;
+ SocketConn* accept();
- /**
- * Opens a socket and begins listening.
- *
- * @return int Returns 0 on success.
- */
- int open() override;
+ void listenThread();
+
+ void removeClosedConnections();
+};
+
+/**
+ * SocketConn represents a single connection to a client.
+ */
+class SocketConn : public CommConn {
+ public:
+ SocketConn(MessageProcessor* messageProcessor, int sfd);
+ virtual ~SocketConn() = default;
/**
* Blocking call to read data from the connection.
@@ -75,10 +107,9 @@
*/
int write(const std::vector<uint8_t>& data) override;
-private:
- int mCurSockFd;
- std::atomic<int> mExit;
- std::mutex mMutex;
+ inline bool isOpen() override { return mSockFd > 0; }
+
+ private:
int mSockFd;
};
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.cpp b/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.cpp
index bf7be09..356a6b9 100644
--- a/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.cpp
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.cpp
@@ -16,9 +16,10 @@
#define LOG_TAG "VehicleEmulator_v2_0"
#include <android/log.h>
-#include <algorithm>
#include <android-base/properties.h>
+#include <log/log.h>
#include <utils/SystemClock.h>
+#include <algorithm>
#include <vhal_v2_0/VehicleUtils.h>
@@ -35,32 +36,45 @@
namespace impl {
-std::unique_ptr<CommBase> CommFactory::create() {
- bool isEmulator = android::base::GetBoolProperty("ro.kernel.qemu", false);
+VehicleEmulator::VehicleEmulator(EmulatedVehicleHalIface* hal) : mHal{hal} {
+ mHal->registerEmulator(this);
- if (isEmulator) {
- return std::make_unique<PipeComm>();
- } else {
- return std::make_unique<SocketComm>();
+ ALOGI("Starting SocketComm");
+ mSocketComm = std::make_unique<SocketComm>(this);
+ mSocketComm->start();
+
+ if (android::base::GetBoolProperty("ro.kernel.qemu", false)) {
+ ALOGI("Starting PipeComm");
+ mPipeComm = std::make_unique<PipeComm>(this);
+ mPipeComm->start();
}
}
VehicleEmulator::~VehicleEmulator() {
- mExit = true; // Notify thread to finish and wait for it to terminate.
- mComm->stop(); // Close emulator socket if it is open.
- if (mThread.joinable()) mThread.join();
+ mSocketComm->stop();
+ if (mPipeComm) {
+ mPipeComm->stop();
+ }
}
+/**
+ * This is called by the HAL when a property changes. We need to notify our clients that it has
+ * changed.
+ */
void VehicleEmulator::doSetValueFromClient(const VehiclePropValue& propValue) {
emulator::EmulatorMessage msg;
emulator::VehiclePropValue *val = msg.add_value();
populateProtoVehiclePropValue(val, &propValue);
msg.set_status(emulator::RESULT_OK);
msg.set_msg_type(emulator::SET_PROPERTY_ASYNC);
- txMsg(msg);
+
+ mSocketComm->sendMessage(msg);
+ if (mPipeComm) {
+ mPipeComm->sendMessage(msg);
+ }
}
-void VehicleEmulator::doGetConfig(VehicleEmulator::EmulatorMessage& rxMsg,
+void VehicleEmulator::doGetConfig(VehicleEmulator::EmulatorMessage const& rxMsg,
VehicleEmulator::EmulatorMessage& respMsg) {
std::vector<VehiclePropConfig> configs = mHal->listProperties();
emulator::VehiclePropGet getProp = rxMsg.prop(0);
@@ -79,7 +93,7 @@
}
}
-void VehicleEmulator::doGetConfigAll(VehicleEmulator::EmulatorMessage& /* rxMsg */,
+void VehicleEmulator::doGetConfigAll(VehicleEmulator::EmulatorMessage const& /* rxMsg */,
VehicleEmulator::EmulatorMessage& respMsg) {
std::vector<VehiclePropConfig> configs = mHal->listProperties();
@@ -92,8 +106,8 @@
}
}
-void VehicleEmulator::doGetProperty(VehicleEmulator::EmulatorMessage& rxMsg,
- VehicleEmulator::EmulatorMessage& respMsg) {
+void VehicleEmulator::doGetProperty(VehicleEmulator::EmulatorMessage const& rxMsg,
+ VehicleEmulator::EmulatorMessage& respMsg) {
int32_t areaId = 0;
emulator::VehiclePropGet getProp = rxMsg.prop(0);
int32_t propId = getProp.prop();
@@ -119,8 +133,8 @@
respMsg.set_status(status);
}
-void VehicleEmulator::doGetPropertyAll(VehicleEmulator::EmulatorMessage& /* rxMsg */,
- VehicleEmulator::EmulatorMessage& respMsg) {
+void VehicleEmulator::doGetPropertyAll(VehicleEmulator::EmulatorMessage const& /* rxMsg */,
+ VehicleEmulator::EmulatorMessage& respMsg) {
respMsg.set_msg_type(emulator::GET_PROPERTY_ALL_RESP);
respMsg.set_status(emulator::RESULT_OK);
@@ -132,7 +146,7 @@
}
}
-void VehicleEmulator::doSetProperty(VehicleEmulator::EmulatorMessage& rxMsg,
+void VehicleEmulator::doSetProperty(VehicleEmulator::EmulatorMessage const& rxMsg,
VehicleEmulator::EmulatorMessage& respMsg) {
emulator::VehiclePropValue protoVal = rxMsg.value(0);
VehiclePropValue val = {
@@ -173,58 +187,28 @@
respMsg.set_status(halRes ? emulator::RESULT_OK : emulator::ERROR_INVALID_PROPERTY);
}
-void VehicleEmulator::txMsg(emulator::EmulatorMessage& txMsg) {
- int numBytes = txMsg.ByteSize();
- std::vector<uint8_t> msg(static_cast<size_t>(numBytes));
-
- if (!txMsg.SerializeToArray(msg.data(), static_cast<int32_t>(msg.size()))) {
- ALOGE("%s: SerializeToString failed!", __func__);
- return;
- }
-
- if (mExit) {
- ALOGW("%s: unable to transmit a message, connection closed", __func__);
- return;
- }
-
- // Send the message
- int retVal = mComm->write(msg);
- if (retVal < 0) {
- ALOGE("%s: Failed to tx message: retval=%d, errno=%d", __func__, retVal, errno);
- }
-}
-
-void VehicleEmulator::parseRxProtoBuf(std::vector<uint8_t>& msg) {
- emulator::EmulatorMessage rxMsg;
- emulator::EmulatorMessage respMsg;
-
- if (rxMsg.ParseFromArray(msg.data(), static_cast<int32_t>(msg.size()))) {
- switch (rxMsg.msg_type()) {
- case emulator::GET_CONFIG_CMD:
- doGetConfig(rxMsg, respMsg);
- break;
- case emulator::GET_CONFIG_ALL_CMD:
- doGetConfigAll(rxMsg, respMsg);
- break;
- case emulator::GET_PROPERTY_CMD:
- doGetProperty(rxMsg, respMsg);
- break;
- case emulator::GET_PROPERTY_ALL_CMD:
- doGetPropertyAll(rxMsg, respMsg);
- break;
- case emulator::SET_PROPERTY_CMD:
- doSetProperty(rxMsg, respMsg);
- break;
- default:
- ALOGW("%s: Unknown message received, type = %d", __func__, rxMsg.msg_type());
- respMsg.set_status(emulator::ERROR_UNIMPLEMENTED_CMD);
- break;
- }
-
- // Send the reply
- txMsg(respMsg);
- } else {
- ALOGE("%s: ParseFromString() failed. msgSize=%d", __func__, static_cast<int>(msg.size()));
+void VehicleEmulator::processMessage(emulator::EmulatorMessage const& rxMsg,
+ emulator::EmulatorMessage& respMsg) {
+ switch (rxMsg.msg_type()) {
+ case emulator::GET_CONFIG_CMD:
+ doGetConfig(rxMsg, respMsg);
+ break;
+ case emulator::GET_CONFIG_ALL_CMD:
+ doGetConfigAll(rxMsg, respMsg);
+ break;
+ case emulator::GET_PROPERTY_CMD:
+ doGetProperty(rxMsg, respMsg);
+ break;
+ case emulator::GET_PROPERTY_ALL_CMD:
+ doGetPropertyAll(rxMsg, respMsg);
+ break;
+ case emulator::SET_PROPERTY_CMD:
+ doSetProperty(rxMsg, respMsg);
+ break;
+ default:
+ ALOGW("%s: Unknown message received, type = %d", __func__, rxMsg.msg_type());
+ respMsg.set_status(emulator::ERROR_UNIMPLEMENTED_CMD);
+ break;
}
}
@@ -316,40 +300,6 @@
}
}
-void VehicleEmulator::rxMsg() {
- while (!mExit) {
- std::vector<uint8_t> msg = mComm->read();
-
- if (msg.size() > 0) {
- // Received a message.
- parseRxProtoBuf(msg);
- } else {
- // This happens when connection is closed
- ALOGD("%s: msgSize=%zu", __func__, msg.size());
- break;
- }
- }
-}
-
-void VehicleEmulator::rxThread() {
- if (mExit) return;
-
- int retVal = mComm->open();
- if (retVal != 0) mExit = true;
-
- // Comms are properly opened
- while (!mExit) {
- retVal = mComm->connect();
-
- if (retVal >= 0) {
- rxMsg();
- }
-
- // Check every 100ms for a new connection
- std::this_thread::sleep_for(std::chrono::milliseconds(100));
- }
-}
-
} // impl
} // namespace V2_0
diff --git a/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.h b/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.h
index 1a8cfe2..58e387a 100644
--- a/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.h
+++ b/automotive/vehicle/2.0/default/impl/vhal_v2_0/VehicleEmulator.h
@@ -24,7 +24,9 @@
#include "vhal_v2_0/VehicleHal.h"
-#include "CommBase.h"
+#include "CommConn.h"
+#include "PipeComm.h"
+#include "SocketComm.h"
#include "VehicleHalProto.pb.h"
namespace android {
@@ -61,48 +63,36 @@
VehicleEmulator* mEmulator;
};
-struct CommFactory {
- static std::unique_ptr<CommBase> create();
-};
-
/**
* Emulates vehicle by providing controlling interface from host side either through ADB or Pipe.
*/
-class VehicleEmulator {
-public:
- VehicleEmulator(EmulatedVehicleHalIface* hal,
- std::unique_ptr<CommBase> comm = CommFactory::create())
- : mHal { hal },
- mComm(comm.release()),
- mThread { &VehicleEmulator::rxThread, this} {
- mHal->registerEmulator(this);
- }
+class VehicleEmulator : public MessageProcessor {
+ public:
+ VehicleEmulator(EmulatedVehicleHalIface* hal);
virtual ~VehicleEmulator();
void doSetValueFromClient(const VehiclePropValue& propValue);
+ void processMessage(emulator::EmulatorMessage const& rxMsg,
+ emulator::EmulatorMessage& respMsg) override;
-private:
+ private:
+ friend class ConnectionThread;
using EmulatorMessage = emulator::EmulatorMessage;
- void doGetConfig(EmulatorMessage& rxMsg, EmulatorMessage& respMsg);
- void doGetConfigAll(EmulatorMessage& rxMsg, EmulatorMessage& respMsg);
- void doGetProperty(EmulatorMessage& rxMsg, EmulatorMessage& respMsg);
- void doGetPropertyAll(EmulatorMessage& rxMsg, EmulatorMessage& respMsg);
- void doSetProperty(EmulatorMessage& rxMsg, EmulatorMessage& respMsg);
- void txMsg(emulator::EmulatorMessage& txMsg);
- void parseRxProtoBuf(std::vector<uint8_t>& msg);
+ void doGetConfig(EmulatorMessage const& rxMsg, EmulatorMessage& respMsg);
+ void doGetConfigAll(EmulatorMessage const& rxMsg, EmulatorMessage& respMsg);
+ void doGetProperty(EmulatorMessage const& rxMsg, EmulatorMessage& respMsg);
+ void doGetPropertyAll(EmulatorMessage const& rxMsg, EmulatorMessage& respMsg);
+ void doSetProperty(EmulatorMessage const& rxMsg, EmulatorMessage& respMsg);
void populateProtoVehicleConfig(emulator::VehiclePropConfig* protoCfg,
const VehiclePropConfig& cfg);
void populateProtoVehiclePropValue(emulator::VehiclePropValue* protoVal,
const VehiclePropValue* val);
- void rxMsg();
- void rxThread();
private:
- std::atomic<bool> mExit { false };
EmulatedVehicleHalIface* mHal;
- std::unique_ptr<CommBase> mComm;
- std::thread mThread;
+ std::unique_ptr<SocketComm> mSocketComm;
+ std::unique_ptr<PipeComm> mPipeComm;
};
} // impl