EnableStreamingTaskRunner extends BaseStreamingTaskRunner { public String performTask(String taskParameters) { EnableStreamingTaskParameters taskParams = EnableStreamingTaskParameters.deserialize(taskParameters); String spaceId = taskParams.getSpaceId(); boolean secure = taskParams.isSecure(); log.info("Performing " + TASK_NAME + " task on space " + spaceId + ". Secure streaming set to " + secure); String bucketName = unwrappedS3Provider.getBucketName(spaceId); String domainName = null; String distId = null; String oaIdentityId = getOriginAccessId(); EnableStreamingTaskResult taskResult = new EnableStreamingTaskResult(); StreamingDistributionSummary existingDist = getExistingDistribution(bucketName); if (existingDist != null) { boolean existingSecure = !existingDist.getTrustedSigners().getItems().isEmpty(); if ((secure && !existingSecure) || (!secure && existingSecure)) { throw new UnsupportedTaskException(TASK_NAME, "The space " + spaceId + " is already configured to stream as " + (secure ? "OPEN" : "SECURE") + " and cannot be updated to stream as " + (secure ? "SECURE" : "OPEN") + ". To do this, you must first execute the " + StorageTaskConstants.DELETE_STREAMING_TASK_NAME + " task."); } distId = existingDist.getId(); if (!existingDist.isEnabled()) { setDistributionState(distId, true); } domainName = existingDist.getDomainName(); } else { S3Origin origin = new S3Origin(bucketName + S3_ORIGIN_SUFFIX, S3_ORIGIN_OAI_PREFIX + oaIdentityId); TrustedSigners signers = new TrustedSigners(); if (secure) { signers.setItems(Collections.singletonList(cfAccountId)); signers.setEnabled(true); signers.setQuantity(1); } else { signers.setEnabled(false); signers.setQuantity(0); } StreamingDistribution dist = cfClient.createStreamingDistribution( new CreateStreamingDistributionRequest( new StreamingDistributionConfig() .withCallerReference("" + System.currentTimeMillis()) .withS3Origin(origin) .withEnabled(true) .withComment("Streaming space: " + spaceId) .withTrustedSigners(signers))) .getStreamingDistribution(); domainName = dist.getDomainName(); } setBucketAccessPolicy(bucketName, oaIdentityId); Map<String, String> spaceProps = s3Provider.getSpaceProperties(spaceId); spaceProps.put(STREAMING_HOST_PROP, domainName); spaceProps.put(STREAMING_TYPE_PROP, secure ? STREAMING_TYPE.SECURE.name() : STREAMING_TYPE.OPEN.name()); unwrappedS3Provider.setNewSpaceProperties(spaceId, spaceProps); taskResult.setResult("Enable Streaming Task completed successfully"); taskResult.setStreamingHost(domainName); String toReturn = taskResult.serialize(); log.info("Result of " + TASK_NAME + " task: " + toReturn); return toReturn; } EnableStreamingTaskRunner(StorageProvider s3Provider,
                                     S3StorageProvider unwrappedS3Provider,
                                     AmazonS3 s3Client,
                                     AmazonCloudFrontClient cfClient,
                                     String cfAccountId); String getName(); String performTask(String taskParameters);  }
@Test public void testPerformTask2() throws Exception { EnableStreamingTaskRunner runner = createRunner(createMockStorageProviderV2(false), createMockUnwrappedS3StorageProviderV2(), createMockS3ClientV3(), createMockCFClientV2(), cfAccountId); EnableStreamingTaskParameters taskParams = new EnableStreamingTaskParameters(); taskParams.setSpaceId(spaceId); boolean secure = false; taskParams.setSecure(secure); String results = runner.performTask(taskParams.serialize()); assertNotNull(results); testResults(results); testCapturedProps(secure); }