Parallelize Notifications

lib/Listener/CalendarListener.php

@@ -37,6 +37,7 @@
 use OCA\DAV\Events\CardUpdatedEvent;
 
 use Psr\Log\LoggerInterface;
+use GuzzleHttp\Promise;
 
 use OCA\DavPush\Service\SubscriptionService;
 use OCA\DavPush\Transport\TransportManager;
@@ -59,13 +60,16 @@ public function handle(Event $event): void {
 		$collectionName = $event->getCalendarData()['uri'];
 		$subscriptions = $this->subscriptionService->findAll($collectionName);
 
-		foreach($subscriptions as $subscription) {
-			$transport = $this->transportManager->getTransport($subscription->getTransport());
-			try {
-				$transport->notify($subscription->getUserId(), $collectionName, $subscription->getId());
-			} catch (\Exception $e) {
-				$this->logger->error("transport " .  $subscription->getTransport() . " failed to deliver notification to subscription " . $subscription->getId());
+		$notificationPromises = (function () use ($collectionName, $subscriptions): Generator {
+			foreach($subscriptions as $subscription) {
+				$transport = $this->transportManager->getTransport($subscription->getTransport());
+				yield $transport->notify($subscription->getUserId(), $collectionName, $subscription->getId());
 			}
-		}
+		})();
+
+		$responses = Promise\Utils::settle($notificationPromises)->wait();
+
+		// TODO: iterate over responses and log errors
+		// $this->logger->error("transport " .  $subscription->getTransport() . " failed to deliver notification to subscription " . $subscription->getId());
     }
 }

lib/PushTransports/WebPushTransport.php

@@ -30,13 +30,17 @@
 use OCA\DavPush\Service\WebPushSubscriptionService;
 use OCA\DavPush\Errors\WebPushSubscriptionNotFound;
 
+use OCP\Http\Client\IClientService;
+use OCP\Http\Client\IPromise;
+
 use Sabre\Xml\Service;
 
 class WebPushTransport extends Transport {
 	protected $id = "web-push";
 
 	public function __construct(
 		private WebPushSubscriptionService $webPushSubscriptionService,
+		private IClientService $httpClientService
 	) {}
 
 	private function parseOptions(array $options): array {
@@ -79,7 +83,7 @@ public function registerSubscription($subsciptionId, $options) {
 		];
 	}
 
-	public function notify(string $userId, string $collectionName, int $subscriptionId) {
+	public function notify(string $userId, string $collectionName, int $subscriptionId): IPromise {
 		$xmlService = new Service();
 
 		$pushResource = $this->webPushSubscriptionService->findBySubscriptionId($subscriptionId)->getPushResource();
@@ -88,15 +92,15 @@ public function notify(string $userId, string $collectionName, int $subscription
 			'{DAV:Push}topic' => $collectionName,
 		]);
 
-		$options = [
-			'http' => [
-				'method' => 'POST',
-				'content' => $content,
-			],
-		];
+		$httpClient = $this->httpClientService->newClient();
 
-		$context = stream_context_create($options);
-		$result = file_get_contents($pushResource, false, $context);
+		return $httpClient->postAsync($pushResource, [
+			"body" => $content,
+			"timeout" => 10,
+			"headers" => [
+				"Content-Type" => "application/xml",
+			],
+		]);
 	}
 
 	public function getSubscriptionIdFromOptions(string $userId, string $collectionName, $options): ?int {

lib/Transport/Transport.php

@@ -26,6 +26,8 @@
 
 namespace OCA\DavPush\Transport;
 
+use OCP\Http\Client\IPromise;
+
 abstract class Transport {
 	protected $id;
 
@@ -69,5 +71,5 @@ abstract public function getSubscriptionIdFromOptions(string $userId, string $co
 	// Change mutable options of the subscription (if any exist)
 	abstract public function updateSubscription($subsciptionId, $options);
 
-	abstract public function notify(string $userId, string $collectionName, int $subscriptionId);
+	abstract public function notify(string $userId, string $collectionName, int $subscriptionId): IPromise;
 }