[TA] Add analyze tasks feature support

sdk/textanalytics/azure-ai-textanalytics/CHANGELOG.md

@@ -1,7 +1,9 @@
 # Release History
 ## 5.1.0-beta.3 (Unreleased)
 **New features**
-- Added support for Healthcare analysis, it is a long-running operation, and the cancellation supported. 
+- Added support for healthcare analysis feature. It is represented as a long-running operation. Cancellation supported. 
+- Added support for analyze tasks feature, It analyzes multiple tasks (such as, entity recognition, PII entity recognition 
+and key phrases extraction) simultaneously in a list of document.
 
 ## 5.1.0-beta.2 (2020-10-06)
 ### Breaking changes

sdk/textanalytics/azure-ai-textanalytics/src/main/java/com/azure/ai/textanalytics/AnalyzeHealthcareAsyncClient.java

@@ -35,6 +35,7 @@
 import com.azure.core.util.polling.PollingContext;
 import reactor.core.publisher.Mono;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -44,7 +45,7 @@
 import java.util.stream.Collectors;
 
 import static com.azure.ai.textanalytics.TextAnalyticsAsyncClient.COGNITIVE_TRACING_NAMESPACE_VALUE;
-import static com.azure.ai.textanalytics.implementation.Utility.DEFAULT_POLL_DURATION;
+import static com.azure.ai.textanalytics.implementation.Utility.DEFAULT_POLL_INTERVAL;
 import static com.azure.ai.textanalytics.implementation.Utility.inputDocumentsValidation;
 import static com.azure.ai.textanalytics.implementation.Utility.parseModelId;
 import static com.azure.ai.textanalytics.implementation.Utility.parseNextLink;
@@ -74,12 +75,17 @@ PollerFlux<TextAnalyticsOperationResult, PagedFlux<HealthcareTaskResult>> beginA
         try {
             inputDocumentsValidation(documents);
             String modelVersion = null;
+            Duration pollInterval = DEFAULT_POLL_INTERVAL;
             if (options != null) {
                 modelVersion = options.getModelVersion();
+                pollInterval = options.getPollInterval();
             }
+            // the variable used in the lambda function has to be 'final' or 'effective final'.
             final Boolean finalIncludeStatistics = options == null ? null : options.isIncludeStatistics();
+            final Integer finalTop = options == null ? null : options.getTop();
+            final Integer finalSkip = options == null ? null : options.getSkip();
             return new PollerFlux<>(
-                DEFAULT_POLL_DURATION,
+                pollInterval,
                 activationOperation(service.healthWithResponseAsync(
                     new MultiLanguageBatchInput().setDocuments(toMultiLanguageInput(documents)),
                     modelVersion,
@@ -92,12 +98,12 @@ PollerFlux<TextAnalyticsOperationResult, PagedFlux<HealthcareTaskResult>> beginA
                             parseModelId(healthResponse.getDeserializedHeaders().getOperationLocation()));
                         return textAnalyticsOperationResult;
                     })),
-                pollingOperation(jobId -> service.healthStatusWithResponseAsync(jobId, null, null,
-                    finalIncludeStatistics, context)),
+                pollingOperation(jobId -> service.healthStatusWithResponseAsync(jobId,
+                    finalTop, finalSkip, finalIncludeStatistics, context)),
                 (activationResponse, pollingContext) ->
                     monoError(logger, new RuntimeException("Use the `beginCancelHealthcareJob` to cancel the job")),
-                fetchingOperation(resultId -> Mono.just(getHealthcareFluxPage(resultId,
-                    finalIncludeStatistics == null ? false : finalIncludeStatistics, context)))
+                fetchingOperation(resultId -> Mono.just(getHealthcareFluxPage(resultId, finalTop, finalSkip,
+                    finalIncludeStatistics, context)))
             );
         } catch (RuntimeException ex) {
             return PollerFlux.error(ex);
@@ -109,12 +115,16 @@ PollerFlux<TextAnalyticsOperationResult, PagedIterable<HealthcareTaskResult>> be
         try {
             inputDocumentsValidation(documents);
             String modelVersion = null;
+            Duration pollInterval = DEFAULT_POLL_INTERVAL;
             if (options != null) {
                 modelVersion = options.getModelVersion();
+                pollInterval = options.getPollInterval();
             }
             final Boolean finalIncludeStatistics = options == null ? null : options.isIncludeStatistics();
+            final Integer finalTop = options == null ? null : options.getTop();
+            final Integer finalSkip = options == null ? null : options.getSkip();
             return new PollerFlux<>(
-                DEFAULT_POLL_DURATION,
+                pollInterval,
                 activationOperation(service.healthWithResponseAsync(
                     new MultiLanguageBatchInput().setDocuments(toMultiLanguageInput(documents)),
                     modelVersion,
@@ -132,21 +142,22 @@ PollerFlux<TextAnalyticsOperationResult, PagedIterable<HealthcareTaskResult>> be
                 (activationResponse, pollingContext) ->
                     monoError(logger, new RuntimeException("Use the `beginCancelHealthcareJob` to cancel the job")),
                 fetchingOperationIterable(resultId -> Mono.just(new PagedIterable<>(getHealthcareFluxPage(resultId,
-                    finalIncludeStatistics == null ? false : finalIncludeStatistics, context))))
+                    finalTop, finalSkip, finalIncludeStatistics, context))))
             );
         } catch (RuntimeException ex) {
             return PollerFlux.error(ex);
         }
     }
 
-    PagedFlux<HealthcareTaskResult> getHealthcareFluxPage(UUID jobID, boolean showStats, Context context) {
+    PagedFlux<HealthcareTaskResult> getHealthcareFluxPage(UUID jobID, Integer top, Integer skip, Boolean showStats,
+        Context context) {
         return new PagedFlux<>(
-            () -> getPage(null, jobID, showStats, context),
-            continuationToken -> getPage(continuationToken, jobID, showStats, context));
+            () -> getPage(null, jobID, top, skip, showStats, context),
+            continuationToken -> getPage(continuationToken, jobID, top, skip, showStats, context));
     }
 
-    Mono<PagedResponse<HealthcareTaskResult>> getPage(String continuationToken, UUID jobID,
-        boolean showStats, Context context) {
+    Mono<PagedResponse<HealthcareTaskResult>> getPage(String continuationToken, UUID jobID, Integer top, Integer skip,
+        Boolean showStats, Context context) {
         try {
             if (continuationToken != null) {
                 final Map<String, Integer> continuationTokenMap = parseNextLink(continuationToken);
@@ -156,7 +167,7 @@ Mono<PagedResponse<HealthcareTaskResult>> getPage(String continuationToken, UUID
                     .map(this::toTextAnalyticsPagedResponse)
                     .onErrorMap(Utility::mapToHttpResponseExceptionIfExist);
             } else {
-                return service.healthStatusWithResponseAsync(jobID, null, null, showStats, context)
+                return service.healthStatusWithResponseAsync(jobID, top, skip, showStats, context)
                     .map(this::toTextAnalyticsPagedResponse)
                     .onErrorMap(Utility::mapToHttpResponseExceptionIfExist);
             }
@@ -174,8 +185,8 @@ private PagedResponse<HealthcareTaskResult> toTextAnalyticsPagedResponse(
         final List<TextAnalyticsError> errors = healthcareJobState.getErrors();
 
         final HealthcareTaskResult healthcareTaskResult = new HealthcareTaskResult(
-            // TODO: change back to UUID after service support it.
-            healthcareJobState.getJobId().toString(),
+            // TODO: [Service-Bug] change back to UUID after service support it.
+            healthcareJobState.getJobId(),
             healthcareJobState.getCreatedDateTime(),
             healthcareJobState.getLastUpdateDateTime(),
             toJobState(healthcareJobState.getStatus()),
@@ -196,12 +207,13 @@ private PagedResponse<HealthcareTaskResult> toTextAnalyticsPagedResponse(
             null);
     }
 
-    PollerFlux<TextAnalyticsOperationResult, Void> beginCancelAnalyzeHealthcare(UUID jobId, Context context) {
+    PollerFlux<TextAnalyticsOperationResult, Void> beginCancelAnalyzeHealthcare(String healthTaskId,
+        RecognizeHealthcareEntityOptions options, Context context) {
         try {
-            Objects.requireNonNull(jobId, "'jobId' is required and cannot be null.");
+            Objects.requireNonNull(healthTaskId, "'healthTaskId' is required and cannot be null.");
             return new PollerFlux<>(
-                DEFAULT_POLL_DURATION,
-                activationOperation(service.cancelHealthJobWithResponseAsync(jobId,
+                options == null ? DEFAULT_POLL_INTERVAL : options.getPollInterval(),
+                activationOperation(service.cancelHealthJobWithResponseAsync(UUID.fromString(healthTaskId),
                     context.addData(AZ_TRACING_NAMESPACE_KEY, COGNITIVE_TRACING_NAMESPACE_VALUE))
                     .map(healthResponse -> {
                         final TextAnalyticsOperationResult textAnalyticsOperationResult =
@@ -286,7 +298,7 @@ private Mono<PollResponse<TextAnalyticsOperationResult>> processAnalyzeModelResp
                 logger.info("LongRunningOperation-Cancelling");
                 status = LongRunningOperationStatus.IN_PROGRESS;
                 break;
-            case NOTSTARTED:
+            case NOT_STARTED:
             case RUNNING:
                 status = LongRunningOperationStatus.IN_PROGRESS;
                 break;