feat(task-streaming): rename ISynchronousOptionsProvider to ISynchronousOptionsAwareProvider

Signed-off-by: Julien Veyssier <julien-nc@posteo.net>

Author: julien-nc

lib/private/TaskProcessing/Manager.php

@@ -19,6 +19,7 @@
 use OCP\App\IAppManager;
 use OCP\AppFramework\Db\DoesNotExistException;
 use OCP\AppFramework\Db\MultipleObjectsReturnedException;
+use OCP\AppFramework\Utility\ITimeFactory;
 use OCP\BackgroundJob\IJobList;
 use OCP\DB\Exception;
 use OCP\EventDispatcher\IEventDispatcher;
@@ -59,7 +60,7 @@
 use OCP\TaskProcessing\IInternalTaskType;
 use OCP\TaskProcessing\IManager;
 use OCP\TaskProcessing\IProvider;
-use OCP\TaskProcessing\ISynchronousOptionsProvider;
+use OCP\TaskProcessing\ISynchronousOptionsAwareProvider;
 use OCP\TaskProcessing\ISynchronousProvider;
 use OCP\TaskProcessing\ISynchronousWatermarkingProvider;
 use OCP\TaskProcessing\ITaskType;
@@ -158,6 +159,7 @@ public function __construct(
 		private IUserSession $userSession,
 		ICacheFactory $cacheFactory,
 		private IFactory $l10nFactory,
+		private ITimeFactory $timeFactory,
 	) {
 		$this->appData = $appDataFactory->get('core');
 		$this->distributedCache = $cacheFactory->createDistributed('task_processing::');
@@ -1134,7 +1136,7 @@ public function processTask(Task $task, ISynchronousProvider $provider): bool {
 				$this->setTaskStatus($task, Task::STATUS_RUNNING);
 				if ($provider instanceof ISynchronousWatermarkingProvider) {
 					$output = $provider->process($task->getUserId(), $input, fn (float $progress) => $this->setTaskProgress($task->getId(), $progress), $task->getIncludeWatermark());
-				} elseif ($provider instanceof ISynchronousOptionsProvider) {
+				} elseif ($provider instanceof ISynchronousOptionsAwareProvider) {
 					$options = new SynchronousProviderOptions(
 						$task->getIncludeWatermark(),
 						$task->getPreferStreaming(),
@@ -1238,7 +1240,7 @@ public function setTaskIntermediateOutput(int $id, array $output): bool {
 		if ($userId !== null
 			&& $userId !== ''
 			&& $this->appManager->isEnabledForAnyone('notify_push')
-			&& class_exists('\OCA\NotifyPush\Queue\IQueue')
+			&& interface_exists('\OCA\NotifyPush\Queue\IQueue')
 		) {
 			try {
 				/** @psalm-suppress UndefinedClass */
@@ -1249,14 +1251,17 @@ public function setTaskIntermediateOutput(int $id, array $output): bool {
 					'message' => 'task_' . $task->getId(),
 					'body' => $output,
 				]);
-				// we don't update the DB if something was sent via notify_push
-				// so if the push messages are not received for some reason, the polling will still not see any intermediate output
-				// but will receive the final output
-				return true;
 			} catch (ContainerExceptionInterface|NotFoundExceptionInterface $e) {
 				$this->logger->debug('OCA\NotifyPush\IQueue not found, not sending to queue');
 			}
 		}
+
+		// throttle DB update
+		$now = $this->timeFactory->now()->getTimestamp();
+		if ($now - $task->getLastUpdated() < 2) {
+			return true;
+		}
+
 		// no output shape validation for now
 		$task->setOutput($output);
 		$taskEntity = \OC\TaskProcessing\Db\Task::fromPublicTask($task);

…ocessing/ISynchronousOptionsProvider.php …ing/ISynchronousOptionsAwareProvider.phplib/public/TaskProcessing/ISynchronousOptionsProvider.php renamed to lib/public/TaskProcessing/ISynchronousOptionsAwareProvider.php lib/public/TaskProcessing/ISynchronousOptionsProvider.php renamed to lib/public/TaskProcessing/ISynchronousOptionsAwareProvider.php

@@ -17,7 +17,7 @@
  * implement a task processing provider
  * @since 35.0.0
  */
-interface ISynchronousOptionsProvider extends ISynchronousProvider {
+interface ISynchronousOptionsAwareProvider extends ISynchronousProvider {
 
 	/**
 	 * Returns the shape of optional output parameters