diff --git a/automotive/vehicle/aidl/impl/vhal/Android.bp b/automotive/vehicle/aidl/impl/vhal/Android.bp
index ef1d0f1..79d3ebd 100644
--- a/automotive/vehicle/aidl/impl/vhal/Android.bp
+++ b/automotive/vehicle/aidl/impl/vhal/Android.bp
@@ -53,7 +53,10 @@
     ],
     local_include_dirs: ["include"],
     export_include_dirs: ["include"],
-    srcs: ["src/DefaultVehicleHal.cpp"],
+    srcs: [
+        "src/ConnectedClient.cpp",
+        "src/DefaultVehicleHal.cpp",
+    ],
     static_libs: [
         "VehicleHalUtils",
         "android-automotive-large-parcelable-vendor-lib",
diff --git a/automotive/vehicle/aidl/impl/vhal/include/ConnectedClient.h b/automotive/vehicle/aidl/impl/vhal/include/ConnectedClient.h
new file mode 100644
index 0000000..43a9603
--- /dev/null
+++ b/automotive/vehicle/aidl/impl/vhal/include/ConnectedClient.h
@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2021 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_aidl_impl_vhal_include_ConnectedClient_H_
+#define android_hardware_automotive_vehicle_aidl_impl_vhal_include_ConnectedClient_H_
+
+#include <VehicleHalTypes.h>
+
+#include <aidl/android/hardware/automotive/vehicle/IVehicleCallback.h>
+#include <android-base/result.h>
+
+#include <memory>
+#include <unordered_set>
+#include <vector>
+
+namespace android {
+namespace hardware {
+namespace automotive {
+namespace vehicle {
+
+// A class to represent a binder client with a callback interface. Each callback function, e.g.
+// GetValues or SetValues for a specific binder client is a separate {@code ConnectedClient}.
+// For one {@code ConnectedClient}, we use one pending request pool to manage all pending requests,
+// so the request IDs must be unique for one client. We also manage a set of callback functions
+// for one client, e.g. timeoutCallback which could be passed to hardware.
+// This class is thread-safe.
+class ConnectedClient {
+  public:
+    ConnectedClient(
+            std::shared_ptr<::aidl::android::hardware::automotive::vehicle::IVehicleCallback>
+                    callback);
+
+    virtual ~ConnectedClient() = default;
+
+  protected:
+    const std::shared_ptr<::aidl::android::hardware::automotive::vehicle::IVehicleCallback>
+            mCallback;
+};
+
+// A class to represent a client that calls {@code IVehicle.setValues} or {@code
+// IVehicle.getValues}.
+template <class ResultType, class ResultsType>
+class GetSetValuesClient final : public ConnectedClient {
+  public:
+    GetSetValuesClient(
+            std::shared_ptr<::aidl::android::hardware::automotive::vehicle::IVehicleCallback>
+                    callback);
+
+    // Sends the results to this client.
+    void sendResults(const std::vector<ResultType>& results);
+
+    // Sends each result separately to this client. Each result would be sent through one callback
+    // invocation.
+    void sendResultsSeparately(const std::vector<ResultType>& results);
+
+    // Gets the callback to be called when the request for this client has finished.
+    std::shared_ptr<const std::function<void(std::vector<ResultType>)>> getResultCallback();
+
+  private:
+    // The following members are only initialized during construction.
+    std::shared_ptr<const std::function<void(std::vector<ResultType>)>> mResultCallback;
+};
+
+}  // namespace vehicle
+}  // namespace automotive
+}  // namespace hardware
+}  // namespace android
+
+#endif  // android_hardware_automotive_vehicle_aidl_impl_vhal_include_ConnectedClient_H_
diff --git a/automotive/vehicle/aidl/impl/vhal/src/ConnectedClient.cpp b/automotive/vehicle/aidl/impl/vhal/src/ConnectedClient.cpp
new file mode 100644
index 0000000..656dfaf
--- /dev/null
+++ b/automotive/vehicle/aidl/impl/vhal/src/ConnectedClient.cpp
@@ -0,0 +1,160 @@
+/*
+ * Copyright (C) 2021 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.
+ */
+
+#include "ConnectedClient.h"
+#include "ParcelableUtils.h"
+
+#include <VehicleHalTypes.h>
+
+#include <utils/Log.h>
+
+#include <inttypes.h>
+#include <unordered_set>
+#include <vector>
+
+namespace android {
+namespace hardware {
+namespace automotive {
+namespace vehicle {
+
+namespace {
+
+using ::aidl::android::hardware::automotive::vehicle::GetValueResult;
+using ::aidl::android::hardware::automotive::vehicle::GetValueResults;
+using ::aidl::android::hardware::automotive::vehicle::IVehicleCallback;
+using ::aidl::android::hardware::automotive::vehicle::SetValueResult;
+using ::aidl::android::hardware::automotive::vehicle::SetValueResults;
+using ::aidl::android::hardware::automotive::vehicle::StatusCode;
+using ::aidl::android::hardware::automotive::vehicle::VehiclePropValue;
+using ::aidl::android::hardware::automotive::vehicle::VehiclePropValues;
+using ::android::base::Result;
+using ::ndk::ScopedAStatus;
+
+// A function to call the specific callback based on results type.
+template <class T>
+ScopedAStatus callCallback(std::shared_ptr<IVehicleCallback> callback, const T& results);
+
+template <>
+ScopedAStatus callCallback<GetValueResults>(std::shared_ptr<IVehicleCallback> callback,
+                                            const GetValueResults& results) {
+    return callback->onGetValues(results);
+}
+
+template <>
+ScopedAStatus callCallback<SetValueResults>(std::shared_ptr<IVehicleCallback> callback,
+                                            const SetValueResults& results) {
+    return callback->onSetValues(results);
+}
+
+// Send a single GetValue/SetValue result through the callback.
+template <class ResultType, class ResultsType>
+void sendGetOrSetValueResult(std::shared_ptr<IVehicleCallback> callback, const ResultType& result) {
+    ResultsType parcelableResults;
+    parcelableResults.payloads.resize(1);
+    parcelableResults.payloads[0] = result;
+    if (ScopedAStatus callbackStatus = callCallback(callback, parcelableResults);
+        !callbackStatus.isOk()) {
+        ALOGE("failed to call callback, error: %s, code: %d", callbackStatus.getMessage(),
+              callbackStatus.getServiceSpecificError());
+    }
+}
+
+// Send all the GetValue/SetValue results through callback, one result in each callback invocation.
+template <class ResultType, class ResultsType>
+void sendGetOrSetValueResultsSeparately(std::shared_ptr<IVehicleCallback> callback,
+                                        const std::vector<ResultType>& results) {
+    for (const auto& result : results) {
+        sendGetOrSetValueResult<ResultType, ResultsType>(callback, result);
+    }
+}
+
+// Send all the GetValue/SetValue results through callback in a single callback invocation.
+template <class ResultType, class ResultsType>
+void sendGetOrSetValueResults(std::shared_ptr<IVehicleCallback> callback,
+                              const std::vector<ResultType>& results) {
+    ResultsType parcelableResults;
+    ScopedAStatus status = vectorToStableLargeParcelable(results, &parcelableResults);
+    if (status.isOk()) {
+        if (ScopedAStatus callbackStatus = callCallback(callback, parcelableResults);
+            !callbackStatus.isOk()) {
+            ALOGE("failed to call callback, error: %s, code: %d", status.getMessage(),
+                  status.getServiceSpecificError());
+        }
+        return;
+    }
+    int statusCode = status.getServiceSpecificError();
+    ALOGE("failed to marshal result into large parcelable, error: "
+          "%s, code: %d",
+          status.getMessage(), statusCode);
+    sendGetOrSetValueResultsSeparately<ResultType, ResultsType>(callback, results);
+}
+
+// Specify the functions for GetValues and SetValues types.
+template void sendGetOrSetValueResult<GetValueResult, GetValueResults>(
+        std::shared_ptr<IVehicleCallback> callback, const GetValueResult& result);
+template void sendGetOrSetValueResult<SetValueResult, SetValueResults>(
+        std::shared_ptr<IVehicleCallback> callback, const SetValueResult& result);
+
+template void sendGetOrSetValueResults<GetValueResult, GetValueResults>(
+        std::shared_ptr<IVehicleCallback> callback, const std::vector<GetValueResult>& results);
+template void sendGetOrSetValueResults<SetValueResult, SetValueResults>(
+        std::shared_ptr<IVehicleCallback> callback, const std::vector<SetValueResult>& results);
+
+template void sendGetOrSetValueResultsSeparately<GetValueResult, GetValueResults>(
+        std::shared_ptr<IVehicleCallback> callback, const std::vector<GetValueResult>& results);
+template void sendGetOrSetValueResultsSeparately<SetValueResult, SetValueResults>(
+        std::shared_ptr<IVehicleCallback> callback, const std::vector<SetValueResult>& results);
+
+}  // namespace
+
+ConnectedClient::ConnectedClient(std::shared_ptr<IVehicleCallback> callback)
+    : mCallback(callback) {}
+
+template <class ResultType, class ResultsType>
+GetSetValuesClient<ResultType, ResultsType>::GetSetValuesClient(
+        std::shared_ptr<IVehicleCallback> callback)
+    : ConnectedClient(callback) {
+    mResultCallback = std::make_shared<const std::function<void(std::vector<ResultType>)>>(
+            [callback](std::vector<ResultType> results) {
+                return sendGetOrSetValueResults<ResultType, ResultsType>(callback, results);
+            });
+}
+
+template <class ResultType, class ResultsType>
+std::shared_ptr<const std::function<void(std::vector<ResultType>)>>
+GetSetValuesClient<ResultType, ResultsType>::getResultCallback() {
+    return mResultCallback;
+}
+
+template <class ResultType, class ResultsType>
+void GetSetValuesClient<ResultType, ResultsType>::sendResults(
+        const std::vector<ResultType>& results) {
+    return sendGetOrSetValueResults<ResultType, ResultsType>(mCallback, results);
+}
+
+template <class ResultType, class ResultsType>
+void GetSetValuesClient<ResultType, ResultsType>::sendResultsSeparately(
+        const std::vector<ResultType>& results) {
+    return sendGetOrSetValueResultsSeparately<ResultType, ResultsType>(mCallback, results);
+}
+
+template class GetSetValuesClient<GetValueResult, GetValueResults>;
+template class GetSetValuesClient<SetValueResult, SetValueResults>;
+
+}  // namespace vehicle
+}  // namespace automotive
+}  // namespace hardware
+}  // namespace android
diff --git a/automotive/vehicle/aidl/impl/vhal/test/ConnectedClientTest.cpp b/automotive/vehicle/aidl/impl/vhal/test/ConnectedClientTest.cpp
new file mode 100644
index 0000000..ddd0c65
--- /dev/null
+++ b/automotive/vehicle/aidl/impl/vhal/test/ConnectedClientTest.cpp
@@ -0,0 +1,208 @@
+/*
+ * Copyright (C) 2021 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.
+ */
+
+#include "ConnectedClient.h"
+#include "MockVehicleCallback.h"
+
+#include <aidl/android/hardware/automotive/vehicle/IVehicleCallback.h>
+
+#include <gtest/gtest.h>
+
+namespace android {
+namespace hardware {
+namespace automotive {
+namespace vehicle {
+
+using ::aidl::android::hardware::automotive::vehicle::GetValueResult;
+using ::aidl::android::hardware::automotive::vehicle::GetValueResults;
+using ::aidl::android::hardware::automotive::vehicle::IVehicleCallback;
+using ::aidl::android::hardware::automotive::vehicle::SetValueResult;
+using ::aidl::android::hardware::automotive::vehicle::SetValueResults;
+using ::aidl::android::hardware::automotive::vehicle::StatusCode;
+using ::aidl::android::hardware::automotive::vehicle::VehiclePropValue;
+
+class ConnectedClientTest : public ::testing::Test {
+  public:
+    void SetUp() override {
+        mCallback = ndk::SharedRefBase::make<MockVehicleCallback>();
+        mCallbackClient = IVehicleCallback::fromBinder(mCallback->asBinder());
+    }
+
+    std::shared_ptr<IVehicleCallback> getCallbackClient() { return mCallbackClient; }
+
+    MockVehicleCallback* getCallback() { return mCallback.get(); }
+
+  protected:
+    using GetValuesClient = GetSetValuesClient<GetValueResult, GetValueResults>;
+    using SetValuesClient = GetSetValuesClient<SetValueResult, SetValueResults>;
+
+  private:
+    std::shared_ptr<MockVehicleCallback> mCallback;
+    std::shared_ptr<IVehicleCallback> mCallbackClient;
+};
+
+TEST_F(ConnectedClientTest, testSendGetValueResults) {
+    std::vector<GetValueResult> results = {{
+                                                   .requestId = 0,
+                                                   .status = StatusCode::OK,
+                                                   .prop =
+                                                           VehiclePropValue{
+                                                                   .prop = 0,
+                                                           },
+                                           },
+                                           {
+                                                   .requestId = 1,
+                                                   .status = StatusCode::OK,
+                                                   .prop =
+                                                           VehiclePropValue{
+                                                                   .prop = 1,
+                                                           },
+                                           }};
+
+    GetValuesClient client(getCallbackClient());
+
+    client.sendResults(results);
+
+    auto maybeGetValueResults = getCallback()->nextGetValueResults();
+    ASSERT_TRUE(maybeGetValueResults.has_value());
+    ASSERT_EQ(maybeGetValueResults.value().payloads, results);
+}
+
+TEST_F(ConnectedClientTest, testSendGetValueResultsSeparately) {
+    std::vector<GetValueResult> results = {{
+                                                   .requestId = 0,
+                                                   .status = StatusCode::OK,
+                                                   .prop =
+                                                           VehiclePropValue{
+                                                                   .prop = 0,
+                                                           },
+                                           },
+                                           {
+                                                   .requestId = 1,
+                                                   .status = StatusCode::OK,
+                                                   .prop =
+                                                           VehiclePropValue{
+                                                                   .prop = 1,
+                                                           },
+                                           }};
+
+    GetValuesClient client(getCallbackClient());
+
+    client.sendResultsSeparately(results);
+
+    for (auto& result : results) {
+        auto maybeGetValueResults = getCallback()->nextGetValueResults();
+        EXPECT_TRUE(maybeGetValueResults.has_value());
+        if (!maybeGetValueResults.has_value()) {
+            continue;
+        }
+        EXPECT_EQ(maybeGetValueResults.value().payloads, std::vector<GetValueResult>({result}));
+    }
+}
+
+TEST_F(ConnectedClientTest, testGetValuesGnResultCallback) {
+    std::vector<GetValueResult> results = {{
+                                                   .requestId = 0,
+                                                   .status = StatusCode::OK,
+                                                   .prop =
+                                                           VehiclePropValue{
+                                                                   .prop = 0,
+                                                           },
+                                           },
+                                           {
+                                                   .requestId = 1,
+                                                   .status = StatusCode::OK,
+                                                   .prop =
+                                                           VehiclePropValue{
+                                                                   .prop = 1,
+                                                           },
+                                           }};
+
+    GetValuesClient client(getCallbackClient());
+
+    (*(client.getResultCallback()))(results);
+
+    auto maybeGetValueResults = getCallback()->nextGetValueResults();
+    ASSERT_TRUE(maybeGetValueResults.has_value());
+    ASSERT_EQ(maybeGetValueResults.value().payloads, results);
+}
+
+TEST_F(ConnectedClientTest, testSendSetValueResults) {
+    std::vector<SetValueResult> results = {{
+                                                   .requestId = 0,
+                                                   .status = StatusCode::OK,
+                                           },
+                                           {
+                                                   .requestId = 1,
+                                                   .status = StatusCode::OK,
+                                           }};
+
+    SetValuesClient client(getCallbackClient());
+
+    client.sendResults(results);
+
+    auto maybeSetValueResults = getCallback()->nextSetValueResults();
+    ASSERT_TRUE(maybeSetValueResults.has_value());
+    ASSERT_EQ(maybeSetValueResults.value().payloads, results);
+}
+
+TEST_F(ConnectedClientTest, testSendSetValueResultsSeparately) {
+    std::vector<SetValueResult> results = {{
+                                                   .requestId = 0,
+                                                   .status = StatusCode::OK,
+                                           },
+                                           {
+                                                   .requestId = 1,
+                                                   .status = StatusCode::OK,
+                                           }};
+
+    SetValuesClient client(getCallbackClient());
+
+    client.sendResultsSeparately(results);
+
+    for (auto& result : results) {
+        auto maybeSetValueResults = getCallback()->nextSetValueResults();
+        EXPECT_TRUE(maybeSetValueResults.has_value());
+        if (!maybeSetValueResults.has_value()) {
+            continue;
+        }
+        EXPECT_EQ(maybeSetValueResults.value().payloads, std::vector<SetValueResult>({result}));
+    }
+}
+
+TEST_F(ConnectedClientTest, testSetValuesGetResultCallback) {
+    std::vector<SetValueResult> results = {{
+                                                   .requestId = 0,
+                                                   .status = StatusCode::OK,
+                                           },
+                                           {
+                                                   .requestId = 1,
+                                                   .status = StatusCode::OK,
+                                           }};
+
+    SetValuesClient client(getCallbackClient());
+
+    (*(client.getResultCallback()))(results);
+
+    auto maybeSetValueResults = getCallback()->nextSetValueResults();
+    ASSERT_TRUE(maybeSetValueResults.has_value());
+    ASSERT_EQ(maybeSetValueResults.value().payloads, results);
+}
+
+}  // namespace vehicle
+}  // namespace automotive
+}  // namespace hardware
+}  // namespace android