fix: subscribe method update

examples/topic-example.php

@@ -44,11 +44,11 @@ function printBanner(string $message, LoggerInterface $logger): void
 
 // Subscribe to topic
 $logger->info("Subscribing to topic: $TOPIC_NAME\n");
-$callback = function ($message) use ($logger) {
+$onMessage = function ($message) use ($logger) {
     $logger->info("Received message: " . json_encode($message));
     file_put_contents('message_received.flag', '1');
 };
-$response = $topicClient->subscribe($CACHE_NAME, $TOPIC_NAME, $callback);
+$response = $topicClient->subscribe($CACHE_NAME, $TOPIC_NAME, $onMessage);
 if ($response->asSuccess()) {
     $logger->info("SUCCESS: Subscribed to topic: " . $TOPIC_NAME . "\n");
 } elseif ($response->asError()) {

src/Topic/Internal/ScsTopicClient.php

@@ -113,9 +113,10 @@ public function publish(string $cacheName, string $topicName, string $value): To
      *
      * @param string   $cacheName The name of the cache.
      * @param string   $topicName The name of the topic to subscribe to.
-     * @param callable $callback  The callback function to handle incoming messages.
+     * @param callable $onMessage Callback for handling incoming messages.
+     * @return TopicSubscribeResponse
      */
-    public function subscribe(string $cacheName, string $topicName, callable $callback): TopicSubscribeResponse
+    public function subscribe(string $cacheName, string $topicName, callable $onMessage): TopicSubscribeResponse
     {
         $this->logger->info("Subscribing to topic: $topicName in cache $cacheName\n");
 
@@ -128,16 +129,23 @@ public function subscribe(string $cacheName, string $topicName, callable $callba
             $call = $this->grpcManager->client->Subscribe($request);
 
             foreach ($call->responses() as $response) {
-                $this->logger->info("Received message from topic $topicName in cache $cacheName\n");
-                $callback($response);
+                try {
+                    $this->logger->info("Received message from topic $topicName in cache $cacheName\n");
+                    $onMessage($response);
+                } catch (\Exception $e) {
+                    $this->logger->error("Error processing message: " . $e->getMessage());
+                }
             }
+
+
         } catch (SdkError $e) {
             $this->logger->debug("Failed to subscribe to topic $topicName in cache $cacheName: {$e->getMessage()}");
             return new TopicSubscribeResponseError($e);
         } catch (\Exception $e) {
             $this->logger->debug("Failed to subscribe to topic $topicName in cache $cacheName: {$e->getMessage()}");
             return new TopicSubscribeResponseError(new UnknownError($e->getMessage()));
         }
+
         return new TopicSubscribeResponseSubscription();
     }
 

src/Topic/TopicClient.php

@@ -63,9 +63,9 @@ public function publish(string $cacheName, string $topicName, string $message):
         return $this->topicClient->publish($cacheName, $topicName, $message);
     }
 
-    public function subscribe(string $cacheName, string $topicName, callable $callback): TopicSubscribeResponse
+    public function subscribe(string $cacheName, string $topicName, callable $onMessage): TopicSubscribeResponse
     {
         $this->logger->info("Subscribing to topic: $topicName\n");
-        return $this->topicClient->subscribe($cacheName, $topicName, $callback);
+        return $this->topicClient->subscribe($cacheName, $topicName, $onMessage);
     }
 }