[android] Use attribute subscriptions in sensors screen (#11318)

* [android] Update device address on sensors screen start

Automatically resolve the device address using DNS-SD upon
entering the "Sensor Clusters" screen.

* [android] Use attribute subscriptions in sensors screen

The sensors screen currently actively polls the connected
sensor device every 3 seconds. Use attribute subscriptions,
instead.

Add a new method for cancelling all active subscriptions
for a given device. Previously, there was only a method
for cancelling a specific subscription, but controller
applications currently have no way of knowing the
subscription ID.

* Address code review comments

src/android/CHIPTool/app/src/main/java/com/google/chip/chiptool/clusterclient/SensorClientFragment.kt

@@ -5,6 +5,7 @@ import android.util.Log
 import android.view.LayoutInflater
 import android.view.View
 import android.view.ViewGroup
+import android.view.inputmethod.EditorInfo
 import android.widget.AdapterView
 import android.widget.ArrayAdapter
 import android.widget.Toast
@@ -28,38 +29,53 @@ private typealias ReadCallback = ChipClusters.IntegerAttributeCallback
 class SensorClientFragment : Fragment() {
   private val scope = CoroutineScope(Dispatchers.Main + Job())
 
-  // Job for sending periodic sensor read requests
-  private var sensorWatchJob: Job? = null
-
   // History of sensor values
   private val sensorData = LineGraphSeries<DataPoint>()
 
+  // Device whose attribute is subscribed
+  private var subscribedDevicePtr = 0L
+
   override fun onCreateView(
       inflater: LayoutInflater,
       container: ViewGroup?,
       savedInstanceState: Bundle?
   ): View {
     return inflater.inflate(R.layout.sensor_client_fragment, container, false).apply {
+      ChipClient.getDeviceController(requireContext()).setCompletionListener(null)
+      deviceIdEd.setOnEditorActionListener { textView, actionId, _ ->
+        if (actionId == EditorInfo.IME_ACTION_DONE) {
+          updateAddress(textView.text.toString())
+          resetSensorGraph() // reset the graph on device change
+        }
+        actionId == EditorInfo.IME_ACTION_DONE
+      }
+      endpointIdEd.setOnEditorActionListener { textView, actionId, _ ->
+        if (actionId == EditorInfo.IME_ACTION_DONE)
+          resetSensorGraph() // reset the graph on endpoint change
+        actionId == EditorInfo.IME_ACTION_DONE
+      }
       clusterNameSpinner.adapter = makeClusterNamesAdapter()
       clusterNameSpinner.onItemSelectedListener = object : AdapterView.OnItemSelectedListener {
         override fun onNothingSelected(parent: AdapterView<*>?) = Unit
         override fun onItemSelected(parent: AdapterView<*>?, view: View?, position: Int, id: Long) {
           resetSensorGraph() // reset the graph on cluster change
         }
       }
-      readSensorBtn.setOnClickListener { scope.launch { readSensorButtonClick() } }
+
+      readSensorBtn.setOnClickListener { scope.launch { readSensorCluster() } }
       watchSensorBtn.setOnCheckedChangeListener { _, isChecked ->
         if (isChecked) {
-          watchSensorButtonChecked()
+          scope.launch { subscribeSensorCluster() }
         } else {
-          watchSensorButtonUnchecked()
+          unsubscribeSensorCluster()
         }
       }
+
       val currentTime = Calendar.getInstance().time.time
       sensorGraph.addSeries(sensorData)
       sensorGraph.viewport.isXAxisBoundsManual = true
       sensorGraph.viewport.setMinX(currentTime.toDouble())
-      sensorGraph.viewport.setMaxX(currentTime.toDouble() + REFRESH_PERIOD_MS * MAX_DATA_POINTS)
+      sensorGraph.viewport.setMaxX(currentTime.toDouble() + MIN_REFRESH_PERIOD_S * 1000 * MAX_DATA_POINTS)
       sensorGraph.gridLabelRenderer.padding = 20
       sensorGraph.gridLabelRenderer.numHorizontalLabels = 4
       sensorGraph.gridLabelRenderer.setHorizontalLabelsAngle(150)
@@ -77,12 +93,24 @@ class SensorClientFragment : Fragment() {
   override fun onStart() {
     super.onStart()
     deviceIdEd.setText(DeviceIdUtil.getLastDeviceId(requireContext()).toString())
+    updateAddress(deviceIdEd.text.toString())
   }
 
   override fun onStop() {
-    super.onStop()
     scope.cancel()
     resetSensorGraph() // reset the graph on fragment exit
+    super.onStop()
+  }
+
+  private fun updateAddress(deviceId: String) {
+    try {
+      ChipClient.getDeviceController(requireContext()).updateDevice(
+          /* fabric ID */ 5544332211,
+          deviceId.toULong().toLong()
+      )
+    } catch (ex: Exception) {
+      showMessage(R.string.update_device_address_failure, ex.toString())
+    }
   }
 
   private fun resetSensorGraph() {
@@ -101,41 +129,53 @@ class SensorClientFragment : Fragment() {
     }
   }
 
-  private suspend fun readSensorButtonClick() {
+  private suspend fun readSensorCluster() {
     try {
-      readSensorCluster(clusterNameSpinner.selectedItem.toString(), false)
+      val deviceId = deviceIdEd.text.toString().toULong().toLong()
+      val endpointId = endpointIdEd.text.toString().toInt()
+      val clusterName = clusterNameSpinner.selectedItem.toString()
+      val clusterRead = CLUSTERS[clusterName]!!["read"] as (Long, Int, ReadCallback) -> Unit
+      val device = ChipClient.getConnectedDevicePointer(requireContext(), deviceId)
+      val callback = makeReadCallback(clusterName, false)
+
+      clusterRead(device, endpointId, callback)
     } catch (ex: Exception) {
       showMessage(R.string.sensor_client_read_error_text, ex.toString())
     }
   }
 
-  private fun watchSensorButtonChecked() {
-    sensorWatchJob = scope.launch {
-      while (isActive) {
-        try {
-          readSensorCluster(clusterNameSpinner.selectedItem.toString(), true)
-        } catch (ex: Exception) {
-          showMessage(R.string.sensor_client_read_error_text, ex.toString())
-        }
-        delay(REFRESH_PERIOD_MS)
-      }
+  private suspend fun subscribeSensorCluster() {
+    try {
+      val deviceId = deviceIdEd.text.toString().toULong().toLong()
+      val endpointId = endpointIdEd.text.toString().toInt()
+      val clusterName = clusterNameSpinner.selectedItem.toString()
+      val clusterSubscribe = CLUSTERS[clusterName]!!["subscribe"] as (Long, Int, ReadCallback) -> Unit
+      val device = ChipClient.getConnectedDevicePointer(requireContext(), deviceId)
+      val callback = makeReadCallback(clusterName, true)
+
+      clusterSubscribe(device, endpointId, callback)
+      subscribedDevicePtr = device
+    } catch (ex: Exception) {
+      showMessage(R.string.sensor_client_subscribe_error_text, ex.toString())
     }
   }
 
-  private fun watchSensorButtonUnchecked() {
-    sensorWatchJob?.cancel()
-    sensorWatchJob = null
-  }
+  private fun unsubscribeSensorCluster() {
+    if (subscribedDevicePtr == 0L)
+      return
 
-  private suspend fun readSensorCluster(clusterName: String, addToGraph: Boolean) {
-    val deviceId = deviceIdEd.text.toString().toULong().toLong()
-    val endpointId = endpointIdEd.text.toString().toInt()
-    val clusterConfig = CLUSTERS[clusterName]
-    val clusterRead = clusterConfig!!["read"] as (Long, Int, ReadCallback) -> Unit
+    try {
+      ChipClient.getDeviceController(requireContext()).shutdownSubscriptions(subscribedDevicePtr)
+      subscribedDevicePtr = 0
+    } catch (ex: Exception) {
+      showMessage(R.string.sensor_client_unsubscribe_error_text, ex.toString())
+    }
+  }
 
-    val device = ChipClient.getConnectedDevicePointer(requireContext(), deviceId)
+  private fun makeReadCallback(clusterName: String, addToGraph: Boolean): ReadCallback {
+    return object : ReadCallback {
+      val clusterConfig = CLUSTERS[clusterName]!!
 
-    clusterRead(device, endpointId, object : ReadCallback {
       override fun onSuccess(value: Int) {
         val unitValue = clusterConfig["unitValue"] as Double
         val unitSymbol = clusterConfig["unitSymbol"] as String
@@ -145,7 +185,7 @@ class SensorClientFragment : Fragment() {
       override fun onError(ex: Exception) {
         showMessage(R.string.sensor_client_read_error_text, ex.toString())
       }
-    })
+    }
   }
 
   private fun consumeSensorValue(value: Double, unitSymbol: String, addToGraph: Boolean) {
@@ -180,14 +220,25 @@ class SensorClientFragment : Fragment() {
 
   companion object {
     private const val TAG = "SensorClientFragment"
-    private const val REFRESH_PERIOD_MS = 3000L
+    private const val MIN_REFRESH_PERIOD_S = 2
+    private const val MAX_REFRESH_PERIOD_S = 10
     private const val MAX_DATA_POINTS = 60
     private val CLUSTERS = mapOf(
         "Temperature" to mapOf(
             "read" to { device: Long, endpointId: Int, callback: ReadCallback ->
               val cluster = ChipClusters.TemperatureMeasurementCluster(device, endpointId)
               cluster.readMeasuredValueAttribute(callback)
             },
+            "subscribe" to { device: Long, endpointId: Int, callback: ReadCallback ->
+              val cluster = ChipClusters.TemperatureMeasurementCluster(device, endpointId)
+              cluster.reportMeasuredValueAttribute(callback)
+              cluster.subscribeMeasuredValueAttribute(object : ChipClusters.DefaultClusterCallback {
+                override fun onSuccess() = Unit
+                override fun onError(ex: Exception) {
+                  callback.onError(ex)
+                }
+              }, MIN_REFRESH_PERIOD_S, MAX_REFRESH_PERIOD_S)
+            },
             "unitValue" to 0.01,
             "unitSymbol" to "\u00B0C"
         ),
@@ -196,6 +247,16 @@ class SensorClientFragment : Fragment() {
               val cluster = ChipClusters.PressureMeasurementCluster(device, endpointId)
               cluster.readMeasuredValueAttribute(callback)
             },
+            "subscribe" to { device: Long, endpointId: Int, callback: ReadCallback ->
+              val cluster = ChipClusters.PressureMeasurementCluster(device, endpointId)
+              cluster.reportMeasuredValueAttribute(callback)
+              cluster.subscribeMeasuredValueAttribute(object : ChipClusters.DefaultClusterCallback {
+                override fun onSuccess() = Unit
+                override fun onError(ex: Exception) {
+                  callback.onError(ex)
+                }
+              }, MIN_REFRESH_PERIOD_S, MAX_REFRESH_PERIOD_S)
+            },
             "unitValue" to 1.0,
             "unitSymbol" to "hPa"
         ),
@@ -204,6 +265,16 @@ class SensorClientFragment : Fragment() {
               val cluster = ChipClusters.RelativeHumidityMeasurementCluster(device, endpointId)
               cluster.readMeasuredValueAttribute(callback)
             },
+            "subscribe" to { device: Long, endpointId: Int, callback: ReadCallback ->
+              val cluster = ChipClusters.RelativeHumidityMeasurementCluster(device, endpointId)
+              cluster.reportMeasuredValueAttribute(callback)
+              cluster.subscribeMeasuredValueAttribute(object : ChipClusters.DefaultClusterCallback {
+                override fun onSuccess() = Unit
+                override fun onError(ex: Exception) {
+                  callback.onError(ex)
+                }
+              }, MIN_REFRESH_PERIOD_S, MAX_REFRESH_PERIOD_S)
+            },
             "unitValue" to 0.01,
             "unitSymbol" to "%"
         )

src/android/CHIPTool/app/src/main/res/values/strings.xml

@@ -29,6 +29,7 @@
     <string name="enter_device_id_hint_text">Enter Device ID</string>
     <string name="enter_endpoint_id_hint_text">Enter Endpoint ID</string>
     <string name="update_device_address_btn_text">Update address</string>
+    <string name="update_device_address_failure">Device address update failed: %1$s</string>
 
     <string name="address_commissioning_title_text">Commission with IP address</string>
     <string name="default_discriminator">3840</string>
@@ -86,6 +87,8 @@
     <string name="sensor_client_watch_btn_text">Watch</string>
     <string name="sensor_client_last_value_text">Last value: %1$.2f %2$s</string>
     <string name="sensor_client_read_error_text">Failed to read the sensor: %1$s</string>
+    <string name="sensor_client_subscribe_error_text">Failed to subscribe to the sensor: %1$s</string>
+    <string name="sensor_client_unsubscribe_error_text">Failed to unsubscribe from the sensor: %1$s</string>
     <string name="cluster_interaction_tool">Cluster Interaction Tool</string>
 
     <string name="multi_admin_client_btn_text">Multi-admin cluster</string>

src/app/InteractionModelEngine.cpp

@@ -220,6 +220,23 @@ CHIP_ERROR InteractionModelEngine::ShutdownSubscription(uint64_t aSubscriptionId
     return err;
 }
 
+CHIP_ERROR InteractionModelEngine::ShutdownSubscriptions(FabricIndex aFabricIndex, NodeId aPeerNodeId)
+{
+    CHIP_ERROR err = CHIP_ERROR_KEY_NOT_FOUND;
+
+    for (ReadClient & readClient : mReadClients)
+    {
+        if (!readClient.IsFree() && readClient.IsSubscriptionType() && readClient.GetFabricIndex() == aFabricIndex &&
+            readClient.GetPeerNodeId() == aPeerNodeId)
+        {
+            readClient.Shutdown();
+            err = CHIP_NO_ERROR;
+        }
+    }
+
+    return err;
+}
+
 CHIP_ERROR InteractionModelEngine::NewWriteClient(WriteClientHandle & apWriteClient, WriteClient::Callback * apCallback)
 {
     apWriteClient.SetWriteClient(nullptr);

src/app/InteractionModelEngine.h

@@ -118,6 +118,15 @@ class InteractionModelEngine : public Messaging::ExchangeDelegate, public Comman
      * @retval #CHIP_NO_ERROR On success.
      */
     CHIP_ERROR ShutdownSubscription(uint64_t aSubscriptionId);
+
+    /**
+     * Tears down active subscriptions for a given peer node ID.
+     *
+     * @retval #CHIP_ERROR_KEY_NOT_FOUND If no active subscription is found.
+     * @retval #CHIP_NO_ERROR On success.
+     */
+    CHIP_ERROR ShutdownSubscriptions(FabricIndex aFabricIndex, NodeId aPeerNodeId);
+
     /**
      *  Retrieve a WriteClient that the SDK consumer can use to send a write.  If the call succeeds,
      *  see WriteClient documentation for lifetime handling.

src/app/ReadClient.cpp

@@ -79,6 +79,7 @@ void ReadClient::ShutdownInternal(CHIP_ERROR aError)
     mpExchangeCtx              = nullptr;
     mInitialReport             = true;
     mPeerNodeId                = kUndefinedNodeId;
+    mFabricIndex               = kUndefinedFabricIndex;
     MoveToState(ClientState::Uninitialized);
 }
 
@@ -173,7 +174,8 @@ CHIP_ERROR ReadClient::SendReadRequest(ReadPrepareParams & aReadPrepareParams)
                                      Messaging::SendFlags(Messaging::SendMessageFlags::kExpectResponse));
     SuccessOrExit(err);
 
-    mPeerNodeId = aReadPrepareParams.mSessionHandle.GetPeerNodeId();
+    mPeerNodeId  = aReadPrepareParams.mSessionHandle.GetPeerNodeId();
+    mFabricIndex = aReadPrepareParams.mSessionHandle.GetFabricIndex();
 
     MoveToState(ClientState::AwaitingInitialReport);
 
@@ -655,7 +657,8 @@ CHIP_ERROR ReadClient::SendSubscribeRequest(ReadPrepareParams & aReadPreparePara
                                      Messaging::SendFlags(Messaging::SendMessageFlags::kExpectResponse));
     SuccessOrExit(err);
 
-    mPeerNodeId = aReadPrepareParams.mSessionHandle.GetPeerNodeId();
+    mPeerNodeId  = aReadPrepareParams.mSessionHandle.GetPeerNodeId();
+    mFabricIndex = aReadPrepareParams.mSessionHandle.GetFabricIndex();
     MoveToState(ClientState::AwaitingInitialReport);
 
 exit:

src/app/ReadClient.h

@@ -175,6 +175,7 @@ class ReadClient : public Messaging::ExchangeDelegate
         return mInteractionType == InteractionType::Subscribe ? returnType(mSubscriptionId) : returnType::Missing();
     }
 
+    FabricIndex GetFabricIndex() const { return mFabricIndex; }
     NodeId GetPeerNodeId() const { return mPeerNodeId; }
     bool IsReadType() { return mInteractionType == InteractionType::Read; }
     bool IsSubscriptionType() const { return mInteractionType == InteractionType::Subscribe; };
@@ -184,7 +185,7 @@ class ReadClient : public Messaging::ExchangeDelegate
     friend class TestReadInteraction;
     friend class InteractionModelEngine;
 
-    enum class ClientState
+    enum class ClientState : uint8_t
     {
         Uninitialized = 0,         ///< The client has not been initialized
         Initialized,               ///< The client has been initialized and is ready for a SendReadRequest
@@ -268,6 +269,7 @@ class ReadClient : public Messaging::ExchangeDelegate
     uint16_t mMaxIntervalCeilingSeconds        = 0;
     uint64_t mSubscriptionId                   = 0;
     NodeId mPeerNodeId                         = kUndefinedNodeId;
+    FabricIndex mFabricIndex                   = kUndefinedFabricIndex;
     InteractionType mInteractionType           = InteractionType::Read;
 };
 

src/controller/CHIPDevice.cpp

@@ -820,6 +820,11 @@ CHIP_ERROR Device::SendSubscribeAttributeRequest(app::AttributePathParams aPath,
     return CHIP_NO_ERROR;
 }
 
+CHIP_ERROR Device::ShutdownSubscriptions()
+{
+    return app::InteractionModelEngine::GetInstance()->ShutdownSubscriptions(mFabricIndex, mDeviceId);
+}
+
 CHIP_ERROR Device::SendWriteAttributeRequest(app::WriteClientHandle aHandle, Callback::Cancelable * onSuccessCallback,
                                              Callback::Cancelable * onFailureCallback)
 {

src/controller/CHIPDevice.h

@@ -139,6 +139,7 @@ class Device : public Messaging::ExchangeDelegate, public SessionEstablishmentDe
     CHIP_ERROR SendSubscribeAttributeRequest(app::AttributePathParams aPath, uint16_t mMinIntervalFloorSeconds,
                                              uint16_t mMaxIntervalCeilingSeconds, Callback::Cancelable * onSuccessCallback,
                                              Callback::Cancelable * onFailureCallback);
+    CHIP_ERROR ShutdownSubscriptions();
 
     CHIP_ERROR SendWriteAttributeRequest(app::WriteClientHandle aHandle, Callback::Cancelable * onSuccessCallback,
                                          Callback::Cancelable * onFailureCallback);

src/controller/java/CHIPDeviceController-JNI.cpp

@@ -304,6 +304,14 @@ JNI_METHOD(jboolean, isActive)(JNIEnv * env, jobject self, jlong handle)
     return chipDevice->IsActive();
 }
 
+JNI_METHOD(void, shutdownSubscriptions)(JNIEnv * env, jobject self, jlong handle, jlong devicePtr)
+{
+    chip::DeviceLayer::StackLock lock;
+
+    Device * device = reinterpret_cast<Device *>(devicePtr);
+    device->ShutdownSubscriptions();
+}
+
 JNI_METHOD(jstring, getIpAddress)(JNIEnv * env, jobject self, jlong handle, jlong deviceId)
 {
     chip::DeviceLayer::StackLock lock;