IBMWatsonTranscriptionService extends AbstractJobProducer implements TranscriptionService { public void activate(ComponentContext cc) { if (cc != null) { enabled = OsgiUtil.getOptCfgAsBoolean(cc.getProperties(), ENABLED_CONFIG).get(); if (enabled) { Option<String> urlOpt = OsgiUtil.getOptCfg(cc.getProperties(), IBM_WATSON_SERVICE_URL_CONFIG); if (urlOpt.isSome()) { watsonServiceUrl = UrlSupport.concat(urlOpt.get(), API_VERSION); } Option<String> keyOpt = OsgiUtil.getOptCfg(cc.getProperties(), IBM_WATSON_API_KEY_CONFIG); if (keyOpt.isSome()) { user = APIKEY; psw = keyOpt.get(); logger.info("Using transcription service at {} with api key", watsonServiceUrl); } else { user = OsgiUtil.getComponentContextProperty(cc, IBM_WATSON_USER_CONFIG); psw = OsgiUtil.getComponentContextProperty(cc, IBM_WATSON_PSW_CONFIG); logger.info("Using transcription service at {} with username {}", watsonServiceUrl, user); } Option<String> modelOpt = OsgiUtil.getOptCfg(cc.getProperties(), IBM_WATSON_MODEL_CONFIG); if (modelOpt.isSome()) { model = modelOpt.get(); language = StringUtils.substringBefore(model, "-"); logger.info("Model is {}", model); } else { logger.info("Default model will be used"); } Option<String> wfOpt = OsgiUtil.getOptCfg(cc.getProperties(), WORKFLOW_CONFIG); if (wfOpt.isSome()) workflowDefinitionId = wfOpt.get(); logger.info("Workflow definition is {}", workflowDefinitionId); Option<String> intervalOpt = OsgiUtil.getOptCfg(cc.getProperties(), DISPATCH_WORKFLOW_INTERVAL_CONFIG); if (intervalOpt.isSome()) { try { workflowDispatchInterval = Long.parseLong(intervalOpt.get()); } catch (NumberFormatException e) { } } logger.info("Workflow dispatch interval is {} seconds", workflowDispatchInterval); Option<String> bufferOpt = OsgiUtil.getOptCfg(cc.getProperties(), COMPLETION_CHECK_BUFFER_CONFIG); if (bufferOpt.isSome()) { try { completionCheckBuffer = Long.parseLong(bufferOpt.get()); } catch (NumberFormatException e) { logger.warn("Invalid configuration for {} : {}. Default used instead: {}", COMPLETION_CHECK_BUFFER_CONFIG, bufferOpt.get(), completionCheckBuffer); } } logger.info("Completion check buffer is {} seconds", completionCheckBuffer); Option<String> maxProcessingOpt = OsgiUtil.getOptCfg(cc.getProperties(), MAX_PROCESSING_TIME_CONFIG); if (maxProcessingOpt.isSome()) { try { maxProcessingSeconds = Long.parseLong(maxProcessingOpt.get()); } catch (NumberFormatException e) { } } logger.info("Maximum time a job is checked after it should have ended is {} seconds", maxProcessingSeconds); Option<String> cleaupOpt = OsgiUtil.getOptCfg(cc.getProperties(), CLEANUP_RESULTS_DAYS_CONFIG); if (cleaupOpt.isSome()) { try { cleanupResultDays = Integer.parseInt(cleaupOpt.get()); } catch (NumberFormatException e) { } } logger.info("Cleanup result files after {} days", cleanupResultDays); Option<String> maxAttemptsOpt = OsgiUtil.getOptCfg(cc.getProperties(), MAX_ATTEMPTS_CONFIG); if (maxAttemptsOpt.isSome()) { try { maxAttempts = Integer.parseInt(maxAttemptsOpt.get()); retryWfDefId = OsgiUtil.getComponentContextProperty(cc, RETRY_WORKLFOW_CONFIG); } catch (NumberFormatException e) { logger.warn("Invalid configuration for {} : {}. Default used instead: no retries", MAX_ATTEMPTS_CONFIG, maxAttemptsOpt.get()); } } else { logger.info("No retries in case of errors"); } serverUrl = OsgiUtil.getContextProperty(cc, OpencastConstants.SERVER_URL_PROPERTY); systemAccount = OsgiUtil.getContextProperty(cc, DIGEST_USER_PROPERTY); jobLoad = LoadUtil.getConfiguredLoadValue(cc.getProperties(), START_TRANSCRIPTION_JOB_LOAD_KEY, DEFAULT_START_TRANSCRIPTION_JOB_LOAD, serviceRegistry); scheduledExecutor.scheduleWithFixedDelay(new WorkflowDispatcher(), 120, workflowDispatchInterval, TimeUnit.SECONDS); scheduledExecutor.scheduleWithFixedDelay(new ResultsFileCleanup(), 1, 1, TimeUnit.DAYS); Option<String> optTo = OsgiUtil.getOptCfg(cc.getProperties(), NOTIFICATION_EMAIL_CONFIG); if (optTo.isSome()) toEmailAddress = optTo.get(); else { optTo = OsgiUtil.getOptContextProperty(cc, ADMIN_EMAIL_PROPERTY); if (optTo.isSome()) toEmailAddress = optTo.get(); } if (toEmailAddress != null) logger.info("Notification email set to {}", toEmailAddress); else logger.warn("Email notification disabled"); Option<String> optCluster = OsgiUtil.getOptContextProperty(cc, CLUSTER_NAME_PROPERTY); if (optCluster.isSome()) clusterName = optCluster.get(); logger.info("Environment name is {}", clusterName); logger.info("Activated!"); } else { logger.info("Service disabled. If you want to enable it, please update the service configuration."); } } else throw new IllegalArgumentException("Missing component context"); } IBMWatsonTranscriptionService(); void activate(ComponentContext cc); @Override Job startTranscription(String mpId, Track track); @Override Job startTranscription(String mpId, Track track, String language); @Override void transcriptionDone(String mpId, Object obj); @Override void transcriptionError(String mpId, Object obj); @Override String getLanguage(); @Override MediaPackageElement getGeneratedTranscription(String mpId, String jobId); boolean isCallbackAlreadyRegistered(); void setServiceRegistry(ServiceRegistry serviceRegistry); void setSecurityService(SecurityService securityService); void setUserDirectoryService(UserDirectoryService userDirectoryService); void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectoryService); void setSmtpService(SmtpService service); void setWorkspace(Workspace ws); void setWorkingFileRepository(WorkingFileRepository wfr); void setDatabase(TranscriptionDatabase service); void setAssetManager(AssetManager service); void setWorkflowService(WorkflowService service); boolean isEnabled(); static final String DEFAULT_WF_DEF; static final String ADMIN_URL_PROPERTY; static final float DEFAULT_START_TRANSCRIPTION_JOB_LOAD; static final String START_TRANSCRIPTION_JOB_LOAD_KEY; static final String ENABLED_CONFIG; static final String IBM_WATSON_SERVICE_URL_CONFIG; static final String IBM_WATSON_USER_CONFIG; static final String IBM_WATSON_PSW_CONFIG; static final String IBM_WATSON_API_KEY_CONFIG; static final String IBM_WATSON_MODEL_CONFIG; static final String WORKFLOW_CONFIG; static final String DISPATCH_WORKFLOW_INTERVAL_CONFIG; static final String COMPLETION_CHECK_BUFFER_CONFIG; static final String MAX_PROCESSING_TIME_CONFIG; static final String NOTIFICATION_EMAIL_CONFIG; static final String CLEANUP_RESULTS_DAYS_CONFIG; static final String MAX_ATTEMPTS_CONFIG; static final String RETRY_WORKLFOW_CONFIG; }
@Test public void testWorkflowDispatcherRunTranscriptionCompletedState() throws Exception { service.activate(cc); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), TRACK_DURATION, DATE_EXPECTED, PROVIDER); database.storeJobControl("mpId2", "audioTrack2", "jobId2", TranscriptionJobControl.Status.InProgress.name(), TRACK_DURATION, DATE_EXPECTED, PROVIDER); database.updateJobControl(JOB_ID, TranscriptionJobControl.Status.TranscriptionComplete.name()); Capture<Set<String>> capturedMpIds = mockAssetManagerAndWorkflow(IBMWatsonTranscriptionService.DEFAULT_WF_DEF, true); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(1, capturedMpIds.getValue().size()); Assert.assertEquals(MP_ID, capturedMpIds.getValue().iterator().next()); TranscriptionJobControl job = database.findByJob(JOB_ID); Assert.assertNotNull(job); Assert.assertEquals(TranscriptionJobControl.Status.Closed.name(), job.getStatus()); }
@Test public void testWorkflowDispatcherRunProgressState() throws Exception { service.activate(cc); InputStream stream = IBMWatsonTranscriptionServiceTest.class.getResourceAsStream("/" + PULLED_TRANSCRIPTION_FILE); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), 0, DATE_EXPECTED, PROVIDER); database.storeJobControl("mpId2", "audioTrack2", "jobId2", TranscriptionJobControl.Status.InProgress.name(), TRACK_DURATION, DATE_EXPECTED, PROVIDER); EasyMock.expect(workspace.putInCollection(EasyMock.anyObject(String.class), EasyMock.anyObject(String.class), EasyMock.anyObject(InputStream.class))).andReturn(new URI("http: EasyMock.replay(workspace); HttpEntity httpEntity = EasyMock.createNiceMock(HttpEntity.class); EasyMock.expect(httpEntity.getContent()).andReturn(stream); CloseableHttpResponse response = EasyMock.createNiceMock(CloseableHttpResponse.class); StatusLine status = EasyMock.createNiceMock(StatusLine.class); EasyMock.expect(response.getStatusLine()).andReturn(status).anyTimes(); EasyMock.expect(response.getEntity()).andReturn(httpEntity).anyTimes(); EasyMock.expect(status.getStatusCode()).andReturn(HttpStatus.SC_OK).anyTimes(); EasyMock.replay(httpEntity, response, status); Capture<HttpGet> capturedGet = Capture.newInstance(); EasyMock.expect(httpClient.execute(EasyMock.capture(capturedGet))).andReturn(response); EasyMock.replay(httpClient); Capture<Set<String>> capturedMpIds = mockAssetManagerAndWorkflow(IBMWatsonTranscriptionService.DEFAULT_WF_DEF, true); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(WATSON_URL + "/v1/recognitions/" + JOB_ID, capturedGet.getValue().getURI().toString()); Assert.assertEquals(1, capturedMpIds.getValue().size()); Assert.assertEquals(MP_ID, capturedMpIds.getValue().iterator().next()); TranscriptionJobControl job = database.findByJob(JOB_ID); Assert.assertNotNull(job); Assert.assertEquals(TranscriptionJobControl.Status.Closed.name(), job.getStatus()); }
@Test public void testWorkflowDispatcherFailedState() throws Exception { service.activate(cc); InputStream stream = IBMWatsonTranscriptionServiceTest.class .getResourceAsStream("/" + PULLED_TRANSCRIPTION_ERROR_FILE); HttpEntity httpEntity = EasyMock.createNiceMock(HttpEntity.class); EasyMock.expect(httpEntity.getContent()).andReturn(stream); CloseableHttpResponse response = EasyMock.createNiceMock(CloseableHttpResponse.class); StatusLine status = EasyMock.createNiceMock(StatusLine.class); EasyMock.expect(response.getStatusLine()).andReturn(status).anyTimes(); EasyMock.expect(response.getEntity()).andReturn(httpEntity).anyTimes(); EasyMock.expect(status.getStatusCode()).andReturn(HttpStatus.SC_OK).anyTimes(); EasyMock.replay(httpEntity, response, status); Capture<HttpGet> capturedGet = Capture.newInstance(); EasyMock.expect(httpClient.execute(EasyMock.capture(capturedGet))).andReturn(response).anyTimes(); EasyMock.replay(httpClient); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), 0, DATE_EXPECTED, PROVIDER); EasyMock.replay(workspace); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(WATSON_URL + "/v1/recognitions/" + JOB_ID, capturedGet.getValue().getURI().toString()); TranscriptionJobControl j = database.findByJob(JOB_ID); Assert.assertNotNull(j); Assert.assertEquals(TranscriptionJobControl.Status.Error.toString(), j.getStatus()); EasyMock.verify(smtpService); }
@Test public void testWorkflowDispatcherJobNotFound() throws Exception { service.activate(cc); CloseableHttpResponse response = EasyMock.createNiceMock(CloseableHttpResponse.class); StatusLine status = EasyMock.createNiceMock(StatusLine.class); EasyMock.expect(response.getStatusLine()).andReturn(status).anyTimes(); EasyMock.expect(status.getStatusCode()).andReturn(HttpStatus.SC_NOT_FOUND).anyTimes(); EasyMock.replay(response, status); Capture<HttpGet> capturedGet = Capture.newInstance(); EasyMock.expect(httpClient.execute(EasyMock.capture(capturedGet))).andReturn(response).anyTimes(); EasyMock.replay(httpClient); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), 0, DATE_EXPECTED, PROVIDER); EasyMock.replay(workspace); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(WATSON_URL + "/v1/recognitions/" + JOB_ID, capturedGet.getValue().getURI().toString()); TranscriptionJobControl j = database.findByJob(JOB_ID); Assert.assertNotNull(j); Assert.assertEquals(TranscriptionJobControl.Status.Canceled.toString(), j.getStatus()); EasyMock.verify(smtpService); }
@Test public void testWorkflowDispatcherJobInProgressTooLong() throws Exception { service.activate(cc); InputStream stream = IBMWatsonTranscriptionServiceTest.class.getResourceAsStream("/" + IN_PROGRESS_JOB); HttpEntity httpEntity = EasyMock.createNiceMock(HttpEntity.class); EasyMock.expect(httpEntity.getContent()).andReturn(stream); CloseableHttpResponse response = EasyMock.createNiceMock(CloseableHttpResponse.class); StatusLine status = EasyMock.createNiceMock(StatusLine.class); EasyMock.expect(response.getStatusLine()).andReturn(status).anyTimes(); EasyMock.expect(response.getEntity()).andReturn(httpEntity).anyTimes(); EasyMock.expect(status.getStatusCode()).andReturn(HttpStatus.SC_OK).anyTimes(); EasyMock.replay(httpEntity, response, status); Capture<HttpGet> capturedGet = Capture.newInstance(); EasyMock.expect(httpClient.execute(EasyMock.capture(capturedGet))).andReturn(response).anyTimes(); EasyMock.replay(httpClient); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), 0, DATE_EXPECTED, PROVIDER); EasyMock.replay(workspace); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(WATSON_URL + "/v1/recognitions/" + JOB_ID, capturedGet.getValue().getURI().toString()); TranscriptionJobControl j = database.findByJob(JOB_ID); Assert.assertNotNull(j); Assert.assertEquals(TranscriptionJobControl.Status.Error.toString(), j.getStatus()); EasyMock.verify(smtpService); }
@Test public void testWorkflowDispatcherJobInProgressTooLongWithMaxAttempts() throws Exception { props.put(IBMWatsonTranscriptionService.MAX_ATTEMPTS_CONFIG, 2); props.put(IBMWatsonTranscriptionService.RETRY_WORKLFOW_CONFIG, RETRY_WORKFLOW); service.activate(cc); InputStream stream = IBMWatsonTranscriptionServiceTest.class.getResourceAsStream("/" + IN_PROGRESS_JOB); HttpEntity httpEntity = EasyMock.createNiceMock(HttpEntity.class); EasyMock.expect(httpEntity.getContent()).andReturn(stream); CloseableHttpResponse response = EasyMock.createNiceMock(CloseableHttpResponse.class); StatusLine status = EasyMock.createNiceMock(StatusLine.class); EasyMock.expect(response.getStatusLine()).andReturn(status).anyTimes(); EasyMock.expect(response.getEntity()).andReturn(httpEntity).anyTimes(); EasyMock.expect(status.getStatusCode()).andReturn(HttpStatus.SC_OK).anyTimes(); EasyMock.replay(httpEntity, response, status); Capture<HttpGet> capturedGet = Capture.newInstance(); EasyMock.expect(httpClient.execute(EasyMock.capture(capturedGet))).andReturn(response).anyTimes(); EasyMock.replay(httpClient); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), 0, DATE_EXPECTED, PROVIDER); Capture<Set<String>> capturedMpIds = mockAssetManagerAndWorkflow(RETRY_WORKFLOW, true); EasyMock.replay(workspace); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(WATSON_URL + "/v1/recognitions/" + JOB_ID, capturedGet.getValue().getURI().toString()); TranscriptionJobControl j = database.findByJob(JOB_ID); Assert.assertNotNull(j); Assert.assertEquals(TranscriptionJobControl.Status.Error.toString(), j.getStatus()); }
@Test public void testWorkflowDispatcherRetry() throws Exception { props.put(IBMWatsonTranscriptionService.MAX_ATTEMPTS_CONFIG, 2); props.put(IBMWatsonTranscriptionService.RETRY_WORKLFOW_CONFIG, RETRY_WORKFLOW); service.activate(cc); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.Retry.name(), TRACK_DURATION, DATE_EXPECTED, PROVIDER); Capture<Set<String>> capturedMpIds = mockAssetManagerAndWorkflow(RETRY_WORKFLOW, true); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(1, capturedMpIds.getValue().size()); Assert.assertEquals(MP_ID, capturedMpIds.getValue().iterator().next()); TranscriptionJobControl job = database.findByJob(JOB_ID); Assert.assertNotNull(job); Assert.assertEquals(TranscriptionJobControl.Status.Error.name(), job.getStatus()); }